tuxgyver
ed1d40826b
Le téléphone balayait quarante radios toutes les trois minutes, cherchait dans quatre catalogues et enregistrait des directs. Tout cela le tenait éveillé pour un travail qui n'a aucune raison de se faire sur batterie. Un serveur reprend la collecte : recherche, guet, enregistrement, file de téléchargement, piloté depuis un navigateur. Le mobile demande ce qui a changé depuis son curseur, télécharge et se rendort — toutes les six heures, en Wi-Fi non facturé. Vingt fichiers quittent l'application au passage. Le balayage devient continu plutôt qu'espacé de trois minutes : une sonde ne lit que le premier bloc de métadonnées avant de refermer, ce qui rend la permanence tenable quand garder les flux ouverts coûterait plusieurs mégabits par seconde. Un morceau guetté a été capté en moins de vingt-cinq secondes. La recherche interroge les quatre catalogues à la fois et fond leurs résultats, les plus légers d'abord. Quatre défauts corrigés en chemin, tous trouvés en essayant plutôt qu'en relisant : un jeton régénéré à chaque démarrage, qui aurait obligé à reconfigurer le mobile ; un 403 de Wikimedia dû à un agent trop générique, que la fusion des sources masquait ; un dossier de données par défaut inaccessible au compte qui lance le serveur ; et une lecture de métadonnées qui n'itérait rien, faute d'être asynchrone.
93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
"""Annuaire des stations et lecture des métadonnées ICY."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
import urllib.parse
|
|
|
|
import httpx
|
|
|
|
UA = {"User-Agent": "Noxely/1.0 (https://git.tuxtech.fr/tuxgyver/Noxely)"}
|
|
BASE = "https://de1.api.radio-browser.info/json/stations/search"
|
|
|
|
#: Un titre ICY court jusqu'au « '; » : les apostrophes y sont trop fréquentes pour s'arrêter
|
|
#: à la première (« Ain't That a Kick in the Head » deviendrait « Ain »).
|
|
STREAM_TITLE = re.compile(r"StreamTitle='(.*?)';")
|
|
|
|
GENRES = ["pop", "rock", "jazz", "classical", "electronic", "news", "chanson"]
|
|
|
|
|
|
async def stations(genre: str | None = None, name: str | None = None, limit: int = 60) -> list[dict]:
|
|
params = {"limit": str(limit), "hidebroken": "true", "order": "clickcount", "reverse": "true"}
|
|
if genre:
|
|
params["tag"] = genre
|
|
if name:
|
|
params["name"] = name
|
|
|
|
# L'annuaire est un service communautaire : une coupure passagère ne doit pas faire
|
|
# tomber le balayage. Mesuré en conditions réelles, d'où cette seconde tentative.
|
|
rows = None
|
|
async with httpx.AsyncClient(headers=UA, timeout=httpx.Timeout(30.0)) as client:
|
|
for attempt in range(2):
|
|
try:
|
|
response = await client.get(BASE, params=params)
|
|
response.raise_for_status()
|
|
rows = response.json()
|
|
break
|
|
except Exception:
|
|
if attempt == 1:
|
|
raise
|
|
await asyncio.sleep(2)
|
|
|
|
out = []
|
|
for row in rows:
|
|
url = row.get("url_resolved") or ""
|
|
# Les flux HLS se lisent mais ne s'enregistrent pas en copiant les octets.
|
|
if int(row.get("hls") or 0) != 0 or not url.startswith("http"):
|
|
continue
|
|
out.append({
|
|
"id": row.get("stationuuid") or url,
|
|
"name": (row.get("name") or "Station").strip(),
|
|
"url": url,
|
|
"codec": row.get("codec") or "",
|
|
"bitrate": int(row.get("bitrate") or 0),
|
|
"country": row.get("country") or "",
|
|
})
|
|
return out
|
|
|
|
|
|
async def now_playing(stream_url: str) -> str | None:
|
|
"""Le morceau diffusé à l'instant : aucun annuaire ne l'expose, il faut ouvrir le flux."""
|
|
headers = {**UA, "Icy-MetaData": "1"}
|
|
try:
|
|
async with httpx.AsyncClient(headers=headers, timeout=httpx.Timeout(10.0)) as client:
|
|
async with client.stream("GET", stream_url) as response:
|
|
if response.status_code >= 300:
|
|
return None
|
|
interval = int(response.headers.get("icy-metaint") or 0)
|
|
if interval <= 0:
|
|
return None
|
|
return await _read_title(response, interval)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def _read_title(response, interval: int) -> str | None:
|
|
# Client asynchrone : `iter_bytes` n'itère rien ici, il faut `aiter_bytes`.
|
|
buffer = bytearray()
|
|
needed = interval + 1
|
|
async for chunk in response.aiter_bytes():
|
|
buffer.extend(chunk)
|
|
if len(buffer) < needed:
|
|
continue
|
|
length = buffer[interval] * 16
|
|
if length == 0:
|
|
return None
|
|
needed = interval + 1 + length
|
|
if len(buffer) < needed:
|
|
continue
|
|
meta = bytes(buffer[interval + 1:needed]).decode("utf-8", "replace")
|
|
found = STREAM_TITLE.search(meta)
|
|
return found.group(1).strip() or None if found else None
|
|
return None
|