chatbot-maisonnette/bot/presence_sources.py

"""Animes, musiques et jeux réels pour le statut Discord, lus sur AniList, Deezer et Steam."""

from __future__ import annotations

import asyncio
import json
import logging
from dataclasses import asdict, dataclass
from pathlib import Path

import aiohttp

logger = logging.getLogger("chatbot.presence")

ANILIST_URL = "https://graphql.anilist.co"
ANILIST_QUERY = """
query ($page: Int) {
  Page(page: $page, perPage: 50) {
    media(type: ANIME, sort: POPULARITY_DESC, isAdult: false) {
      title { romaji } episodes duration nextAiringEpisode { episode }
    }
  }
}
"""
ANIME_PAGES = 10  # 10 pages de 50 : le top 500 des animes les plus populaires
# Les tops Deezer contiennent 100 titres.
DEEZER_TRACKS_URL = "https://api.deezer.com/playlist/{}/tracks?limit=100"
STEAM_URL = "https://api.steampowered.com/IPlayerService/GetOwnedGames/v1/"
DEFAULT_TRACK_DURATION = 180  # secondes, si Deezer n'indique pas la durée
DEFAULT_EPISODE_DURATION = 24  # minutes, si AniList n'indique pas la durée
TIMEOUT = aiohttp.ClientTimeout(total=30)


class SourceError(Exception):
    """Lecture d'une source impossible. Le message ne contient jamais la clé Steam."""


@dataclass(frozen=True)
class Anime:
    title: str
    episodes: int | None  # épisodes déjà sortis (None : inconnu) ; 1 pour un film
    duration: int         # minutes par épisode


@dataclass(frozen=True)
class Track:
    title: str
    artist: str
    duration: int  # secondes


async def get_json(session: aiohttp.ClientSession, method: str, url: str, **kwargs) -> dict:
    try:
        async with session.request(method, url, **kwargs) as response:
            if response.status in (401, 403):
                raise SourceError(f"accès refusé (HTTP {response.status})")
            if response.status != 200:
                raise SourceError(f"réponse HTTP {response.status}")
            return await response.json(content_type=None)
    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError) as exc:
        # Pas de str(exc) : l'adresse de la requête, qu'il peut contenir, porte la clé Steam.
        raise SourceError(type(exc).__name__) from None


class PresenceSources:
    """Listes téléchargées au démarrage puis une fois par jour, gardées en cache sur le disque.

    Si une source ne répond pas, sa liste précédente (ou celle du cache) est conservée."""

    def __init__(self, cache_file: Path, deezer_playlists: tuple[int, ...],
                 steam_api_key: str | None, steam_id: str | None):
        self.cache_file = cache_file
        self.deezer_playlists = deezer_playlists
        self.steam_api_key = steam_api_key
        self.steam_id = steam_id
        self.anime: list[Anime] = []
        self.music: list[Track] = []
        self.games: list[str] = []

    @property
    def steam_enabled(self) -> bool:
        return bool(self.steam_api_key and self.steam_id)

    def load_cache(self) -> None:
        try:
            data = json.loads(self.cache_file.read_text(encoding="utf-8"))
            self.anime = [Anime(**a) for a in data.get("anime", [])]
            self.music = [Track(**t) for t in data.get("music", [])]
            if self.steam_enabled:
                self.games = list(data.get("games", []))
        except FileNotFoundError:
            pass
        except (OSError, ValueError, TypeError) as exc:
            logger.warning("Cache des statuts %s illisible, ignoré : %s", self.cache_file, exc)

    def save_cache(self) -> None:
        data = {"anime": [asdict(a) for a in self.anime], "music": [asdict(t) for t in self.music], "games": self.games}
        try:
            self.cache_file.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
        except OSError as exc:
            logger.warning("Impossible d'écrire le cache des statuts %s : %s", self.cache_file, exc)

    async def refresh(self) -> None:
        sources = [("anime", "Animes (AniList)", self.fetch_anime), ("music", "Musiques (Deezer)", self.fetch_music)]
        if self.steam_enabled:
            sources.append(("games", "Jeux (Steam)", self.fetch_games))
        async with aiohttp.ClientSession(timeout=TIMEOUT) as session:
            for attr, label, fetch in sources:
                try:
                    items = await fetch(session)
                    if not items:
                        raise SourceError("liste vide")
                except SourceError as exc:
                    logger.warning("%s : lecture impossible (%s), %d élément(s) conservé(s).",
                                   label, exc, len(getattr(self, attr)))
                else:
                    setattr(self, attr, items)
                    logger.info("%s : %d élément(s).", label, len(items))
        self.save_cache()

    @staticmethod
    async def fetch_anime(session: aiohttp.ClientSession) -> list[Anime]:
        anime: dict[str, Anime] = {}
        for page in range(1, ANIME_PAGES + 1):
            data = await get_json(session, "POST", ANILIST_URL,
                                  json={"query": ANILIST_QUERY, "variables": {"page": page}})
            try:
                media = data["data"]["Page"]["media"]
            except (KeyError, TypeError):
                raise SourceError("réponse inattendue") from None
            for m in media:
                title = (m.get("title") or {}).get("romaji")
                if not title:
                    continue
                # Anime en cours de diffusion : seuls les épisodes déjà sortis comptent.
                episodes = m.get("episodes")
                if episodes is None and m.get("nextAiringEpisode"):
                    episodes = m["nextAiringEpisode"]["episode"] - 1 or None
                anime[title] = Anime(title, episodes, m.get("duration") or DEFAULT_EPISODE_DURATION)
        return list(anime.values())

    async def fetch_music(self, session: aiohttp.ClientSession) -> list[Track]:
        tracks: dict[int, Track] = {}  # par identifiant Deezer : un titre présent dans deux tops ne compte qu'une fois
        for playlist in self.deezer_playlists:
            data = await get_json(session, "GET", DEEZER_TRACKS_URL.format(playlist))
            if "error" in data:  # Deezer signale ses erreurs avec un code HTTP 200
                raise SourceError(f"playlist {playlist} : {data['error'].get('message', 'erreur')}")
            for t in data.get("data", []):
                if t.get("id") and t.get("title") and t.get("artist", {}).get("name"):
                    tracks[t["id"]] = Track(t["title"], t["artist"]["name"], t.get("duration") or DEFAULT_TRACK_DURATION)
        return list(tracks.values())

    async def fetch_games(self, session: aiohttp.ClientSession) -> list[str]:
        params = {"key": self.steam_api_key, "steamid": self.steam_id, "include_appinfo": "1",
                  "include_played_free_games": "1", "format": "json"}
        data = await get_json(session, "GET", STEAM_URL, params=params)
        games = data.get("response", {}).get("games")
        if not games:
            raise SourceError("bibliothèque vide ou privée : « Détails des jeux » doit être public dans le profil Steam")
        return sorted({g["name"] for g in games if g.get("name")})