tuxgyver
51b60cda70
Recopier un jeton de trente-trois caractères sur un clavier tactile n'était pas un réglage, c'était une corvée. On ouvre désormais une fenêtre depuis le serveur, on appuie sur Appairer côté téléphone, et c'est fait. Le jeton n'est pas pour autant distribué : ouvrir la fenêtre l'exige déjà, elle dure deux minutes, se referme dès qu'un appareil s'est servi, et ne répond qu'à une adresse privée. Le champ de saisie reste là pour qui préfère le coller. Les cinquante fichiers de code portent en outre leur notice de licence, comme la GPL le recommande. Elle est en français, comme le reste ; le texte qui fait foi demeure celui de LICENSE, non traduit.
103 lines
3.9 KiB
Python
103 lines
3.9 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.
|
|
|
|
"""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
|