Noxely/server/noxely_server/recorder.py

# 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.

"""Enregistrement d'une station à la demande, jusqu'à l'arrêt ou l'expiration de la durée.

À la différence du guet, qui s'arrête quand la station annonce un autre morceau, on capte ici
le flux tel quel. Le fichier rejoint la bibliothèque comme n'importe quel média : le mobile le
récupère à sa prochaine synchronisation, et l'interface web peut le réécouter.
"""
from __future__ import annotations

import asyncio
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime

import httpx

from .config import Config
from .sources import radio
from .storage import Storage
from .workers import UA, safe_name

#: Enregistrements simultanés au plus : chacun tient un flux ouvert en permanence.
MAX_PARALLEL = 4
#: Nouvelles tentatives d'affilée avant d'abandonner un flux qui ne revient pas.
MAX_RECONNECTS = 5
#: En deçà, un enregistrement interrompu ne vaut pas d'être conservé.
MIN_KEPT_BYTES = 64 * 1024


class AlreadyRecording(Exception):
    """La station est déjà en cours d'enregistrement."""


class TooMany(Exception):
    """Trop d'enregistrements simultanés."""


@dataclass
class Active:
    media_id: str
    station: str
    url: str
    started_at: int
    limit_minutes: int
    written: int = 0
    stop: asyncio.Event = field(default_factory=asyncio.Event)
    task: asyncio.Task | None = None

    def public(self) -> dict:
        return {
            "media_id": self.media_id, "station": self.station, "started_at": self.started_at,
            "written": self.written, "limit_minutes": self.limit_minutes,
        }


class Recorder:
    def __init__(self, config: Config, storage: Storage) -> None:
        self.config = config
        self.storage = storage
        self._active: dict[str, Active] = {}

    # --- Pilotage -------------------------------------------------------

    def start(self, station: dict, minutes: int | None = None) -> str:
        """Lance l'enregistrement d'une station ; à appeler depuis la boucle d'événements."""
        if any(a.url == station["url"] for a in self._active.values()):
            raise AlreadyRecording(station["name"])
        if len(self._active) >= MAX_PARALLEL:
            raise TooMany(str(MAX_PARALLEL))

        cap = self.config.record_max_minutes
        limit = min(minutes, cap) if minutes and minutes > 0 else cap
        codec = (station.get("codec") or "").upper()
        extension, mime = (
            ("aac", "audio/aac") if codec.startswith("AAC")
            else ("ogg", "audio/ogg") if codec.startswith("OGG")
            else ("mp3", "audio/mpeg")
        )
        stamp = datetime.now()
        title = f"{station['name']} — {stamp:%d/%m/%Y %H:%M}"
        # Le suffixe aléatoire évite qu'un second enregistrement de la même station, lancé dans
        # la même minute, écrase le fichier du premier.
        filename = safe_name(f"{station['name']}_{stamp:%Y-%m-%d_%H%M}_{uuid.uuid4().hex[:6]}", extension)
        media_id = self.storage.add_media(
            title=title, artist=station["name"], source="radio", origin=station["name"],
            filename=filename, remote_url=None, mime=mime, state="downloading",
        )
        entry = Active(media_id, station["name"], station["url"], _now(), limit)
        entry.task = asyncio.create_task(self._run(entry, filename))
        self._active[media_id] = entry
        return media_id

    def stop(self, media_id: str) -> bool:
        entry = self._active.get(media_id)
        if not entry:
            return False
        entry.stop.set()
        return True

    async def stop_all(self, timeout: float = 8.0) -> None:
        """À l'arrêt du serveur : clôt proprement les fichiers plutôt que de les laisser en cours."""
        tasks = [a.task for a in self._active.values() if a.task]
        for media_id in list(self._active):
            self.stop(media_id)
        if tasks:
            await asyncio.wait(tasks, timeout=timeout)

    def active(self) -> list[dict]:
        return [a.public() for a in self._active.values()]

    def recover(self) -> None:
        """Au démarrage : ce qui était « en cours » ne l'est plus, l'ancien processus est mort.

        Un fichier qui contient de l'audio reste lisible : on le déclare prêt plutôt que de
        perdre une écoute entière sur un simple redémarrage du service.
        """
        for item in self.storage.stuck_recordings():
            path = self.config.media_dir / item.filename
            size = path.stat().st_size if path.exists() else 0
            if size >= MIN_KEPT_BYTES:
                self.storage.update_media(item.id, state="ready", size=size,
                                          error="interrompu par un arrêt du serveur")
            else:
                path.unlink(missing_ok=True)
                self.storage.update_media(item.id, state="failed", error="interrompu, rien de capté")

    # --- Capture --------------------------------------------------------

    async def _run(self, entry: Active, filename: str) -> None:
        target = self.config.media_dir / filename
        began = time.monotonic()
        deadline = began + entry.limit_minutes * 60
        failures = 0
        problem = ""

        def keep_going(total: int, _title: str | None) -> bool:
            entry.written = base + total
            return not entry.stop.is_set() and time.monotonic() < deadline

        try:
            while not entry.stop.is_set() and time.monotonic() < deadline:
                base = entry.written
                try:
                    async with httpx.AsyncClient(
                        headers={**UA, "Icy-MetaData": "1"}, follow_redirects=True,
                        timeout=httpx.Timeout(30.0, read=60.0),
                    ) as client:
                        async with client.stream("GET", entry.url) as response:
                            response.raise_for_status()
                            interval = int(response.headers.get("icy-metaint") or 0)
                            with target.open("ab") as handle:
                                await radio.copy_stream(response, handle, interval, keep_going)
                except Exception as error:
                    problem = str(error)[:150] or type(error).__name__
                # Une connexion qui a livré de l'audio remet le compteur à zéro : seules les
                # tentatives infructueuses d'affilée comptent, pas une coupure après une heure.
                failures = 0 if entry.written > base else failures + 1
                if entry.stop.is_set() or time.monotonic() >= deadline:
                    break
                if failures > MAX_RECONNECTS:
                    break
                await asyncio.sleep(3)
        finally:
            self._finish(entry, target, time.monotonic() - began, problem)
            self._active.pop(entry.media_id, None)

    def _finish(self, entry: Active, target, elapsed: float, problem: str) -> None:
        size = target.stat().st_size if target.exists() else 0
        completed = entry.stop.is_set() or elapsed >= entry.limit_minutes * 60
        if size >= MIN_KEPT_BYTES or (size > 0 and completed):
            self.storage.update_media(
                entry.media_id, state="ready", size=size, duration_ms=int(elapsed * 1000),
                error=None if completed else (f"interrompu : {problem}" if problem else "interrompu"),
            )
        else:
            target.unlink(missing_ok=True)
            self.storage.update_media(
                entry.media_id, state="failed",
                error=problem or "aucune donnée reçue de la station",
            )


def _now() -> int:
    return int(time.time() * 1000)