"""FilyX Music Bridge — Windows-only Now Playing detector. This client reads Windows GSMTC sessions locally. No streaming provider API is called: a bridge token is paired once with the FilyX account then protected by Windows DPAPI in the current user profile. """ from __future__ import annotations import asyncio import base64 import ctypes import hashlib import io import json import os import platform import random import re import sys import threading import time import subprocess from ctypes import wintypes import urllib.error import urllib.parse import urllib.request import uuid import webbrowser from dataclasses import dataclass, asdict from pathlib import Path from typing import Any from PIL import Image from PySide6.QtCore import QObject, Qt, Signal, QSharedMemory from PySide6.QtGui import QAction, QIcon from PySide6.QtWidgets import ( QApplication, QCheckBox, QComboBox, QFormLayout, QFrame, QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMenu, QMessageBox, QPushButton, QSystemTrayIcon, QVBoxLayout, QWidget, ) from websockets.asyncio.client import connect from winsdk.windows.media.control import ( GlobalSystemMediaTransportControlsSessionManager, GlobalSystemMediaTransportControlsSessionPlaybackStatus, ) from winsdk.windows.storage.streams import DataReader import win32crypt import winreg APP_NAME = 'FilyX Desktop Bridge' APP_VERSION = '1.1.6' PROTOCOL_VERSION = 1 FILYX_URL = 'https://filyx.fr' WS_URL = 'wss://filyx.fr/ws/music-bridge' SINGLE_INSTANCE_KEY = 'FilyXMusicBridge.Singleton.v1' DATA_DIR = Path(os.environ.get('LOCALAPPDATA', Path.home())) / 'FilyX Music Bridge' SETTINGS_FILE = DATA_DIR / 'settings.json' TOKEN_FILE = DATA_DIR / 'device-token.dpapi' SNAP_COMMAND_URL = FILYX_URL + '/api/camera-filters/bridge/next' SNAP_ACK_URL = FILYX_URL + '/api/camera-filters/bridge/ack' SNAP_CAMERA_PATHS = ( Path(os.environ.get('ProgramFiles(x86)', r'C:\Program Files (x86)')) / 'Snap Camera' / 'Snap Camera.exe', Path(os.environ.get('ProgramFiles', r'C:\Program Files')) / 'Snap Camera' / 'Snap Camera.exe', Path(r'D:\Program Files (x86)\Snap Camera\Snap Camera.exe'), ) def bundled_asset(name: str) -> Path: """Resolve an asset both from source and the one-file Windows build.""" root = Path(getattr(sys, '_MEIPASS', Path(__file__).resolve().parent.parent)) return root / 'assets' / name def bridge_icon() -> QIcon: return QIcon(str(bundled_asset('filyx-icon.ico'))) @dataclass class Preferences: device_id: str = '' source_preference: str = 'automatic' minimize_to_tray: bool = True start_with_windows: bool = False class ProtectedTokenStore: """DPAPI keeps the bearer token tied to the signed-in Windows account.""" @staticmethod def load() -> str: try: encrypted = TOKEN_FILE.read_bytes() return win32crypt.CryptUnprotectData(encrypted, None, None, None, 0)[1].decode('utf-8') except Exception: return '' @staticmethod def save(token: str) -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) protected = win32crypt.CryptProtectData(token.encode('utf-8'), APP_NAME, None, None, None, 0) TOKEN_FILE.write_bytes(protected) @staticmethod def clear() -> None: try: TOKEN_FILE.unlink() except FileNotFoundError: pass def load_preferences() -> Preferences: try: raw = json.loads(SETTINGS_FILE.read_text(encoding='utf-8')) return Preferences(**{field: raw.get(field, getattr(Preferences(), field)) for field in Preferences.__dataclass_fields__}) except Exception: return Preferences() def save_preferences(preferences: Preferences) -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) SETTINGS_FILE.write_text(json.dumps(asdict(preferences), ensure_ascii=False, indent=2), encoding='utf-8') def milliseconds(value: Any) -> int: if value is None: return 0 if hasattr(value, 'total_seconds'): return max(0, int(value.total_seconds() * 1000)) if hasattr(value, 'duration'): return milliseconds(value.duration) return 0 def is_playing_status(value: Any) -> bool: """Compare the WinRT enum directly; string rendering differs by winsdk version.""" try: if value == GlobalSystemMediaTransportControlsSessionPlaybackStatus.PLAYING: return True except Exception: pass text = str(value or '').lower() return text.endswith('playing') or text == '4' def acquire_single_instance() -> Any | None: """Keep one visible Bridge instance without fragile native mutex calls.""" shared = QSharedMemory(SINGLE_INSTANCE_KEY) if not shared.create(1): return None return shared VLC_WINDOW_SOURCE = '__filyx_vlc_window__' NETFLIX_WINDOW_SOURCE = '__filyx_netflix_window__' BROWSER_PROCESS_NAMES = ('chrome', 'msedge', 'edge', 'brave', 'firefox', 'opera', 'vivaldi') SOURCE_LIST_REFRESH_SECONDS = 15.0 SNAP_COMMAND_POLL_SECONDS = 1.0 SNAP_PROCESS_REFRESH_SECONDS = 12.0 def looks_like_netflix(*values: str) -> bool: """Recognise Netflix from local media/window metadata only. No browser profile, account, page content, audio or video is read. This only lets the overlay show the useful source label for a Netflix tab. """ return any('netflix' in str(value or '').casefold() for value in values) def source_name(app_id: str, *metadata: str) -> str: value = app_id.lower() if 'netflix' in value or looks_like_netflix(*metadata): return 'netflix' if 'spotify' in value: return 'spotify' if 'deezer' in value: return 'deezer' if 'apple' in value or 'itunes' in value: return 'apple music' if 'vlc' in value: return 'vlc' if 'media player' in value or 'zune' in value: return 'windows media player' return app_id.rsplit('!', 1)[0].rsplit('.', 1)[-1] or 'windows media' def is_browser_session(app_id: str) -> bool: value = app_id.casefold() return any(name in value for name in BROWSER_PROCESS_NAMES) def window_process_name(hwnd: Any) -> str: """Return the executable name for a top-level Windows window.""" try: process_id = wintypes.DWORD() ctypes.windll.user32.GetWindowThreadProcessId(hwnd, ctypes.byref(process_id)) handle = ctypes.windll.kernel32.OpenProcess(0x1000, False, process_id.value) # PROCESS_QUERY_LIMITED_INFORMATION if not handle: return '' try: path = ctypes.create_unicode_buffer(32768) length = wintypes.DWORD(len(path)) if not ctypes.windll.kernel32.QueryFullProcessImageNameW(handle, 0, path, ctypes.byref(length)): return '' return Path(path.value).name.casefold() finally: ctypes.windll.kernel32.CloseHandle(handle) except Exception: return '' def window_title(hwnd: Any) -> str: try: user32 = ctypes.windll.user32 length = int(user32.GetWindowTextLengthW(hwnd)) if length <= 0: return '' buffer = ctypes.create_unicode_buffer(length + 1) user32.GetWindowTextW(hwnd, buffer, len(buffer)) return str(buffer.value).strip() except Exception: return '' def visible_window_titles(processes: tuple[str, ...]) -> list[str]: """List visible titles for requested local executables. This is the VLC fallback for versions which do not publish a Windows media session. It does not inspect a media provider or its content. """ wanted = {f'{name}.exe' for name in processes} titles: list[str] = [] try: user32 = ctypes.windll.user32 @ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM) def visit(hwnd: Any, _: Any) -> bool: if len(titles) >= 8: return False if not user32.IsWindowVisible(hwnd): return True title = window_title(hwnd) if title and window_process_name(hwnd) in wanted: titles.append(title) return True user32.EnumWindows(visit, 0) except Exception: return [] return titles def foreground_window_details() -> tuple[str, str]: try: hwnd = ctypes.windll.user32.GetForegroundWindow() if not hwnd: return '', '' return window_process_name(hwnd), window_title(hwnd) except Exception: return '', '' def clean_vlc_title(value: str) -> str: title = re.sub( r'\s*(?:[-—–|]\s*)?(?:\(\s*)?(?:vlc media player|lecteur multimédia vlc|lecteur multimedia vlc)(?:\s*\))?\s*$', '', str(value or ''), flags=re.IGNORECASE, ).strip() return '' if title.casefold() in ('', 'vlc', 'vlc media player', 'lecteur multimédia vlc', 'lecteur multimedia vlc') else title def clean_netflix_title(value: str) -> str: title = re.sub(r'\s*(?:[-—–|])\s*netflix\s*$', '', str(value or ''), flags=re.IGNORECASE).strip() return '' if title.casefold() in ('', 'netflix') else title def snap_camera_installed() -> bool: return any(path.is_file() for path in SNAP_CAMERA_PATHS) def snap_camera_running() -> bool: try: result = subprocess.run(['tasklist', '/FI', 'IMAGENAME eq Snap Camera.exe', '/NH'], capture_output=True, text=True, timeout=2, creationflags=0x08000000) return 'snap camera.exe' in result.stdout.lower() except Exception: return False def snap_virtual_key(key: str) -> int: named = { 'BACKSPACE': 0x08, 'TAB': 0x09, 'ENTER': 0x0D, 'PAUSE': 0x13, 'CAPSLOCK': 0x14, 'ESC': 0x1B, 'SPACE': 0x20, 'PAGEUP': 0x21, 'PAGEDOWN': 0x22, 'END': 0x23, 'HOME': 0x24, 'ARROWLEFT': 0x25, 'ARROWUP': 0x26, 'ARROWRIGHT': 0x27, 'ARROWDOWN': 0x28, 'PRINTSCREEN': 0x2C, 'INSERT': 0x2D, 'DELETE': 0x2E, 'NUMMULTIPLY': 0x6A, 'NUMADD': 0x6B, 'NUMSUBTRACT': 0x6D, 'NUMDECIMAL': 0x6E, 'NUMDIVIDE': 0x6F, 'NUMLOCK': 0x90, 'SCROLLLOCK': 0x91, 'SEMICOLON': 0xBA, 'PLUS': 0xBB, 'COMMA': 0xBC, 'MINUS': 0xBD, 'PERIOD': 0xBE, 'SLASH': 0xBF, 'BACKQUOTE': 0xC0, 'BRACKETLEFT': 0xDB, 'BACKSLASH': 0xDC, 'BRACKETRIGHT': 0xDD, 'QUOTE': 0xDE, } if key in named: return named[key] if key.startswith('NUM') and key[3:].isdigit() and 0 <= int(key[3:]) <= 9: return 0x60 + int(key[3:]) if len(key) == 1 and key.isalnum(): return ord(key) if key.startswith('F') and key[1:].isdigit() and 1 <= int(key[1:]) <= 24: return 0x70 + int(key[1:]) - 1 raise ValueError('Touche Snap Camera non prise en charge.') ULONG_PTR = ctypes.c_ulonglong if ctypes.sizeof(ctypes.c_void_p) == 8 else ctypes.c_ulong class _SnapKeyboardInput(ctypes.Structure): _fields_ = [ ('wVk', wintypes.WORD), ('wScan', wintypes.WORD), ('dwFlags', wintypes.DWORD), ('time', wintypes.DWORD), ('dwExtraInfo', ULONG_PTR), ] class _SnapInputUnion(ctypes.Union): _fields_ = [('ki', _SnapKeyboardInput)] class _SnapInput(ctypes.Structure): _fields_ = [('type', wintypes.DWORD), ('data', _SnapInputUnion)] def send_snap_input(modifiers: list[int], key: int) -> None: """Inject one atomic hardware-like key sequence into the active Snap window.""" events = [ _SnapInput(type=1, data=_SnapInputUnion(ki=_SnapKeyboardInput(code, 0, 0, 0, 0))) for code in [*modifiers, key] ] events.extend( _SnapInput(type=1, data=_SnapInputUnion(ki=_SnapKeyboardInput(code, 0, 0x0002, 0, 0))) for code in [key, *reversed(modifiers)] ) buffer = (_SnapInput * len(events))(*events) sent = ctypes.windll.user32.SendInput(len(events), ctypes.byref(buffer), ctypes.sizeof(_SnapInput)) if sent != len(events): raise ctypes.WinError(ctypes.get_last_error()) def send_snap_hotkey(hotkey: str) -> None: parts = [part for part in str(hotkey).upper().split('+') if part] if not parts: raise ValueError('Raccourci Snap Camera invalide.') key = snap_virtual_key(parts[-1]) modifier_codes = {'CTRL': 0x11, 'ALT': 0x12, 'SHIFT': 0x10, 'WIN': 0x5B} modifiers = [modifier_codes[item] for item in parts[:-1] if item in modifier_codes] if len(modifiers) != len(parts) - 1: raise ValueError('Raccourci Snap Camera invalide.') user32 = ctypes.windll.user32 # Snap Camera accepts its shortcut reliably only when it owns the active # window on some Windows installations. The Bridge briefly activates # Snap, sends the physical key sequence, then restores the streamer's # previous foreground app. This remains fully local: no keyboard input # is sent to any other process. foreground = user32.GetForegroundWindow() snap_window = user32.FindWindowW(None, 'Snap Camera') try: if snap_window: user32.ShowWindow(snap_window, 9) # SW_RESTORE user32.SetForegroundWindow(snap_window) time.sleep(0.08) try: send_snap_input(modifiers, key) except Exception: # Fallback kept for older Windows builds where SendInput can be # restricted by a temporary desktop or accessibility hook. for code in modifiers: user32.keybd_event(code, 0, 0, 0) user32.keybd_event(key, 0, 0, 0) user32.keybd_event(key, 0, 0x0002, 0) finally: try: # The fallback may have pressed these modifiers; releasing an # already released modifier is harmless and prevents a stuck key. for code in reversed(modifiers): user32.keybd_event(code, 0, 0x0002, 0) except Exception: pass if foreground and foreground != snap_window: user32.SetForegroundWindow(foreground) def snap_api(method: str, url: str, token: str, device_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: query = urllib.parse.urlencode({'device_id': device_id}) request_url = f'{url}?{query}' data = json.dumps(payload).encode('utf-8') if payload is not None else None request = urllib.request.Request(request_url, data=data, method=method, headers={ 'Accept': 'application/json', 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json', 'X-FilyX-Device-Id': device_id, }) with urllib.request.urlopen(request, timeout=3) as response: parsed = json.loads(response.read().decode('utf-8')) if not isinstance(parsed, dict) or not parsed.get('ok'): raise RuntimeError(str(parsed.get('error') if isinstance(parsed, dict) else 'Réponse Snap FilyX invalide.')) return parsed SOURCE_PRIORITY = {'spotify': 0, 'deezer': 1, 'apple music': 2, 'netflix': 3, 'vlc': 4, 'windows media player': 5} class WindowsMediaDetector: def __init__(self, preference: str = 'automatic') -> None: self.preference = preference self.manager: Any | None = None # Reading a Windows thumbnail opens an async stream, decodes it and # recompresses it for the wire. It must happen only when the track # changes, never on each progress tick. self._artwork_key = '' self._cached_artwork: dict[str, str] | None = None async def ensure(self) -> None: if self.manager is None: self.manager = await GlobalSystemMediaTransportControlsSessionManager.request_async() async def available_sources(self) -> list[tuple[str, str]]: await self.ensure() items: list[tuple[str, str]] = [] for session in self.manager.get_sessions(): app_id = str(session.source_app_user_model_id or '') if not app_id: continue try: props = await session.try_get_media_properties_async() label = source_name(app_id, str(props.title or ''), str(props.artist or ''), str(props.album_title or '')) except Exception: label = source_name(app_id) items.append((app_id, label)) if self._vlc_window_track() is not None: items.append((VLC_WINDOW_SOURCE, 'VLC (titre local)')) if self._netflix_window_track() is not None: items.append((NETFLIX_WINDOW_SOURCE, 'Netflix (onglet actif)')) return items async def snapshot(self) -> dict[str, Any] | None: await self.ensure() if self.preference == VLC_WINDOW_SOURCE: return self._vlc_window_track() if self.preference == NETFLIX_WINDOW_SOURCE: return self._netflix_window_track() candidates: list[tuple[tuple[int, int, int], Any, Any, Any]] = [] sessions = list(self.manager.get_sessions()) current = self.manager.get_current_session() selected: tuple[Any, Any, Any] | None = None # Windows already designates the active media session. Prefer it before # sorting generic browser sessions: several Brave tabs may expose the # same app id and otherwise make the timeline jump between tabs. if self.preference == 'automatic' and current is not None: try: current_props = await current.try_get_media_properties_async() current_playback = current.get_playback_info() current_app_id = str(current.source_app_user_model_id or '') current_source = self._session_source(current_app_id, current_props) current_title = str(current_props.title or '').strip() if current_source == 'vlc': current_title = clean_vlc_title(current_title) if current_title: selected = (current, current_props, current_playback) except Exception: selected = None if selected is None: for index, session in enumerate(sessions): try: props = await session.try_get_media_properties_async() playback = session.get_playback_info() app_id = str(session.source_app_user_model_id or '') source = self._session_source(app_id, props) title = str(props.title or '').strip() if source == 'vlc': title = clean_vlc_title(title) if not title: continue playing = is_playing_status(getattr(playback, 'playback_status', None)) forced = self.preference != 'automatic' and self.preference == app_id priority = SOURCE_PRIORITY.get(source, 30) rank = (0 if forced else 1, 0 if playing else 1, priority, index) candidates.append((rank, session, props, playback)) except Exception: continue if not candidates: return self._fallback_track() _, session, props, playback = min(candidates, key=lambda item: item[0]) else: session, props, playback = selected timeline = session.get_timeline_properties() app_id = str(session.source_app_user_model_id or '') source = self._session_source(app_id, props) title = str(props.title or '').strip() if source == 'vlc': title = clean_vlc_title(title) if not title: # Do not publish VLC's generic application caption as a track. # A real VLC window title is tried by the fallback before the # overlay is told that no useful media title is available. return self._fallback_track() duration = milliseconds(getattr(timeline, 'end_time', None)) position = milliseconds(getattr(timeline, 'position', None)) playing = is_playing_status(getattr(playback, 'playback_status', None)) artist = str(props.artist or '').strip() album = str(props.album_title or '').strip() artwork_key = '\x1f'.join((app_id, title, artist, album, str(duration))) if artwork_key != self._artwork_key: self._artwork_key = artwork_key self._cached_artwork = await self._artwork(getattr(props, 'thumbnail', None)) artwork = self._cached_artwork updated_at = int(time.time() * 1000) # Do not keep a local browser-specific "playing" grace period here. # It made real pause/resume events look like a one-second blink for # some players. The server relay now debounces all clients uniformly # from their measured media timeline instead. # Position is kept raw here. The OBS overlay owns the smooth local # clock between Bridge reports; adding elapsed time in both places # would make a player appear to run twice as fast. if duration and position > duration: position = duration return { 'source': source, 'source_app_id': app_id, 'title': title, 'artist': artist, 'album': album, 'album_artist': str(props.album_artist or '').strip(), 'duration_ms': duration, 'position_ms': position, 'is_playing': playing, 'artwork': artwork, 'updated_at': updated_at, } def _session_source(self, app_id: str, props: Any) -> str: title = str(props.title or '').strip() artist = str(props.artist or '').strip() album = str(props.album_title or '').strip() source = source_name(app_id, title, artist, album) # Browsers can expose the episode title without its provider. If the # foreground tab itself is Netflix and it belongs to this same local # browser process, name this media session Netflix. Background tabs, # account data, audio and video are never inspected. process, active_title = foreground_window_details() if source != 'netflix' and is_browser_session(app_id) and looks_like_netflix(active_title): if process and process.removesuffix('.exe') in app_id.casefold(): return 'netflix' return source def _fallback_track(self) -> dict[str, Any] | None: if self.preference in ('automatic', VLC_WINDOW_SOURCE): track = self._vlc_window_track() if track is not None: return track if self.preference in ('automatic', NETFLIX_WINDOW_SOURCE): return self._netflix_window_track() return None def _vlc_window_track(self) -> dict[str, Any] | None: for raw_title in visible_window_titles(('vlc',)): title = clean_vlc_title(raw_title) if title: return { 'source': 'vlc', 'source_app_id': 'vlc.exe', 'title': title, 'artist': '', 'album': '', 'album_artist': '', 'duration_ms': 0, 'position_ms': 0, 'is_playing': True, 'artwork': None, 'updated_at': int(time.time() * 1000), 'playback_state_known': False, } return None def _netflix_window_track(self) -> dict[str, Any] | None: process, raw_title = foreground_window_details() if process.removesuffix('.exe') not in BROWSER_PROCESS_NAMES or not looks_like_netflix(raw_title): return None title = clean_netflix_title(raw_title) if not title: return None return { 'source': 'netflix', 'source_app_id': process, 'title': title, 'artist': 'Netflix', 'album': '', 'album_artist': '', 'duration_ms': 0, 'position_ms': 0, 'is_playing': True, 'artwork': None, 'updated_at': int(time.time() * 1000), 'playback_state_known': False, } async def _artwork(self, thumbnail: Any) -> dict[str, str] | None: if thumbnail is None: return None try: stream = await thumbnail.open_read_async() size = int(stream.size) if size <= 0 or size > 8 * 1024 * 1024: return None reader = DataReader(stream) await reader.load_async(size) data = bytearray(size) reader.read_bytes(data) image = Image.open(io.BytesIO(bytes(data))).convert('RGB') image.thumbnail((512, 512), Image.Resampling.LANCZOS) for quality in (88, 82, 76, 70): output = io.BytesIO(); image.save(output, format='JPEG', quality=quality, optimize=True) payload = output.getvalue() if len(payload) <= 512 * 1024: return {'mime': 'image/jpeg', 'data': base64.b64encode(payload).decode('ascii')} except Exception: return None return None class RuntimeSignals(QObject): status = Signal(dict) paired = Signal(str) sources = Signal(list) snap_status = Signal(dict) class BridgeRuntime: def __init__(self, preferences: Preferences, signals: RuntimeSignals) -> None: self.preferences = preferences self.signals = signals self.token = ProtectedTokenStore.load() self.stop_event = threading.Event() self.loop: asyncio.AbstractEventLoop | None = None self.detector: WindowsMediaDetector | None = None self.last_signature = '' self.last_progress_sent = 0.0 self.last_snap_poll = 0.0 self.last_snap_notice = '' self.last_source_refresh = 0.0 self.last_source_signature = '' self.last_snap_process_check = 0.0 self.snap_running_cached = False def start(self) -> None: threading.Thread(target=self._thread_main, daemon=True, name='FilyXMusicBridge').start() def _thread_main(self) -> None: asyncio.run(self._run()) async def _run(self) -> None: self.loop = asyncio.get_running_loop() self.detector = WindowsMediaDetector(self.preferences.source_preference) # The Windows media session service can briefly be unavailable just # after login or while a browser starts. Do not let that transient # state terminate the whole Bridge before it has connected to FilyX. try: await self._announce_sources(force=True) except Exception as error: self.signals.status.emit({'connected': False, 'message': f'En attente du lecteur Windows · {str(error)[:70]}'}) snap_task = asyncio.create_task(self._snap_loop()) try: delay = 1.0 while not self.stop_event.is_set(): if not self.token: self.signals.status.emit({'connected': False, 'message': 'Non associé à FilyX'}) await asyncio.sleep(1) continue try: async with connect(WS_URL, ping_interval=25, ping_timeout=20, max_size=900 * 1024) as ws: await ws.send(json.dumps({'type': 'hello', 'protocol': PROTOCOL_VERSION, 'device_id': self.preferences.device_id, 'token': self.token, 'bridge_version': APP_VERSION, 'os': platform.platform()})) reply = json.loads(await asyncio.wait_for(ws.recv(), 12)) while reply.get('type') == 'ready': reply = json.loads(await asyncio.wait_for(ws.recv(), 12)) if reply.get('type') != 'authenticated': raise RuntimeError(reply.get('error', 'Association refusée')) delay = 1.0; self.signals.status.emit({'connected': True, 'message': 'Connecté à FilyX'}) await self._stream(ws) except Exception as error: self.signals.status.emit({'connected': False, 'message': f'Reconnexion dans {int(delay)} s · {str(error)[:100]}'}) await asyncio.sleep(delay + random.uniform(0, .35)); delay = min(30.0, 2.0 if delay < 2 else delay * 1.7) finally: snap_task.cancel() try: await snap_task except asyncio.CancelledError: pass async def _stream(self, ws: Any) -> None: no_track_sent = False while not self.stop_event.is_set() and ws.close_code is None: assert self.detector is not None self.detector.preference = self.preferences.source_preference try: track = await self.detector.snapshot() await self._announce_sources() except Exception as error: # Keep the authenticated socket alive and retry locally. A # browser media session may appear a few seconds after the # desktop app itself is already running. self.signals.status.emit({'connected': True, 'message': f'Connecté · attente du lecteur Windows ({str(error)[:55]})'}) if not no_track_sent: await ws.send(json.dumps({'type': 'offline', 'protocol': PROTOCOL_VERSION})) no_track_sent = True await asyncio.sleep(2) continue if track is None: if not no_track_sent: await ws.send(json.dumps({'type': 'offline', 'protocol': PROTOCOL_VERSION})); no_track_sent = True self.signals.status.emit({'connected': True, 'message': 'Connecté · aucune musique détectée'}) await asyncio.sleep(1); continue no_track_sent = False artwork = track.pop('artwork', None) signature = hashlib.sha256(json.dumps({key: value for key, value in track.items() if key not in ('position_ms', 'updated_at')}, sort_keys=True).encode()).hexdigest() payload = dict(track) if signature != self.last_signature: if artwork: payload['artwork'] = artwork await ws.send(json.dumps({'type': 'music_update', 'protocol': PROTOCOL_VERSION, 'payload': payload})) self.last_signature = signature elif time.monotonic() - self.last_progress_sent >= 2: await ws.send(json.dumps({'type': 'playback', 'protocol': PROTOCOL_VERSION, 'payload': {'is_playing': track['is_playing'], 'position_ms': track['position_ms'], 'timestamp': track['updated_at']}})) self.last_progress_sent = time.monotonic() self.signals.status.emit({'connected': True, 'message': f"{track['source']} · {track['title']}", 'track': track}) await asyncio.sleep(1) async def _snap_loop(self) -> None: while not self.stop_event.is_set(): await self._poll_snap_camera() # One request per second is imperceptible for a chat-triggered # lens, while avoiding the permanent 4 Hz wake-up used before. await asyncio.sleep(SNAP_COMMAND_POLL_SECONDS) def _snap_camera_running(self, force: bool = False) -> bool: now = time.monotonic() if force or now - self.last_snap_process_check >= SNAP_PROCESS_REFRESH_SECONDS: self.last_snap_process_check = now self.snap_running_cached = snap_camera_running() return self.snap_running_cached def _snap_status(self, message: str, connected: bool = False) -> None: # Avoid repainting the desktop UI every 750 ms when no command is # pending. A change of state remains immediately visible. if message == self.last_snap_notice: return self.last_snap_notice = message self.signals.snap_status.emit({'connected': connected, 'message': message}) async def _poll_snap_camera(self) -> None: if not self.token or not self.preferences.device_id: return now = time.monotonic() if now - self.last_snap_poll < SNAP_COMMAND_POLL_SECONDS: return self.last_snap_poll = now try: reply = await asyncio.to_thread(snap_api, 'GET', SNAP_COMMAND_URL, self.token, self.preferences.device_id) command = reply.get('command') if not command: self._snap_status('Snap Camera prêt · aucune commande en attente.', self._snap_camera_running()) return if not snap_camera_installed(): self._snap_status('Snap Camera n’est pas installé sur ce PC.') return if not self._snap_camera_running(force=True): self._snap_status('Snap Camera est fermé : ouvre-le pour recevoir les commandes.') return hotkey = str(command.get('hotkey') or '') send_snap_hotkey(hotkey) command_type = str(command.get('command_type') or 'apply') await asyncio.to_thread(snap_api, 'POST', SNAP_ACK_URL, self.token, self.preferences.device_id, { 'queue_id': int(command.get('queue_id') or 0), 'command_type': command_type, }) if command_type == 'restore': self._snap_status('Lens Snap par défaut restaurée.', True) else: self._snap_status(f"Lens Snap appliquée : {command.get('filter_name') or command.get('filter_key')}", True) except urllib.error.HTTPError as error: self._snap_status(f'Snap Camera · connexion FilyX refusée ({error.code}).') except Exception as error: self._snap_status(f'Snap Camera · {str(error)[:100]}') async def _announce_sources(self, force: bool = False) -> None: if self.detector is None: return now = time.monotonic() if not force and now - self.last_source_refresh < SOURCE_LIST_REFRESH_SECONDS: return self.last_source_refresh = now try: sources = await self.detector.available_sources() signature = json.dumps(sources, ensure_ascii=False, sort_keys=True) if force or signature != self.last_source_signature: self.last_source_signature = signature self.signals.sources.emit(sources) except Exception: pass def pair(self, code: str) -> None: if self.loop is not None: asyncio.run_coroutine_threadsafe(self._pair(code), self.loop) async def _pair(self, code: str) -> None: device_id = self.preferences.device_id or str(uuid.uuid4()) async with connect(WS_URL, ping_interval=25, ping_timeout=20, max_size=900 * 1024) as ws: await ws.send(json.dumps({'type': 'pair', 'protocol': PROTOCOL_VERSION, 'code': code, 'device_id': device_id, 'device_name': platform.node()[:120] or 'PC Windows', 'bridge_version': APP_VERSION, 'protocol_version': PROTOCOL_VERSION})) reply = json.loads(await asyncio.wait_for(ws.recv(), 12)) while reply.get('type') == 'ready': reply = json.loads(await asyncio.wait_for(ws.recv(), 12)) if reply.get('type') != 'paired': raise RuntimeError(reply.get('error', 'Association refusée')) token = str(reply.get('token') or '') if len(token) < 43: raise RuntimeError('Jeton FilyX incomplet') self.preferences.device_id = device_id; self.token = token; ProtectedTokenStore.save(token); save_preferences(self.preferences) self.signals.paired.emit('Appareil associé à FilyX avec succès.') def disconnect(self) -> None: self.token = ''; ProtectedTokenStore.clear(); self.signals.status.emit({'connected': False, 'message': 'Appareil déconnecté'}) class MainWindow(QMainWindow): def __init__(self) -> None: super().__init__(); self.preferences = load_preferences(); self.signals = RuntimeSignals(); self.runtime = BridgeRuntime(self.preferences, self.signals) self.setWindowTitle(APP_NAME); self.setWindowIcon(bridge_icon()); self.setMinimumWidth(520); self._build(); self._tray(); self.signals.status.connect(self._status); self.signals.snap_status.connect(self._snap_status); self.signals.paired.connect(self._paired); self.signals.sources.connect(self._sources); self.runtime.start() def _build(self) -> None: page = QWidget(); layout = QVBoxLayout(page); layout.setContentsMargins(26, 24, 26, 24); layout.setSpacing(14) title = QLabel('FILYX DESKTOP BRIDGE'); title.setObjectName('title'); subtitle = QLabel('Musique en cours · Snap Camera local · aucune vidéo envoyée'); subtitle.setObjectName('subtitle'); layout.addWidget(title); layout.addWidget(subtitle) self.status = QLabel('● Non associé'); self.status.setObjectName('status'); layout.addWidget(self.status) card = QFrame(); card.setObjectName('card'); card_layout = QVBoxLayout(card); self.track = QLabel('Aucune musique détectée'); self.track.setWordWrap(True); card_layout.addWidget(self.track); layout.addWidget(card) snap_card = QFrame(); snap_card.setObjectName('card'); snap_layout = QVBoxLayout(snap_card); snap_title = QLabel('SNAP CAMERA CONTROL'); snap_title.setObjectName('subtitle'); self.snap_status = QLabel('○ En attente de Snap Camera'); self.snap_status.setWordWrap(True); snap_layout.addWidget(snap_title); snap_layout.addWidget(self.snap_status); layout.addWidget(snap_card) form = QFormLayout(); self.code = QLineEdit(); self.code.setPlaceholderText('ABCD-EFGH'); self.code.setMaxLength(9); self.source = QComboBox(); self.source.addItem('Automatique', 'automatic'); self.source.currentIndexChanged.connect(self._source_changed); form.addRow('Code FilyX', self.code); form.addRow('Source musicale', self.source); layout.addLayout(form) actions = QHBoxLayout(); pair = QPushButton('Associer FilyX'); pair.clicked.connect(self._pair); disconnect = QPushButton('Déconnecter'); disconnect.clicked.connect(self.runtime.disconnect); open_panel = QPushButton('Ouvrir FilyX'); open_panel.clicked.connect(lambda: webbrowser.open(FILYX_URL + '/panel/camerafilters')); actions.addWidget(pair); actions.addWidget(disconnect); actions.addWidget(open_panel); layout.addLayout(actions) self.tray_box = QCheckBox('Réduire dans la zone de notification'); self.tray_box.setChecked(self.preferences.minimize_to_tray); self.tray_box.toggled.connect(self._save_flags); self.startup_box = QCheckBox('Lancer au démarrage de Windows'); self.startup_box.setChecked(self.preferences.start_with_windows); self.startup_box.toggled.connect(self._save_flags); layout.addWidget(self.tray_box); layout.addWidget(self.startup_box); self.setCentralWidget(page) self.setStyleSheet('QWidget{background:#101014;color:#f7f4f1;font-family:Segoe UI;}#title{color:#f27622;font-size:22px;font-weight:800;letter-spacing:2px;}#subtitle{color:#a9a6ae;}#status{padding:11px;border-radius:9px;background:#1b1b21;color:#f27622;font-weight:700;}#card{background:#19191f;border:1px solid #38343a;border-radius:12px;padding:14px;}QLineEdit,QComboBox{padding:9px;border:1px solid #423b39;border-radius:7px;background:#18181d;}QPushButton{padding:9px 12px;border:0;border-radius:7px;background:#f27622;color:#1b1008;font-weight:800;}QCheckBox{color:#d7d3da;}') def _tray(self) -> None: self.tray = QSystemTrayIcon(bridge_icon(), self); self.tray.setToolTip(APP_NAME); menu = QMenu(self); open_action = QAction('Ouvrir FilyX Desktop Bridge', self); open_action.triggered.connect(self.showNormal); reconnect = QAction('Reconnecter', self); reconnect.triggered.connect(lambda: self.runtime.signals.status.emit({'connected': False, 'message': 'Reconnexion demandée'})); panel = QAction('Ouvrir le dashboard FilyX', self); panel.triggered.connect(lambda: webbrowser.open(FILYX_URL + '/panel/camerafilters')); quit_action = QAction('Quitter', self); quit_action.triggered.connect(QApplication.quit); menu.addAction(open_action); menu.addAction(reconnect); menu.addAction(panel); menu.addSeparator(); menu.addAction(quit_action); self.tray.setContextMenu(menu); self.tray.show() def _pair(self) -> None: code = self.code.text().strip() if len(code.replace('-', '')) != 8: QMessageBox.warning(self, APP_NAME, 'Entre le code de 8 caractères généré dans ton panel FilyX.'); return self.runtime.pair(code) def _status(self, data: dict) -> None: connected = bool(data.get('connected')); self.status.setText(('● ' if connected else '○ ') + str(data.get('message', 'En attente'))); self.status.setStyleSheet('color:#4fd68a;' if connected else 'color:#f27622;'); track = data.get('track'); if track: self.track.setText(f"{track.get('artist') or 'Artiste inconnu'}\n{track.get('title') or 'Musique en cours'}\n{track.get('album') or ''}") def _snap_status(self, data: dict) -> None: connected = bool(data.get('connected')); self.snap_status.setText(('● ' if connected else '○ ') + str(data.get('message', 'En attente'))); self.snap_status.setStyleSheet('color:#4fd68a;' if connected else 'color:#f27622;') def _paired(self, message: str) -> None: self.code.clear(); QMessageBox.information(self, APP_NAME, message) def _sources(self, sources: list) -> None: current = self.source.currentData(); self.source.blockSignals(True); self.source.clear(); self.source.addItem('Automatique', 'automatic'); seen = set() for app_id, label in sources: if app_id in seen: continue seen.add(app_id); self.source.addItem(label, app_id) index = self.source.findData(self.preferences.source_preference); self.source.setCurrentIndex(index if index >= 0 else 0); self.source.blockSignals(False) def _source_changed(self) -> None: self.preferences.source_preference = str(self.source.currentData() or 'automatic'); save_preferences(self.preferences) def _save_flags(self) -> None: self.preferences.minimize_to_tray = self.tray_box.isChecked(); self.preferences.start_with_windows = self.startup_box.isChecked(); save_preferences(self.preferences); self._set_startup(self.preferences.start_with_windows) def _set_startup(self, enabled: bool) -> None: command = f'"{sys.executable if getattr(sys, "frozen", False) else Path(__file__).resolve()}"' with winreg.OpenKey(winreg.HKEY_CURRENT_USER, r'Software\Microsoft\Windows\CurrentVersion\Run', 0, winreg.KEY_SET_VALUE) as key: if enabled: winreg.SetValueEx(key, 'FilyXMusicBridge', 0, winreg.REG_SZ, command) else: try: winreg.DeleteValue(key, 'FilyXMusicBridge') except FileNotFoundError: pass def closeEvent(self, event) -> None: if self.preferences.minimize_to_tray: self.hide(); event.ignore() else: event.accept() def main() -> int: if sys.platform != 'win32': raise SystemExit('FilyX Music Bridge est réservé à Windows 10 1809+ et Windows 11.') instance = acquire_single_instance() if instance is None: ctypes.windll.user32.MessageBoxW(None, 'FilyX Music Bridge est déjà ouvert.', APP_NAME, 0x40) return 0 try: app = QApplication(sys.argv); app.setApplicationName(APP_NAME); app.setWindowIcon(bridge_icon()); app.setQuitOnLastWindowClosed(False); window = MainWindow() window.show(); return app.exec() finally: instance.detach() if __name__ == '__main__': raise SystemExit(main())