anicove2 / api /providers /mal_fallback.py
mwask's picture
Upload 124 files
19ecc0d verified
Raw
History Blame Contribute Delete
32.3 kB
"""
Jikan (MyAnimeList) API v4 fallback service.
Used as a third-tier fallback when both AniList GraphQL and Miruro API are
down or rate-limited. Provides home page data and anime info using the
free Jikan REST API (https://api.jikan.moe/v4).
The anime-lists mapping file is used to convert MAL IDs ↔ AniList IDs so
that cards rendered from Jikan data still link to /watch/<anilist_id>.
"""
import asyncio
import logging
import re
import time
from typing import Dict, Any, List, Optional
import aiohttp
logger = logging.getLogger(__name__)
JIKAN_BASE = "https://api.jikan.moe/v4"
MAPPING_URL = (
"https://raw.githubusercontent.com/Fribb/anime-lists/master/anime-list-full.json"
)
# ---------------------------------------------------------------------------
# Rate-limit guard — Jikan allows ~3 req/s
# ---------------------------------------------------------------------------
_JIKAN_SEMAPHORE = asyncio.Semaphore(3)
_JIKAN_MIN_INTERVAL = 0.35 # seconds between calls
_jikan_last_call: float = 0.0
class MalFallbackService:
"""Jikan-based fallback for home page data and anime info."""
def __init__(self):
# ID mapping caches
self._mapping: Optional[List[Dict]] = None
self._mapping_ts: float = 0.0
self._mapping_ttl: float = 86400.0 # 24 h
self._mal_to_al: Dict[int, int] = {}
self._al_to_mal: Dict[int, int] = {}
# Home data cache
self._home_cache: Optional[Dict[str, Any]] = None
self._home_cache_ts: float = 0.0
self._home_cache_ttl: float = 600.0 # 10 min
logger.debug("[MalFallback] Service initialised")
# ======================================================================
# Internal HTTP helpers
# ======================================================================
async def _jikan_get(self, path: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Rate-limited GET to Jikan. Returns parsed JSON or {}."""
global _jikan_last_call
async with _JIKAN_SEMAPHORE:
now = time.monotonic()
wait = _JIKAN_MIN_INTERVAL - (now - _jikan_last_call)
if wait > 0:
await asyncio.sleep(wait)
_jikan_last_call = time.monotonic()
url = f"{JIKAN_BASE}/{path.lstrip('/')}"
timeout = aiohttp.ClientTimeout(total=10)
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url, params=params) as resp:
if resp.status == 429:
logger.warning("[MalFallback] Jikan rate-limited, dropping request")
return {}
if resp.status != 200:
logger.error(f"[MalFallback] Jikan {resp.status} for {url}")
return {}
return await resp.json()
except Exception as e:
logger.error(f"[MalFallback] Jikan request failed for {url}: {e}")
return {}
# ======================================================================
# ID mapping
# ======================================================================
async def _ensure_mapping(self) -> None:
"""Load or refresh the MAL ↔ AniList mapping file."""
now = time.time()
if self._mapping and (now - self._mapping_ts) < self._mapping_ttl:
return
timeout = aiohttp.ClientTimeout(total=30)
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(MAPPING_URL) as resp:
if resp.status != 200:
logger.error(f"[MalFallback] Mapping fetch failed: {resp.status}")
return
data = await resp.json(content_type=None)
self._mapping = data
self._mapping_ts = time.time()
# Build bidirectional lookup dicts
mal_to_al: Dict[int, int] = {}
al_to_mal: Dict[int, int] = {}
for entry in data:
mal_id = entry.get("mal_id")
al_id = entry.get("anilist_id")
if mal_id and al_id:
try:
m = int(mal_id)
a = int(al_id)
mal_to_al[m] = a
al_to_mal[a] = m
except (ValueError, TypeError):
continue
self._mal_to_al = mal_to_al
self._al_to_mal = al_to_mal
logger.debug(
f"[MalFallback] Mapping loaded: {len(mal_to_al)} entries"
)
except Exception as e:
logger.error(f"[MalFallback] Mapping load failed: {e}")
def mal_to_anilist(self, mal_id: int) -> Optional[int]:
return self._mal_to_al.get(int(mal_id))
def anilist_to_mal(self, anilist_id: int) -> Optional[int]:
return self._al_to_mal.get(int(anilist_id))
# ======================================================================
# Normalizers — output matches AnilistHomeService / MiruroAnimeInfoService
# ======================================================================
def _normalize_jikan_anime(
self, item: Dict[str, Any], rank: int = 0
) -> Dict[str, Any]:
"""Convert a Jikan anime object into the unified card format."""
mal_id = item.get("mal_id")
anilist_id = self.mal_to_anilist(mal_id) if mal_id else None
display_id = str(anilist_id) if anilist_id else str(mal_id or "")
images = item.get("images", {})
poster = (
images.get("webp", {}).get("large_image_url")
or images.get("jpg", {}).get("large_image_url")
or images.get("jpg", {}).get("image_url")
or ""
)
english_title = item.get("title_english") or item.get("title") or "Unknown"
total_episodes = item.get("episodes") or 0
# Jikan doesn't directly give "released episodes" for airing shows,
# but if airing==True and episodes is set, use a reasonable estimate.
airing = item.get("airing", False)
released = total_episodes if not airing else max(total_episodes - 1, 1) if total_episodes else 0
# Studios
studios = item.get("studios", [])
studio_name = studios[0].get("name") if studios else ""
# Duration (e.g. "23 min per ep" → "23 min")
raw_duration = item.get("duration") or ""
duration = raw_duration.replace(" per ep", "").strip() if raw_duration else ""
# Rating: Jikan score is 0-10, AniList is 0-100
score = item.get("score")
rating = int(score * 10) if score else None
genres = [g.get("name") for g in (item.get("genres") or []) if g.get("name")]
return {
"id": display_id,
"anilistId": anilist_id,
"malId": mal_id,
"name": english_title,
"jname": item.get("title_japanese") or "",
"poster": poster,
"banner": "", # Jikan doesn't provide banner images
"episodes": {
"sub": total_episodes,
"dub": 0,
"released": released,
},
"type": item.get("type") or "",
"duration": duration,
"rating": rating,
"isAdult": (item.get("rating") or "").startswith("Rx"),
"rank": rank,
"description": "",
"genres": genres,
"otherInfo": [
item.get("type") or "",
duration,
studio_name,
],
}
def _normalize_jikan_spotlight(
self, item: Dict[str, Any], rank: int = 0
) -> Dict[str, Any]:
"""Convert a Jikan anime object into the spotlight card format."""
base = self._normalize_jikan_anime(item, rank)
# Synopsis
synopsis = item.get("synopsis") or ""
# Strip MAL source notes
if synopsis and "[Written by MAL" in synopsis:
synopsis = synopsis[: synopsis.index("[Written by MAL")].strip()
studios = item.get("studios", [])
studio_name = studios[0].get("name") if studios else ""
total_episodes = item.get("episodes") or 0
released = base["episodes"].get("released", 0)
base["description"] = synopsis
base["studio"] = studio_name
base["totalEpisodes"] = total_episodes or None
base["releasedEpisodes"] = released if released else total_episodes
base["season"] = (item.get("season") or "").upper()
base["seasonYear"] = item.get("year") or ""
base["nextEpisode"] = None
return base
def _normalize_jikan_info(self, item: Dict[str, Any], chars_list: Optional[List[Dict[str, Any]]] = None, recs_list: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
"""
Convert a Jikan /anime/{id}/full response into the same shape as
MiruroAnimeInfoService.get_anime_info().
"""
mal_id = item.get("mal_id")
anilist_id = self.mal_to_anilist(mal_id) if mal_id else None
images = item.get("images", {})
poster = (
images.get("webp", {}).get("large_image_url")
or images.get("jpg", {}).get("large_image_url")
or ""
)
english_title = item.get("title_english") or item.get("title") or "Unknown"
# Status mapping
status_map = {
"Currently Airing": "Currently Airing",
"Finished Airing": "Finished Airing",
"Not yet aired": "Not yet aired",
}
status = status_map.get(item.get("status", ""), item.get("status", ""))
total_episodes = item.get("episodes") or 0
airing = item.get("airing", False)
released_episodes = total_episodes if not airing else max(total_episodes - 1, 1) if total_episodes else 0
genres = [g.get("name") for g in (item.get("genres") or []) if g.get("name")]
themes = [t.get("name") for t in (item.get("themes") or []) if t.get("name")]
all_genres = genres + themes
studios = item.get("studios", [])
studios_list = [
{"id": s.get("mal_id"), "name": s.get("name")} for s in studios
]
# Duration
raw_duration = item.get("duration") or ""
duration = raw_duration.replace(" per ep", "").strip() if raw_duration else ""
# Score → AniList scale (0-100)
score = item.get("score")
rating_str = str(int(score * 10)) if score else ""
# Dates
aired = item.get("aired", {}) or {}
aired_string = aired.get("string") or ""
prop = aired.get("prop", {}) or {}
from_d = prop.get("from", {}) or {}
to_d = prop.get("to", {}) or {}
def fmt_date(d):
if not d or not d.get("year"):
return ""
months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
m = months[d["month"] - 1] if d.get("month") and 1 <= d["month"] <= 12 else ""
day = d.get("day", "")
return f"{m} {day}, {d['year']}".strip(", ")
aired_start = fmt_date(from_d)
aired_end = fmt_date(to_d)
season = (item.get("season") or "").upper()
season_year = item.get("year") or ""
premiered = f"{season} {season_year}".strip() if season else ""
# Synopsis
synopsis = item.get("synopsis") or ""
if synopsis and "[Written by MAL" in synopsis:
synopsis = synopsis[: synopsis.index("[Written by MAL")].strip()
# Source
source = (item.get("source") or "").replace("_", " ").title()
# Characters (from characters endpoint)
characters = []
for char_entry in (chars_list or []):
char = char_entry.get("character", {}) or {}
char_images = char.get("images", {}) or {}
role = char_entry.get("role") or "Supporting"
# Pick first Japanese VA
va = None
for v in (char_entry.get("voice_actors") or []):
if v.get("language", "").lower() == "japanese":
va = v
break
if not va and (char_entry.get("voice_actors") or []):
va = char_entry["voice_actors"][0]
va_person = (va or {}).get("person", {}) or {}
va_images = va_person.get("images", {}) or {}
characters.append({
"character": {
"id": str(char.get("mal_id", "")),
"name": char.get("name", ""),
"poster": char_images.get("webp", {}).get("image_url") or char_images.get("jpg", {}).get("image_url") or "",
"cast": role.title(),
},
"voiceActor": {
"id": str(va_person.get("mal_id", "")),
"name": va_person.get("name", ""),
"poster": va_images.get("jpg", {}).get("image_url") or "",
"cast": (va or {}).get("language", "Japanese"),
} if va else None,
})
# Relations
related = []
prequels = []
sequels = []
for rel in (item.get("relations") or []):
rel_type = rel.get("relation", "")
for entry in (rel.get("entry") or []):
if entry.get("type") != "anime":
continue
rel_mal_id = entry.get("mal_id")
rel_al_id = self.mal_to_anilist(rel_mal_id) if rel_mal_id else None
rel_entry = {
"id": str(rel_al_id or rel_mal_id or ""),
"anilistId": rel_al_id,
"malId": rel_mal_id,
"name": entry.get("name") or "",
"jname": "",
"poster": "", # Jikan relations don't include images
"type": "",
"rating": None,
"episodes_sub": 0,
"episodes_dub": 0,
"relation": rel_type.replace("_", " ").title(),
"badge": rel_type.upper(),
}
if rel_type.lower() == "prequel":
prequels.append(rel_entry)
elif rel_type.lower() == "sequel":
sequels.append(rel_entry)
related.append(rel_entry)
# Recommendations (from recommendations endpoint)
recommended = []
for rec in (recs_list or [])[:6]:
rec_entry = rec.get("entry", {}) or {}
rec_mal_id = rec_entry.get("mal_id")
rec_al_id = self.mal_to_anilist(rec_mal_id) if rec_mal_id else None
rec_images = rec_entry.get("images", {}) or {}
recommended.append({
"id": str(rec_al_id or rec_mal_id or ""),
"anilistId": rec_al_id,
"name": rec_entry.get("title") or "",
"jname": "",
"poster": rec_images.get("webp", {}).get("large_image_url") or rec_images.get("jpg", {}).get("large_image_url") or "",
"type": "",
"duration": "",
"rating": None,
"episodes_sub": 0,
"episodes_dub": 0,
})
# Trailer
promotional_videos = []
trailer = item.get("trailer", {}) or {}
if trailer.get("youtube_id") or trailer.get("embed_url"):
yt_id = trailer.get("youtube_id")
url = f"https://www.youtube.com/watch?v={yt_id}" if yt_id else (trailer.get("url") or "")
thumb = (trailer.get("images") or {}).get("maximum_image_url") or ""
if url:
promotional_videos.append({
"title": "Trailer",
"source": url,
"thumbnail": thumb,
})
return {
"anilistId": anilist_id,
"malId": mal_id,
"title": english_title,
"poster": poster,
"banner": "",
"description": synopsis,
"status": status,
"genres": all_genres,
"duration": duration,
"isAdult": (item.get("rating") or "").startswith("Rx"),
"type": item.get("type") or "",
"source": source,
"rating": rating_str,
"quality": "",
"total_sub_episodes": total_episodes,
"total_dub_episodes": total_episodes,
"released_episodes": released_episodes,
"japanese": item.get("title_japanese") or "",
"synonyms": ", ".join(item.get("title_synonyms") or []),
"aired": aired_string,
"aired_start": aired_start,
"aired_end": aired_end,
"premiered": premiered,
"studios": studios_list,
"producers": [
{"id": p.get("mal_id"), "name": p.get("name")}
for p in (item.get("producers") or [])
],
"malScore": str(score) if score else None,
"promotionalVideos": promotional_videos,
"charactersVoiceActors": [],
"characters": characters,
"seasons": [],
"relatedAnimes": related,
"recommendedAnimes": recommended,
"prequels": prequels,
"sequels": sequels,
"stats": {
"rating": rating_str,
"episodes": {
"sub": released_episodes,
"dub": released_episodes,
},
"type": item.get("type") or "",
"duration": duration,
"source": source,
},
"bannerImage": "",
"nextAiringEpisode": None,
}
# ======================================================================
# Episode annotation (same helper as AnilistHomeService)
# ======================================================================
def _annotate_episodes_count(
self, animes: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
for anime in animes:
episodes = anime.get("episodes", {})
sub_count = episodes.get("sub", 0)
released_count = episodes.get("released", 0)
if sub_count > 0:
if released_count > 0 and released_count != sub_count:
anime["episodeCount"] = f"{released_count}/{sub_count}"
else:
anime["episodeCount"] = str(sub_count)
else:
anime["episodeCount"] = "?"
return animes
# ======================================================================
# HOME
# ======================================================================
async def home(self) -> Dict[str, Any]:
"""Fetch home page data from Jikan as fallback."""
now = time.time()
if self._home_cache and (now - self._home_cache_ts) < self._home_cache_ttl:
return self._home_cache
await self._ensure_mapping()
try:
# Fire requests concurrently (within rate-limit)
airing_task = self._jikan_get("top/anime", {"filter": "airing", "limit": 24})
popular_task = self._jikan_get("top/anime", {"filter": "bypopularity", "limit": 24, "type": "tv"})
seasonal_task = self._jikan_get("seasons/now", {"limit": 24})
airing_resp, popular_resp, seasonal_resp = await asyncio.gather(
airing_task, popular_task, seasonal_task,
return_exceptions=True,
)
def safe_data(resp):
if isinstance(resp, Exception) or not resp:
return []
return resp.get("data", [])
def filter_adult(items):
return [
item for item in items
if not (item.get("rating") or "").startswith("Rx")
and "Hentai" not in [g.get("name") for g in (item.get("genres") or [])]
]
def is_valid(anime):
name = anime.get("name", "")
if not name or name == "Unknown":
return False
# Must have an AniList ID to be useful for navigation
if not anime.get("anilistId") and not anime.get("id"):
return False
return True
airing_items = filter_adult(safe_data(airing_resp))
popular_items = filter_adult(safe_data(popular_resp))
seasonal_items = filter_adult(safe_data(seasonal_resp))
# Spotlight = top 10 airing
spotlight = [
a for a in (
self._normalize_jikan_spotlight(item, i + 1)
for i, item in enumerate(airing_items[:10])
) if is_valid(a)
]
# Trending = airing
trending = [
a for a in (
self._normalize_jikan_anime(item, i + 1)
for i, item in enumerate(airing_items)
) if is_valid(a)
]
# Popular
popular = [
a for a in (
self._normalize_jikan_anime(item, i + 1)
for i, item in enumerate(popular_items)
) if is_valid(a)
]
# Latest = seasonal
latest = [
a for a in (
self._normalize_jikan_anime(item, i + 1)
for i, item in enumerate(seasonal_items)
) if is_valid(a)
]
normalized = {
"spotlightAnimes": self._annotate_episodes_count(spotlight),
"trendingAnimes": self._annotate_episodes_count(trending),
"mostPopularAnimes": self._annotate_episodes_count(popular),
"latestEpisodeAnimes": self._annotate_episodes_count(latest),
}
result = {
"success": True,
"data": normalized,
"counts": {k: len(v) for k, v in normalized.items()},
}
self._home_cache = result
self._home_cache_ts = time.time()
logger.debug(
f"[MalFallback] Home fetched: spotlight={len(spotlight)}, "
f"trending={len(trending)}, popular={len(popular)}, latest={len(latest)}"
)
return result
except Exception as e:
logger.error(f"[MalFallback] Home fetch error: {e}")
if self._home_cache:
return self._home_cache
return {
"success": False,
"data": {
"spotlightAnimes": [],
"trendingAnimes": [],
"mostPopularAnimes": [],
"latestEpisodeAnimes": [],
},
"counts": {},
}
# ======================================================================
# ANIME INFO
# ======================================================================
async def get_anime_info_by_anilist_id(self, anilist_id: int) -> Dict[str, Any]:
"""
Fetch anime info from Jikan using an AniList ID.
Requires the mapping file to convert AniList → MAL ID first.
"""
await self._ensure_mapping()
mal_id = self.anilist_to_mal(anilist_id)
if not mal_id:
logger.debug(
f"[MalFallback] No AniList → MAL mapping for ID {anilist_id}. Treating as MAL ID directly."
)
mal_id = anilist_id
return await self.get_anime_info_by_mal_id(mal_id)
async def get_anime_info_by_mal_id(self, mal_id: int) -> Dict[str, Any]:
"""Fetch full anime info from Jikan by MAL ID."""
await self._ensure_mapping()
# Fetch full details, characters, and recommendations in parallel
full_task = self._jikan_get(f"anime/{mal_id}/full")
chars_task = self._jikan_get(f"anime/{mal_id}/characters")
recs_task = self._jikan_get(f"anime/{mal_id}/recommendations")
full_resp, chars_resp, recs_resp = await asyncio.gather(
full_task, chars_task, recs_task,
return_exceptions=True
)
if isinstance(full_resp, Exception) or not full_resp or not full_resp.get("data"):
return {}
# Safely extract lists
chars_list = []
if not isinstance(chars_resp, Exception) and chars_resp and isinstance(chars_resp.get("data"), list):
chars_list = chars_resp["data"]
recs_list = []
if not isinstance(recs_resp, Exception) and recs_resp and isinstance(recs_resp.get("data"), list):
recs_list = recs_resp["data"]
info = self._normalize_jikan_info(full_resp["data"], chars_list, recs_list)
logger.debug(
f"[MalFallback] Anime info fetched for MAL ID {mal_id} → "
f"title={info.get('title')}"
)
return info
async def search(self, q: str, page: int = 1, **kwargs) -> Dict[str, Any]:
"""Search anime using Jikan fallback."""
await self._ensure_mapping()
# Call Jikan search
params = {"q": q, "page": page, "limit": 24}
resp = await self._jikan_get("anime", params=params)
if not resp or not resp.get("data"):
return {
"animes": [],
"mostPopularAnimes": [],
"totalPages": 1,
"hasNextPage": False,
"currentPage": page,
"searchQuery": q,
}
data = resp.get("data", [])
pagination = resp.get("pagination", {}) or {}
has_next = pagination.get("has_next_page", False)
# Filter adult and hentai
def filter_adult(items):
return [
item for item in items
if not (item.get("rating") or "").startswith("Rx")
and "Hentai" not in [g.get("name") for g in (item.get("genres") or [])]
]
filtered = filter_adult(data)
def is_valid(anime):
name = anime.get("name", "")
if not name or name == "Unknown":
return False
if not anime.get("anilistId") and not anime.get("id"):
return False
return True
animes = [
a for a in (
self._normalize_jikan_anime(item, i + 1)
for i, item in enumerate(filtered)
) if is_valid(a)
]
# Build standard output
total_pages = pagination.get("last_visible_page", page)
return {
"animes": animes,
"mostPopularAnimes": [],
"totalPages": total_pages,
"hasNextPage": has_next,
"currentPage": page,
"searchQuery": q,
}
async def search_suggestions(self, q: str) -> Dict[str, Any]:
"""Search autocomplete suggestions using Jikan fallback."""
await self._ensure_mapping()
params = {"q": q, "page": 1, "limit": 10}
resp = await self._jikan_get("anime", params=params)
if not resp or not resp.get("data"):
return {"suggestions": []}
data = resp.get("data", [])
def filter_adult(items):
return [
item for item in items
if not (item.get("rating") or "").startswith("Rx")
and "Hentai" not in [g.get("name") for g in (item.get("genres") or [])]
]
filtered = filter_adult(data)
suggestions = []
for item in filtered:
mal_id = item.get("mal_id")
anilist_id = self.mal_to_anilist(mal_id) if mal_id else None
display_id = str(anilist_id) if anilist_id else str(mal_id or "")
english_title = item.get("title_english") or item.get("title") or ""
if not english_title or english_title == "Unknown":
continue
images = item.get("images", {})
poster = (
images.get("webp", {}).get("image_url")
or images.get("jpg", {}).get("image_url")
or ""
)
more_info = []
fmt = item.get("type")
if fmt:
more_info.append(fmt)
eps = item.get("episodes")
status = item.get("status") or ""
if eps:
more_info.append(f"Ep {eps}")
elif "Not yet aired" in status:
more_info.append("Upcoming")
elif "Currently Airing" in status:
more_info.append("Airing")
year = item.get("year")
if year:
more_info.append(str(year))
suggestions.append({
"id": display_id,
"anilistId": anilist_id,
"name": english_title,
"jname": item.get("title_japanese") or "",
"poster": poster,
"moreInfo": more_info,
})
return {"suggestions": suggestions}
async def az_list(self, sort_option: str = "all", page: int = 1) -> Dict[str, Any]:
"""A-Z anime list using Jikan fallback."""
await self._ensure_mapping()
# Jikan search ordered by title asc
params = {"order_by": "title", "sort": "asc", "page": page, "limit": 24}
resp = await self._jikan_get("anime", params=params)
if not resp or not resp.get("data"):
return {
"animes": [],
"totalPages": 1,
"hasNextPage": False,
"currentPage": page,
}
data = resp.get("data", [])
pagination = resp.get("pagination", {}) or {}
has_next = pagination.get("has_next_page", False)
def filter_adult(items):
return [
item for item in items
if not (item.get("rating") or "").startswith("Rx")
and "Hentai" not in [g.get("name") for g in (item.get("genres") or [])]
]
filtered = filter_adult(data)
def is_valid(anime):
name = anime.get("name", "")
if not name or name == "Unknown":
return False
if not anime.get("anilistId") and not anime.get("id"):
return False
return True
animes = [
a for a in (
self._normalize_jikan_anime(item, i + 1)
for i, item in enumerate(filtered)
) if is_valid(a)
]
total_pages = pagination.get("last_visible_page", page)
return {
"animes": animes,
"totalPages": total_pages,
"hasNextPage": has_next,
"currentPage": page,
}