init
This commit is contained in:
@@ -0,0 +1,231 @@
|
||||
"""
|
||||
mal_resolver.py
|
||||
===============
|
||||
|
||||
Fetches and caches MyAnimeList manga metadata (statistics and characters)
|
||||
using the public Jikan REST API v4.
|
||||
|
||||
Jikan API: https://api.jikan.moe/v4 (no authentication required)
|
||||
Rate limit: 3 req/s, 60 req/min -> a 400 ms delay between calls is applied.
|
||||
|
||||
Provided features
|
||||
-----------------
|
||||
- Title-based MAL ID lookup with best-match scoring (cached)
|
||||
- MAL statistics: score, rank, scored_by, popularity, members, favorites
|
||||
- Character list for a manga (names only, cached)
|
||||
- Convenience: get_characters_for_manga(title) -> list[str]
|
||||
|
||||
Dependencies
|
||||
------------
|
||||
requests -> pip install requests
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import difflib
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
class MALResolver:
|
||||
"""
|
||||
Fetches and caches MyAnimeList manga data via the Jikan API v4.
|
||||
"""
|
||||
|
||||
JIKAN_BASE = "https://api.jikan.moe/v4"
|
||||
|
||||
def __init__(self, *,
|
||||
request_timeout: int = 30,
|
||||
session: "requests.Session | None" = None):
|
||||
self.request_timeout = request_timeout
|
||||
self._session = session or requests.Session()
|
||||
self._session.headers.setdefault("User-Agent", "MALResolver/1.0")
|
||||
|
||||
self._id_cache: dict[str, "int | None"] = {} # title_lower -> mal_id
|
||||
self._stats_cache: dict[int, dict] = {} # mal_id -> stats dict
|
||||
self._char_cache: dict[int, list[str]] = {} # mal_id -> [name, ...]
|
||||
|
||||
self._last_request_at: float = 0.0
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public: ID lookup
|
||||
# ------------------------------------------------------------------
|
||||
def find_mal_id(self, title: str) -> "int | None":
|
||||
"""
|
||||
Searches MAL for a manga by title and returns the best-matching MAL ID.
|
||||
Returns None on failure or when no result is found.
|
||||
"""
|
||||
if not title or not title.strip():
|
||||
return None
|
||||
|
||||
key = title.strip().lower()
|
||||
if key in self._id_cache:
|
||||
return self._id_cache[key]
|
||||
|
||||
try:
|
||||
data = self._get(f"{self.JIKAN_BASE}/manga",
|
||||
{"q": title, "limit": 5, "type": "manga"})
|
||||
results = data.get("data") or []
|
||||
except requests.RequestException:
|
||||
return None
|
||||
|
||||
if not results:
|
||||
self._id_cache[key] = None
|
||||
return None
|
||||
|
||||
results.sort(key=lambda e: _score_title(title, e), reverse=True)
|
||||
mal_id = results[0].get("mal_id")
|
||||
self._id_cache[key] = mal_id
|
||||
return mal_id
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public: statistics
|
||||
# ------------------------------------------------------------------
|
||||
def get_stats(self, mal_id: "int | None") -> "dict | None":
|
||||
"""
|
||||
Returns a statistics dict for the given MAL manga ID:
|
||||
|
||||
{
|
||||
"score": float | None,
|
||||
"rank": int | None,
|
||||
"scored_by": int | None,
|
||||
"popularity": int | None,
|
||||
"members": int | None,
|
||||
"favorites": int | None,
|
||||
"url": str,
|
||||
"title": str,
|
||||
"as_of": str (DD-MM-YYYY),
|
||||
}
|
||||
|
||||
Returns None if mal_id is None or on network failure.
|
||||
"""
|
||||
if mal_id is None:
|
||||
return None
|
||||
|
||||
if mal_id in self._stats_cache:
|
||||
return self._stats_cache[mal_id]
|
||||
|
||||
try:
|
||||
data = self._get(f"{self.JIKAN_BASE}/manga/{mal_id}")
|
||||
entry = data.get("data") or {}
|
||||
except requests.RequestException:
|
||||
return None
|
||||
|
||||
stats: dict = {
|
||||
"score": entry.get("score"),
|
||||
"rank": entry.get("rank"),
|
||||
"scored_by": entry.get("scored_by"),
|
||||
"popularity": entry.get("popularity"),
|
||||
"members": entry.get("members"),
|
||||
"favorites": entry.get("favorites"),
|
||||
"url": (entry.get("url")
|
||||
or f"https://myanimelist.net/manga/{mal_id}"),
|
||||
"title": entry.get("title") or "",
|
||||
"as_of": datetime.date.today().strftime("%d-%m-%Y"),
|
||||
}
|
||||
self._stats_cache[mal_id] = stats
|
||||
return stats
|
||||
|
||||
def get_stats_for_manga(self, title: str) -> "dict | None":
|
||||
"""Convenience: find MAL ID by title, then return stats."""
|
||||
return self.get_stats(self.find_mal_id(title))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public: characters
|
||||
# ------------------------------------------------------------------
|
||||
def get_characters(self, mal_id: "int | None") -> list[str]:
|
||||
"""
|
||||
Returns a list of character names (strings) for the manga.
|
||||
Returns an empty list on failure.
|
||||
"""
|
||||
if mal_id is None:
|
||||
return []
|
||||
|
||||
if mal_id in self._char_cache:
|
||||
return self._char_cache[mal_id]
|
||||
|
||||
try:
|
||||
data = self._get(f"{self.JIKAN_BASE}/manga/{mal_id}/characters")
|
||||
entries = data.get("data") or []
|
||||
except requests.RequestException:
|
||||
return []
|
||||
|
||||
names = []
|
||||
for entry in entries:
|
||||
char = entry.get("character") or {}
|
||||
name = char.get("name")
|
||||
if name:
|
||||
names.append(name)
|
||||
|
||||
self._char_cache[mal_id] = names
|
||||
return names
|
||||
|
||||
def get_characters_for_manga(self, title: str) -> list[str]:
|
||||
"""
|
||||
Convenience: search for manga by title, then return its characters.
|
||||
"""
|
||||
return self.get_characters(self.find_mal_id(title))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public: cache management
|
||||
# ------------------------------------------------------------------
|
||||
def clear_cache(self) -> None:
|
||||
"""Clears all internal caches."""
|
||||
self._id_cache.clear()
|
||||
self._stats_cache.clear()
|
||||
self._char_cache.clear()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal: rate-limited HTTP
|
||||
# ------------------------------------------------------------------
|
||||
def _get(self, url: str, params: "dict | None" = None) -> dict:
|
||||
"""Rate-limited GET request (respects Jikan's 3 req/s limit)."""
|
||||
elapsed = time.monotonic() - self._last_request_at
|
||||
if elapsed < 0.4:
|
||||
time.sleep(0.4 - elapsed)
|
||||
resp = self._session.get(url, params=params, timeout=self.request_timeout)
|
||||
self._last_request_at = time.monotonic()
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Module helper
|
||||
# --------------------------------------------------------------------------
|
||||
def _score_title(query: str, entry: dict) -> float:
|
||||
"""Returns the best title-similarity score for a Jikan manga entry."""
|
||||
candidates = [
|
||||
entry.get("title") or "",
|
||||
entry.get("title_english") or "",
|
||||
entry.get("title_japanese") or "",
|
||||
]
|
||||
for alt in (entry.get("titles") or []):
|
||||
candidates.append(alt.get("title") or "")
|
||||
|
||||
best = 0.0
|
||||
q = query.lower()
|
||||
for t in candidates:
|
||||
if t:
|
||||
ratio = difflib.SequenceMatcher(None, q, t.lower()).ratio()
|
||||
best = max(best, ratio)
|
||||
return best
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Usage example
|
||||
# --------------------------------------------------------------------------
|
||||
if __name__ == "__main__":
|
||||
resolver = MALResolver()
|
||||
|
||||
mal_id = resolver.find_mal_id("Yofukashi no Uta")
|
||||
print("MAL ID :", mal_id)
|
||||
|
||||
stats = resolver.get_stats(mal_id)
|
||||
if stats:
|
||||
print("Score :", stats["score"])
|
||||
print("Rank :", stats["rank"])
|
||||
|
||||
chars = resolver.get_characters(mal_id)
|
||||
print("Characters (first 5):", chars[:5])
|
||||
Reference in New Issue
Block a user