Roman
fed61adc56
Deux défauts enchaînés : la requête Kitsu cumulait les genres en ET (quatre catégories → 1 seul titre, précisément le téléchargement à l'origine des genres), et l'exclusion échouait car « Titre - Saison 1 - E3 » ne normalisait jamais vers « Titre » canonique. - Une requête Kitsu par genre, fusion entrelacée (chaque genre contribue) - normalize_title retire saison/épisode en boucle avec frontières de mots (« - Saison 1 - E3 », « S1 E1 » → titre canonique ; « Série 9 » intact) - test de non-régession : le possédé avec marqueurs n'est plus recommandé
210 lines
8.3 KiB
Python
210 lines
8.3 KiB
Python
"""Enrichissement de métadonnées via l'API Kitsu (https://kitsu.io/api/edge).
|
|
|
|
Fusion intelligente : seuls les champs manquants d'un TitleDetails sont complétés.
|
|
Les échecs (réseau, cache indisponible) sont loggés en warning et jamais remontés
|
|
au caller — la fiche d'origine est retournée inchangée (dégradation gracieuse).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import sqlite3
|
|
|
|
import httpx
|
|
|
|
from app.config import get_settings
|
|
from app.db import db
|
|
from app.scrapers.base import TitleDetails
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_NOISE_WORDS_RE = re.compile(
|
|
r"\b(?:VOSTFR\d*|VOST|VF[IV]?|TRUEFRENCH|FRENCH|MULTI|SUBFR?)\b", re.IGNORECASE
|
|
)
|
|
_TRAILING_SEASON_RE = re.compile(r"[\s\-–—:.]*\b(?:saison|season)\s*\d+\s*$", re.IGNORECASE)
|
|
_TRAILING_CODE_RE = re.compile(r"[\s\-–—:.]*\bS\d+(?:\s*E\d+)?\s*$", re.IGNORECASE)
|
|
_TRAILING_EPISODE_RE = re.compile(r"[\s\-–—:.]*\b(?:e|ep|episode)\s*\d+\s*$", re.IGNORECASE)
|
|
|
|
|
|
def normalize_title(title: str) -> str:
|
|
"""Nettoie un titre de scraping avant comparaison/recherche Kitsu.
|
|
|
|
Retire le bruit (VF/VOSTFR…), les marqueurs de saison/épisode en fin de titre
|
|
(« - Saison 1 - E3 », « S1 E1 ») — appliqués en boucle : un titre canonique
|
|
émergera identique d'un téléchargement (« Titre - Saison 1 - E3 ») ou d'une
|
|
fiche Kitsu (« Titre »), condition de l'exclusion du déjà-possédé.
|
|
"""
|
|
cleaned = _NOISE_WORDS_RE.sub(" ", title)
|
|
previous = None
|
|
while previous != cleaned:
|
|
previous = cleaned
|
|
cleaned = _TRAILING_SEASON_RE.sub("", cleaned)
|
|
cleaned = _TRAILING_CODE_RE.sub("", cleaned)
|
|
cleaned = _TRAILING_EPISODE_RE.sub("", cleaned)
|
|
cleaned = re.sub(r"\s*[-–—_]+\s*", " ", cleaned)
|
|
cleaned = re.sub(r"\s+", " ", cleaned)
|
|
return cleaned.strip(" -–—:.")
|
|
|
|
|
|
class KitsuService:
|
|
"""Recherche et cache des métadonnées anime depuis l'API Kitsu."""
|
|
|
|
async def search_anime(self, title: str) -> dict | None:
|
|
"""Recherche le meilleur match Kitsu pour un titre (None si échec/rien trouvé)."""
|
|
settings = get_settings()
|
|
query = normalize_title(title)
|
|
if not query:
|
|
return None
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
timeout=settings.http_timeout,
|
|
headers={
|
|
"User-Agent": settings.user_agent,
|
|
"Accept": "application/vnd.api+json",
|
|
},
|
|
) as client:
|
|
response = await client.get(
|
|
f"{settings.kitsu_base_url}/anime",
|
|
params={"filter[text]": query, "page[limit]": 5, "include": "genres"},
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
except (httpx.HTTPError, ValueError) as exc:
|
|
logger.warning("Recherche Kitsu échouée pour %r : %s", title, exc)
|
|
return None
|
|
return self._pick_best_match(payload, query)
|
|
|
|
async def enrich(self, details: TitleDetails) -> TitleDetails:
|
|
"""Complète les champs manquants de details via Kitsu (jamais episodes/url/source)."""
|
|
query = normalize_title(details.title)
|
|
cache_key = f"kitsu:anime:{query.casefold()}"
|
|
|
|
result = await self.get_cached(cache_key)
|
|
from_cache = result is not None
|
|
if result is None:
|
|
result = await self.search_anime(details.title)
|
|
if result is not None:
|
|
await self.set_cached(cache_key, result)
|
|
|
|
if result is None:
|
|
logger.info("Aucune métadonnée Kitsu pour %r — fiche inchangée", details.title)
|
|
return details
|
|
|
|
attrs = result.get("attributes", {})
|
|
if not details.synopsis:
|
|
details.synopsis = attrs.get("synopsis")
|
|
if not details.image_url:
|
|
poster = attrs.get("posterImage") or {}
|
|
details.image_url = poster.get("large") or poster.get("medium")
|
|
if not details.banner_url:
|
|
cover = attrs.get("coverImage") or {}
|
|
details.banner_url = cover.get("large") or cover.get("original")
|
|
if not details.genres:
|
|
details.genres = list(result.get("genres", []))
|
|
if details.rating is None:
|
|
details.rating = self._to_rating_10(attrs.get("averageRating"))
|
|
if details.year is None:
|
|
details.year = self._extract_year(attrs.get("startDate"))
|
|
if details.episode_count is None:
|
|
details.episode_count = attrs.get("episodeCount")
|
|
|
|
logger.debug(
|
|
"Fiche %r enrichie via Kitsu (cache=%s, id=%s)",
|
|
details.title,
|
|
from_cache,
|
|
result.get("id"),
|
|
)
|
|
return details
|
|
|
|
async def get_cached(self, cache_key: str) -> dict | None:
|
|
"""Retourne le payload en cache s'il existe et est frais (TTL), sinon None."""
|
|
ttl_hours = get_settings().metadata_cache_ttl_hours
|
|
try:
|
|
row = await db.fetchone(
|
|
"SELECT payload FROM metadata_cache "
|
|
f"WHERE cache_key = ? AND fetched_at > datetime('now', '-{ttl_hours} hours')",
|
|
(cache_key,),
|
|
)
|
|
except (RuntimeError, sqlite3.Error) as exc:
|
|
logger.warning(
|
|
"Cache métadonnées illisible (%s), continuation sans cache : %s", cache_key, exc
|
|
)
|
|
return None
|
|
if row is None:
|
|
return None
|
|
try:
|
|
return json.loads(row["payload"])
|
|
except (TypeError, ValueError) as exc:
|
|
logger.warning("Cache métadonnées corrompu pour %s, re-fetch : %s", cache_key, exc)
|
|
return None
|
|
|
|
async def set_cached(self, cache_key: str, payload: dict) -> None:
|
|
"""Écrit/upsert une entrée de cache (échec non bloquant, warning seulement)."""
|
|
try:
|
|
await db.execute(
|
|
"INSERT INTO metadata_cache (cache_key, payload, fetched_at) "
|
|
"VALUES (?, ?, datetime('now')) "
|
|
"ON CONFLICT(cache_key) DO UPDATE SET "
|
|
"payload = excluded.payload, fetched_at = excluded.fetched_at",
|
|
(cache_key, json.dumps(payload, ensure_ascii=False)),
|
|
)
|
|
except (RuntimeError, sqlite3.Error) as exc:
|
|
logger.warning(
|
|
"Cache métadonnées non écrit (%s), continuation sans cache : %s", cache_key, exc
|
|
)
|
|
|
|
@staticmethod
|
|
def _pick_best_match(payload: dict, query: str) -> dict | None:
|
|
items = [item for item in payload.get("data", []) if item.get("type") == "anime"]
|
|
if not items:
|
|
return None
|
|
|
|
genre_names = {
|
|
obj.get("id"): obj.get("attributes", {}).get("name")
|
|
for obj in payload.get("included", [])
|
|
if obj.get("type") == "genres"
|
|
}
|
|
|
|
def titles_of(item: dict) -> set[str]:
|
|
attrs = item.get("attributes", {})
|
|
titles = attrs.get("titles") or {}
|
|
return {
|
|
t.strip().casefold()
|
|
for t in (attrs.get("canonicalTitle"), titles.get("en"), titles.get("en_jp"))
|
|
if t
|
|
}
|
|
|
|
def popularity(item: dict) -> tuple[int, float]:
|
|
attrs = item.get("attributes", {})
|
|
return (
|
|
attrs.get("userCount") or 0,
|
|
float(attrs.get("averageRating") or 0),
|
|
)
|
|
|
|
best = next((item for item in items if query.casefold() in titles_of(item)), None)
|
|
if best is None:
|
|
best = max(items, key=popularity)
|
|
|
|
genre_ids = best.get("relationships", {}).get("genres", {}).get("data") or []
|
|
genres = [genre_names[g["id"]] for g in genre_ids if g.get("id") in genre_names]
|
|
return {"id": best.get("id"), "attributes": best.get("attributes", {}), "genres": genres}
|
|
|
|
@staticmethod
|
|
def _to_rating_10(average_rating: str | None) -> float | None:
|
|
"""Kitsu note sur 100 (chaîne) → note sur 10 arrondie à 1 décimale."""
|
|
if average_rating is None:
|
|
return None
|
|
try:
|
|
return round(float(average_rating) / 10, 1)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
@staticmethod
|
|
def _extract_year(start_date: str | None) -> int | None:
|
|
if not start_date:
|
|
return None
|
|
try:
|
|
return int(str(start_date)[:4])
|
|
except ValueError:
|
|
return None
|