tuxgyverandClaude Sonnet 5
7810bef8a3
Le balayage ne servait qu'à trouver un morceau guetté, puis oubliait tout. Il note désormais chaque titre annoncé par les stations, même sans guet en cours : un onglet Écoutes montre l'historique, et la rotation — passages et stations distinctes, par titre et par artiste — sur les 100 dernières remontées. Pour que l'historique soit fiable, un tour va maintenant jusqu'au bout après une trouvaille (l'enregistrement part aussitôt, le début du morceau n'est pas perdu), et la liste des stations est mise en cache une demi-heure au lieu d'interroger l'annuaire à chaque tour. Essayé sur de vraies radios : une station alternait le vrai titre et un titre de repli, ce qui doublait les passages. Un titre déjà relevé sur la même station dans les dix minutes est donc ignoré. Première moitié de la reconnaissance des morceaux ; l'écoute continue et l'empreinte sonore suivront. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
244 lines
10 KiB
Python
244 lines
10 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)
|
|
with target.open("wb") as handle:
|
|
written = await self._copy(response, handle, interval, title)
|
|
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
|
|
|
|
async def _copy(self, response, handle, interval: int, wanted: str) -> int:
|
|
"""Retire les blocs ICY du flux : laissés dedans, ils rendraient le fichier illisible."""
|
|
if interval <= 0:
|
|
total = 0
|
|
async for chunk in response.aiter_bytes(32768):
|
|
handle.write(chunk)
|
|
total += len(chunk)
|
|
if total > 40_000_000: # garde-fou : 40 Mo sans annonce de fin
|
|
break
|
|
return total
|
|
|
|
buffer = bytearray()
|
|
total = 0
|
|
async for chunk in response.aiter_bytes(32768):
|
|
buffer.extend(chunk)
|
|
while len(buffer) >= interval + 1:
|
|
length = buffer[interval] * 16
|
|
if len(buffer) < interval + 1 + length:
|
|
break
|
|
handle.write(bytes(buffer[:interval]))
|
|
total += interval
|
|
meta = bytes(buffer[interval + 1:interval + 1 + length]).decode("utf-8", "replace")
|
|
del buffer[:interval + 1 + length]
|
|
announced = radio.STREAM_TITLE.search(meta)
|
|
if announced and announced.group(1).strip() and announced.group(1).strip() != wanted:
|
|
return total # le morceau est terminé
|
|
if total > 60_000_000:
|
|
return total
|
|
return total
|
|
|
|
|
|
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]
|