ohm_streaming/app/services/torznab.py

"""Indexeur Torznab/Newznab : expose le catalogue OhmStreaming à Sonarr/Prowlarr.

OhmStreaming devient une source d'indexeur à part entière : Sonarr (ou Prowlarr,
qui relaiera vers Radarr/Lidarr…) interroge `/torznab/api` comme n'importe quel
indexer Jackett. Chaque « release » correspond à un épisode résolu depuis les
sources de scraping ; le grab (`/torznab/download`) déclenche l'extraction puis
l'ajout dans la file de téléchargements OhmStreaming — le fichier arrive donc
dans la bibliothèque locale, comme un téléchargement manuel.

Formats :
- `t=caps`      → capacités du serveur (catégories TV/Anime, paramètres supportés)
- `t=tvsearch`  → recherche par série + saison + épisode (Sonarr)
- `t=search`    → recherche libre (Prowlarr, recherche manuelle)
"""


import asyncio
import hashlib
import logging
import time
from dataclasses import dataclass
from datetime import UTC, datetime
from email.utils import format_datetime
from urllib.parse import urlencode

from app.scrapers.base import (
    Episode,
    ScrapeError,
    SourceScraper,
    VideoLink,
    all_sources,
    import_all_scrapers,
    resolve_hoster,
)
from app.services.downloads import download_manager
from app.services.settings import is_source_enabled

logger = logging.getLogger(__name__)

import_all_scrapers()

TORZNAB_NS = "http://torznab.com/schemas/2015/feed"
CAT_TV = "5000"
CAT_ANIME = "5070"
_SIZE_ESTIMATE = 400 * 1024 * 1024  # estimation affichée (~400 Mo/épisode)
_MAX_SERIES_PER_SOURCE = 2   # fiches explorées par source lors d'une recherche
_MAX_RELEASES = 100          # borne du flux retourné
_SEARCH_TIMEOUT = 40.0       # scraping lent : garde-fou par source
_EPISODE_CACHE_TTL = 600.0   # listes d'épisodes re-scrapées au bout de 10 min


@dataclass
class Release:
    """Un épisode vu comme une release par Sonarr/Prowlarr."""

    series: str
    season: int
    ep: int
    source: str
    source_id: str
    episode_url: str

    @property
    def sonarr_title(self) -> str:
        return f"{self.series} S{self.season:02d}E{self.ep:02d} VOSTFR WEB-DL"



def _bencode(value) -> bytes:
    if isinstance(value, int):
        return f"i{value}e".encode()
    if isinstance(value, str):
        raw = value.encode()
        return f"{len(raw)}:".encode() + raw
    if isinstance(value, bytes):
        return f"{len(value)}:".encode() + value
    if isinstance(value, dict):
        return b"d" + b"".join(
            _bencode(k) + _bencode(v) for k, v in sorted(value.items())
        ) + b"e"
    if isinstance(value, list):
        return b"l" + b"".join(_bencode(v) for v in value) + b"e"
    raise TypeError(f"type non encodable en bencode : {type(value)!r}")


def torrent_stub(announce_url: str, name: str) -> bytes:
    """Fichier .torrent minimal (le vrai téléchargement est fait par OhmStreaming)."""
    return build_stub(announce_url, name)[0]


def build_stub(announce_url: str, name: str) -> tuple[bytes, str]:
    """Fichier .torrent de service + infohash SHA-1 (identité côté Sonarr).

    L'announce embarque les paramètres du grab (source, sid, season, ep, series) :
    quand Sonarr renvoie ce .torrent à l'API compatible qBittorrent d'Ohm,
    le grab est rejoué à l'identique.
    """
    info = {"name": name + ".mp4", "length": 0, "piece length": 32768, "pieces": b"\x00" * 20}
    data = _bencode(
        {
            "announce": announce_url,
            "created by": "OhmStreaming",
            "comment": name,
            "info": info,
        }
    )
    return data, hashlib.sha1(_bencode(info)).hexdigest()


def bdecode(data: bytes):
    """Décode un flux bencode (les clés dict reviennent en bytes)."""

    def _parse(offset: int) -> tuple[object, int]:
        char = data[offset : offset + 1]
        if char == b"i":
            end = data.index(b"e", offset)
            return int(data[offset + 1 : end]), end + 1
        if char in (b"d", b"l"):
            is_dict = char == b"d"
            items: dict | list = {} if is_dict else []
            offset += 1
            while data[offset : offset + 1] != b"e":
                first, offset = _parse(offset)
                if is_dict:
                    second, offset = _parse(offset)
                    items[first] = second
                else:
                    items.append(first)
            return items, offset + 1
        if char.isdigit():
            colon = data.index(b":", offset)
            length = int(data[offset:colon])
            start = colon + 1
            return data[start : start + length], start + length
        raise ValueError(f"bencode invalide à l'octet {offset}")

    value, end = _parse(0)
    if end != len(data):
        raise ValueError("données après la fin du flux bencode")
    return value


class TorznabService:
    """Recherche multi-sources mappée en releases Torznab + grab → file interne."""

    def __init__(self) -> None:
        self._episodes_cache: dict[tuple[str, str], tuple[float, list[Episode]]] = {}

    # ------------------------------------------------------------ recherche

    async def tvsearch(self, q: str, season: int | None = None, ep: int | None = None) -> list[Release]:
        """Recherche type Sonarr : série (+ saison/épisode optionnels)."""
        releases: list[Release] = []
        for source in await self._enabled_sources():
            outcomes = await self._search_source(source, q)
            for result in outcomes[:_MAX_SERIES_PER_SOURCE]:
                try:
                    episodes = await self._episodes_of(source, result.source_id)
                except (ScrapeError, TimeoutError):
                    continue
                releases.extend(
                    self._releases_for(result.title, source, result.source_id, episodes, season, ep)
                )
                if len(releases) >= _MAX_RELEASES:
                    return releases[:_MAX_RELEASES]
        return releases

    async def search(self, q: str) -> list[Release]:
        """Recherche libre : tous les épisodes des fiches trouvées."""
        return await self.tvsearch(q)

    async def latest_releases(self) -> list[Release]:
        """Flux RSS (Sonarr) : dernier épisode de chaque nouveauté du catalogue."""
        releases: list[Release] = []
        for source in await self._enabled_sources():
            try:
                latest = await asyncio.wait_for(source.latest(), timeout=_SEARCH_TIMEOUT)
            except (ScrapeError, TimeoutError) as exc:
                logger.warning("Torznab : nouveautés %s KO : %s", source.name, exc)
                continue
            for result in latest[:_MAX_SERIES_PER_SOURCE]:
                try:
                    episodes = await self._episodes_of(source, result.source_id)
                except (ScrapeError, TimeoutError):
                    continue
                whole = [e for e in episodes if e.number == int(e.number)]
                if not whole:
                    continue
                newest = max(whole, key=lambda e: (e.season, e.number))
                releases.extend(
                    self._releases_for(result.title, source, result.source_id, [newest], None, None)
                )
                if len(releases) >= _MAX_RELEASES:
                    return releases[:_MAX_RELEASES]
        return releases

    async def _search_source(self, source: SourceScraper, q: str):
        try:
            return await asyncio.wait_for(source.search(q), timeout=_SEARCH_TIMEOUT)
        except (ScrapeError, TimeoutError) as exc:
            logger.warning("Torznab : recherche %s KO pour %r : %s", source.name, q, exc)
            return []

    async def _enabled_sources(self) -> list[SourceScraper]:
        return [s for s in all_sources() if await is_source_enabled(s.name)]

    async def _episodes_of(self, source: SourceScraper, source_id: str) -> list[Episode]:
        """Liste d'épisodes d'une fiche, avec cache mémoire TTL."""
        key = (source.name, source_id)
        cached = self._episodes_cache.get(key)
        if cached is not None and cached[0] > time.monotonic():
            return cached[1]
        episodes = await asyncio.wait_for(source.list_episodes(source_id), timeout=_SEARCH_TIMEOUT)
        self._episodes_cache[key] = (time.monotonic() + _EPISODE_CACHE_TTL, episodes)
        return episodes

    def _releases_for(
        self,
        series: str,
        source: SourceScraper,
        source_id: str,
        episodes: list[Episode],
        season: int | None,
        ep: int | None,
    ) -> list[Release]:
        """Filtre les épisodes selon saison/épisode demandés (entiers uniquement)."""
        out = []
        for item in episodes:
            if item.number != int(item.number):  # OAV 2.5 → ignorée (inparseable Sonarr)
                continue
            if season is not None and item.season != season:
                continue
            if ep is not None and int(item.number) != ep:
                continue
            out.append(
                Release(
                    series=series,
                    season=item.season,
                    ep=int(item.number),
                    source=source.name,
                    source_id=source_id,
                    episode_url=item.url,
                )
            )
        return out

    # ------------------------------------------------------------ grab

    async def grab(
        self,
        source: str,
        source_id: str,
        season: int,
        ep: int,
        series: str,
        sonarr_hash: str | None = None,
    ) -> dict:
        """Résout l'épisode (embed → vidéo directe) puis l'ajoute à la file interne.

        Retourne le dict du téléchargement (existant si doublon actif).
        Lève ScrapeError si introuvable ou qu'aucun hébergeur n'a répondu.
        sonarr_hash : infohash du .torrent de service — les téléchargements
        Sonarr sont préfixés « sonarr:<hash>| » pour rester suivis via l'API
        compatible qBittorrent.
        """
        from app.scrapers.base import get_source

        scraper = get_source(source)
        episodes = await self._episodes_of(scraper, source_id)
        match = next(
            (
                e
                for e in episodes
                if e.season == season and e.number == int(ep)
            ),
            None,
        )
        if match is None:
            raise ScrapeError(f"Épisode S{season:02d}E{ep:02d} introuvable sur {source}")

        link = await self._resolve_video(scraper, match.url)
        title = f"{series} S{season:02d}E{ep:02d}"
        key = f"sonarr:{sonarr_hash}|{link.url}" if sonarr_hash else link.url
        result = await download_manager.enqueue(link.url, match.url, title, source_key=key)
        if link.is_hls or link.headers.get("Referer"):
            result["note"] = "HLS/proxy : OhmStreaming gère le téléchargement via ffmpeg"
        logger.info("Torznab grab %s → download id=%s", title, result.get("id"))
        return result

    async def _resolve_video(self, scraper: SourceScraper, episode_url: str) -> VideoLink:
        """Chaîne complète : page épisode → embeds → première URL directe valide."""
        embeds = await asyncio.wait_for(
            scraper.extract_embed_links(episode_url), timeout=_SEARCH_TIMEOUT
        )
        errors: list[str] = []
        for embed in embeds:
            extractor = resolve_hoster(embed)
            if extractor is None:
                errors.append(f"hébergeur non supporté : {embed}")
                continue
            try:
                link = await asyncio.wait_for(extractor.extract(embed), timeout=_SEARCH_TIMEOUT)
                if link and link.url:
                    return link
            except (ScrapeError, TimeoutError) as exc:
                errors.append(str(exc))
        raise ScrapeError(f"Aucun hébergeur résolu pour {episode_url} ({'; '.join(errors[:3])})")

    # ------------------------------------------------------------ XML

    def download_url(self, base_url: str, apikey: str, release: Release) -> str:
        query = urlencode(
            {
                "apikey": apikey,
                "source": release.source,
                "sid": release.source_id,
                "season": release.season,
                "ep": release.ep,
                "series": release.series,
            }
        )
        return f"{base_url}/torznab/download?{query}"

    def caps_xml(self, base_url: str) -> str:
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<caps>
  <server version="1.0" title="OhmStreaming" url="{base_url}"
          email="ohm@localhost" image="{base_url}/static/img/logo.png"/>
  <searching>
    <search available="yes" supportedParams="q"/>
    <tv-search available="yes" supportedParams="q,season,ep"/>
    <movie-search available="no" supportedParams=""/>
    <audio-search available="no" supportedParams=""/>
  </searching>
  <categories>
    <category id="{CAT_TV}" name="TV">
      <subcat id="{CAT_ANIME}" name="Anime"/>
    </category>
  </categories>
</caps>"""

    def results_xml(self, base_url: str, apikey: str, releases: list[Release]) -> str:
        from xml.sax.saxutils import escape

        items = []
        for release in releases:
            url = escape(self.download_url(base_url, apikey, release))
            pubdate = escape(format_datetime(datetime.now(UTC)))
            items.append(f"""    <item>
      <title>{escape(release.sonarr_title)}</title>
      <guid isPermaLink="true">{url}</guid>
      <link>{url}</link>
      <comments>{escape(release.episode_url)}</comments>
      <pubDate>{pubdate}</pubDate>
      <category>{CAT_ANIME}</category>
      <enclosure url="{url}" length="{_SIZE_ESTIMATE}" type="application/x-bittorrent"/>
      <torznab:attr name="seeders" value="1"/>
      <torznab:attr name="peers" value="1"/>
      <torznab:attr name="downloadvolumefactor" value="0"/>
      <torznab:attr name="uploadvolumefactor" value="0"/>
    </item>""")
        body = "\n".join(items)
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:torznab="{TORZNAB_NS}">
  <channel>
    <title>OhmStreaming</title>
    <link>{escape(base_url)}</link>
    <description>Indexeur OhmStreaming — animes VOSTFR scrapés en direct</description>
    <language>fr-FR</language>
{body}
  </channel>
</rss>"""

    def error_xml(self, code: int, description: str) -> str:
        from xml.sax.saxutils import escape

        return (
            '<?xml version="1.0" encoding="UTF-8"?>\n'
            f'<error code="{code}" description="{escape(description)}"/>'
        )


torznab = TorznabService()