# Noxely — collecte côté serveur, écoute hors connexion côté mobile. # Copyright © 2026 Johnny (tuxgyver) # # Ce programme est un logiciel libre : vous pouvez le redistribuer et le modifier # selon les termes de la GNU General Public License, version 3 ou ultérieure, # telle que publiée par la Free Software Foundation. # # Il est distribué dans l'espoir d'être utile, mais SANS AUCUNE GARANTIE. # Voir le fichier LICENSE à la racine du dépôt pour le texte complet. """Tâches de fond : file de téléchargement, guet radio et enregistrement. C'est ici que se déplace la consommation qui pesait sur le téléphone : le serveur balaie les stations et télécharge, le mobile ne fait plus que récupérer des fichiers prêts. """ from __future__ import annotations import asyncio import re import time from pathlib import Path import httpx from . import rotation from .config import Config from .sources import radio from .storage import Storage UA = {"User-Agent": "Noxely/1.0 (https://git.tuxtech.fr/tuxgyver/Noxely)"} def safe_name(text: str, extension: str) -> str: cleaned = re.sub(r"[^\w \-.]", "", text).strip().replace(" ", "_")[:80] or "media" return f"{cleaned}.{extension}" class Downloader: """Vide la file des médias en attente, un à la fois.""" def __init__(self, config: Config, storage: Storage) -> None: self.config = config self.storage = storage self.config.media_dir.mkdir(parents=True, exist_ok=True) async def run_forever(self) -> None: while True: try: await self.drain() except Exception as error: # une panne ne doit pas arrêter la boucle print(f"[noxely] téléchargement : {error}") await asyncio.sleep(10) async def drain(self) -> None: for item in self.storage.pending_downloads(): await self.fetch(item.id, item.remote_url, item.filename) async def fetch(self, media_id: str, url: str, filename: str) -> None: target = self.config.media_dir / filename partial = target.with_suffix(target.suffix + ".part") self.storage.update_media(media_id, state="downloading", error=None) try: async with httpx.AsyncClient(headers=UA, timeout=httpx.Timeout(120.0), follow_redirects=True) as client: async with client.stream("GET", url) as response: response.raise_for_status() with partial.open("wb") as handle: async for chunk in response.aiter_bytes(65536): handle.write(chunk) partial.replace(target) self.storage.update_media(media_id, state="ready", size=target.stat().st_size) except Exception as error: partial.unlink(missing_ok=True) self.storage.update_media(media_id, state="failed", error=str(error)[:200]) class Watcher: """Balaie les stations à la recherche des morceaux guettés, et les enregistre.""" def __init__(self, config: Config, storage: Storage) -> None: self.config = config self.storage = storage self.recording: str | None = None #: Dernier tour de balayage, pour que l'interface montre que ça tourne. self.last_sweep: dict = {} self.sweeps = 0 self._stations: list[dict] = [] self._stations_at = 0.0 self._pruned_at = 0.0 async def run_forever(self) -> None: """Balaie en continu : un tour enchaîne sur le suivant. Une sonde ne lit que le premier bloc de métadonnées puis referme — quelques dizaines de kilo-octets par station. C'est ce qui rend le balayage permanent tenable, là où garder quarante flux ouverts consommerait plusieurs mégabits par seconde en continu. Le rythme dépend de ce qu'on attend de lui : serré quand un morceau est guetté, plus lâche quand il ne fait que tenir l'historique, arrêté si l'historique est coupé. """ while True: try: guetting = bool(self.storage.pending_watches()) if not guetting and not self.config.history: # Rien à guetter et rien à retenir : inutile de solliciter l'annuaire. await asyncio.sleep(30) continue self.last_sweep = await self.sweep() self.sweeps += 1 self._prune() except Exception as error: print(f"[noxely] guet : {error}") await asyncio.sleep(30) continue await asyncio.sleep(self.config.watch_interval if guetting else self.config.listen_interval) async def _known_stations(self) -> list[dict]: """Les stations à sonder, relues de l'annuaire une fois par demi-heure. Sans ce cache, chaque tour — toutes les quelques secondes en cas de guet — interrogerait l'annuaire, un service communautaire. Si celui-ci est injoignable, on garde la liste précédente plutôt que de renoncer au tour. """ fresh = time.monotonic() - self._stations_at < 1800 if self._stations and fresh: return self._stations try: self._stations = await radio.stations(limit=self.config.watch_stations) self._stations_at = time.monotonic() except Exception: if not self._stations: raise return self._stations def _prune(self) -> None: """Purge l'historique périmé, au plus une fois par heure.""" if time.monotonic() - self._pruned_at < 3600: return self._pruned_at = time.monotonic() limit = _now() - self.config.history_days * 86_400_000 self.storage.prune_heard(limit) async def sweep(self) -> dict: pending = self.storage.pending_watches() found = await self._known_stations() titles = 0 noted = 0 matched: dict | None = None for group in _chunks(found, self.config.watch_concurrency): heard = await asyncio.gather(*(radio.now_playing(s["url"]) for s in group)) for station, title in zip(group, heard): if not title: continue titles += 1 if self.config.history and rotation.is_track(title, station["name"]): artist, name = rotation.split_title(title) noted += self.storage.add_heard( station_id=station["id"], station=station["name"], raw=title, artist=artist, title=name, ) if matched: continue hit = next((w for w in pending if w["query"].lower() in title.lower()), None) if not hit: continue # L'enregistrement part tout de suite : attendre la fin du tour ferait manquer # le début du morceau. Le tour, lui, continue, pour que l'historique soit complet. if self.recording is None: asyncio.create_task(self.record(station, title, hit["query"])) self.storage.mark_watch_found(hit["id"], station["name"], title) matched = {"matched": title, "station": station["name"]} self.storage.mark_watches_scanned([w["id"] for w in pending]) return {"stations": len(found), "titles": titles, "heard": noted, "at": _now(), "matched": matched["matched"] if matched else None, **({"station": matched["station"]} if matched else {})} async def record(self, station: dict, title: str, trigger: str) -> None: """Enregistre jusqu'à ce que la station annonce un autre morceau.""" self.recording = station["name"] extension = "aac" if station["codec"].upper().startswith("AAC") else "mp3" filename = safe_name(title, extension) media_id = self.storage.add_media( title=title, artist=station["name"], source="radio", origin=station["name"], filename=filename, remote_url=None, mime="audio/aac" if extension == "aac" else "audio/mpeg", state="downloading", ) target = self.config.media_dir / filename headers = {**UA, "Icy-MetaData": "1"} written = 0 try: async with httpx.AsyncClient(headers=headers, timeout=httpx.Timeout(60.0)) as client: async with client.stream("GET", station["url"]) as response: response.raise_for_status() interval = int(response.headers.get("icy-metaint") or 0) # Fin du morceau : la station en annonce un autre. Les gardes-fous bornent # le fichier quand aucune annonce ne vient (40 Mo sans ICY, 60 Mo avec). limit = 40_000_000 if interval <= 0 else 60_000_000 def keep_going(total: int, announced: str | None) -> bool: return not (announced and announced != title) and total <= limit with target.open("wb") as handle: written = await radio.copy_stream(response, handle, interval, keep_going) self.storage.update_media(media_id, state="ready", size=written) except Exception as error: self.storage.update_media(media_id, state="failed", error=str(error)[:200]) finally: self.recording = None def _now() -> int: return int(time.time() * 1000) def _chunks(items: list, size: int): for index in range(0, len(items), size): yield items[index:index + size]