tuxgyverandClaude Sonnet 5
5b5e90247f
Le serveur gagne un onglet Radio : les stations de l'annuaire, écoutables en direct dans le navigateur ou enregistrables — jusqu'à l'arrêt ou pour une durée choisie —, puis réécoutables depuis l'interface. Quatre enregistrements en parallèle au plus ; si le flux tombe, il est rouvert et le fichier continue. Les enregistrements rejoignent la bibliothèque : le mobile les reçoit. Un redémarrage du service clôt proprement les captures en cours, et celles restées « en cours » après un plantage sont récupérées au démarrage suivant. La copie de flux sans blocs ICY, propre au guet, est extraite pour être partagée. Côté mobile, la synchronisation manuelle n'était qu'un petit lien dans l'onglet Connexion : elle a maintenant un vrai bouton dans l'onglet Synchronisation. Deux passages ne se chevauchent plus, et une erreur inattendue s'affiche au lieu de fermer l'application. Le code Kotlin n'a pas été compilé sur cette machine (pas de JDK) : il a été relu, mais reste à construire. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N6pmSE168yjyf8vgV8mZx6
221 lines
9.6 KiB
Python
221 lines
9.6 KiB
Python
# 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]
|