Dymas/ani-cli-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use AniDB for watchlist alternative titles

Browse Source
This commit is contained in:
Dymas
2026-05-25 09:24:57 +02:00
parent 64979d5d32
commit 46af0ff47d
7 changed files with 79 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+50 -213
View File
@@ -25,7 +25,6 @@ from app_support import (
ALLANIME_API,
ALLANIME_REFERER,
ALLMANGA_SITE_BASE,
ANILIST_API,
ANIDB_ANIME_PAGE,
ANIDB_TITLES_CACHE,
ANIDB_TITLES_PATH,
@@ -231,123 +230,6 @@ def animeschedule_request(path, params=None, timeout=25):
raise RuntimeError("AnimeSchedule returned unreadable JSON") from exc
def anilist_request(query, variables, timeout=25):
payload = json.dumps({"variables": variables, "query": query}).encode("utf-8")
request = urllib.request.Request(
ANILIST_API,
data=payload,
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": AGENT,
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
raw = response.read().decode("utf-8")
except urllib.error.URLError as exc:
raise RuntimeError(f"Could not reach AniList: {exc}") from exc
except TimeoutError as exc:
raise RuntimeError("AniList request timed out") from exc
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise RuntimeError("AniList returned unreadable JSON") from exc
if data.get("errors"):
message = str(((data.get("errors") or [{}])[0]).get("message") or "AniList returned an error")
raise RuntimeError(message)
return data.get("data", {})
def normalize_anilist_entry(node):
if not isinstance(node, dict):
return None
media_id = node.get("id")
if media_id in {None, ""}:
return None
title = node.get("title") if isinstance(node.get("title"), dict) else {}
synonyms = node.get("synonyms") if isinstance(node.get("synonyms"), list) else []
return {
"id": int(media_id),
"title": {
"romaji": str(title.get("romaji") or "").strip(),
"english": str(title.get("english") or "").strip(),
"native": str(title.get("native") or "").strip(),
"userPreferred": str(title.get("userPreferred") or "").strip(),
},
"synonyms": [str(value or "").strip() for value in synonyms if str(value or "").strip()],
"siteUrl": str(node.get("siteUrl") or "").strip(),
}
def anilist_title_candidates(entry):
if not isinstance(entry, dict):
return []
values = []
title = entry.get("title") if isinstance(entry.get("title"), dict) else {}
for key in ("romaji", "english", "native", "userPreferred"):
text = str(title.get(key) or "").strip()
if text:
values.append(text)
for value in entry.get("synonyms") or []:
text = str(value or "").strip()
if text:
values.append(text)
deduped = []
seen = set()
for value in values:
normalized = normalize_title_key(value)
if normalized and normalized not in seen:
seen.add(normalized)
deduped.append(value)
return deduped
def score_anilist_match(query_variants, entry):
best = None
for candidate in anilist_title_candidates(entry):
normalized = normalize_title_key(candidate)
if not normalized:
continue
score = None
if normalized in query_variants:
score = 110
else:
for variant in query_variants:
if variant and (normalized in variant or variant in normalized):
score = max(score or 0, 80 - abs(len(normalized) - len(variant)))
if score is not None:
best = max(best or score, score)
return best
def extract_anilist_alternative_titles(entry, primary_title=""):
if not isinstance(entry, dict):
return []
primary_key = normalize_title_key(primary_title) or str(primary_title or "").strip().casefold()
results = []
seen = set()
title = entry.get("title") if isinstance(entry.get("title"), dict) else {}
for key, label in (("romaji", "Romaji"), ("english", "English"), ("native", "Native")):
text = str(title.get(key) or "").strip()
normalized = normalize_title_key(text) or text.casefold()
if not normalized or normalized == primary_key or normalized in seen:
continue
seen.add(normalized)
results.append({"label": label, "value": text})
for synonym in entry.get("synonyms") or []:
text = str(synonym or "").strip()
normalized = normalize_title_key(text) or text.casefold()
if not normalized or normalized == primary_key or normalized in seen:
continue
seen.add(normalized)
results.append({"label": "Synonym", "value": text})
return results
def encode_alternative_titles(alternative_titles):
normalized = []
for item in alternative_titles or []:
@@ -382,83 +264,32 @@ def decode_alternative_titles(payload):
return normalized
def fetch_anilist_media_by_id(media_id, timeout=25):
data = anilist_request(
"""
query ($id: Int) {
Media(id: $id, type: ANIME) {
id
siteUrl
title {
romaji
english
native
userPreferred
}
synonyms
}
}
""",
{"id": int(media_id)},
timeout=timeout,
)
entry = normalize_anilist_entry(data.get("Media"))
if not entry:
raise RuntimeError("AniList media response was missing required fields")
return entry
def resolve_anilist_media(title, media_id=None, timeout=25):
stored_media_id = str(media_id or "").strip()
if stored_media_id:
try:
return fetch_anilist_media_by_id(int(stored_media_id), timeout=timeout)
except Exception:
debug_log("anilist.media_refresh_failed", media_id=stored_media_id, title=title)
query_variants = title_match_variants(title)
if not query_variants:
raise RuntimeError("Missing title for AniList lookup")
queries = title_lookup_queries(title)
best = None
for query in queries:
data = anilist_request(
"""
query ($search: String, $page: Int, $perPage: Int) {
Page(page: $page, perPage: $perPage) {
media(search: $search, type: ANIME) {
id
siteUrl
title {
romaji
english
native
userPreferred
}
synonyms
}
}
}
""",
{"search": query, "page": 1, "perPage": 8},
timeout=timeout,
)
for node in ((data.get("Page") or {}).get("media") or []):
entry = normalize_anilist_entry(node)
if not entry:
continue
score = score_anilist_match(query_variants, entry)
if score is None:
continue
if best is None or score > best["score"]:
best = {"entry": entry, "score": score}
if best is not None:
break
if best is None:
raise RuntimeError("AniList lookup did not find a matching anime")
return best["entry"]
def extract_anidb_alternative_titles(aid, primary_title=""):
normalized_aid = str(aid or "").strip()
if not normalized_aid:
return []
primary_key = normalize_title_key(primary_title) or str(primary_title or "").strip().casefold()
results = []
seen = set()
type_labels = {
"official": "English",
"main": "English",
"short": "English",
"syn": "English",
}
for entry in get_cached_anidb_titles():
if str(entry.get("aid") or "").strip() != normalized_aid:
continue
if str(entry.get("lang") or "").strip().lower() != "en":
continue
text = str(entry.get("title") or "").strip()
normalized = normalize_title_key(text) or text.casefold()
if not normalized or normalized == primary_key or normalized in seen:
continue
seen.add(normalized)
label = type_labels.get(str(entry.get("type") or "").strip().lower(), "English")
results.append({"label": label, "value": text})
return results
def parse_nonnegative_int(value, default=0):
@@ -1612,22 +1443,27 @@ class WatchlistStore:
)
except Exception:
schedule = None
anilist = None
anidb_match = None
try:
anilist = resolve_anilist_media(
snapshot["title"] or existing["title"],
media_id=existing.get("anilist_id"),
timeout=request_timeout,
)
if existing.get("anidb_aid"):
anidb_match = {
"aid": str(existing.get("anidb_aid")).strip(),
"title": str(existing.get("anidb_title") or snapshot["title"] or existing["title"]).strip(),
}
else:
anidb_match = resolve_anidb_aid(snapshot["title"] or existing["title"])
except Exception:
anilist = None
anidb_match = None
if self.refresh_stop_event.is_set():
return None
if not self.has_show(normalized_show_id):
return None
expected_count = parse_nonnegative_int((schedule or {}).get("episodes")) or int(existing.get("expected_count") or 0)
airing_status = normalize_animeschedule_status((schedule or {}).get("status") or existing.get("airing_status"))
alternative_titles = extract_anilist_alternative_titles(anilist, primary_title=snapshot["title"] or existing["title"])
alternative_titles = extract_anidb_alternative_titles(
(anidb_match or {}).get("aid"),
primary_title=snapshot["title"] or existing["title"],
)
payload = (
snapshot["title"] or existing["title"],
"updated",
@@ -1647,15 +1483,12 @@ class WatchlistStore:
dub_episodes[-1] if dub_episodes else None,
(schedule or {}).get("route") or existing.get("animeschedule_route"),
(schedule or {}).get("title") or existing.get("animeschedule_title"),
(anilist or {}).get("id") or existing.get("anilist_id"),
(
((anilist or {}).get("title") or {}).get("userPreferred")
or ((anilist or {}).get("title") or {}).get("romaji")
or ((anilist or {}).get("title") or {}).get("english")
or existing.get("anilist_title")
),
(anilist or {}).get("siteUrl") or existing.get("anilist_site_url"),
encode_alternative_titles(alternative_titles) if anilist is not None else existing.get("alternative_titles_json"),
existing.get("anilist_id"),
existing.get("anilist_title"),
existing.get("anilist_site_url"),
encode_alternative_titles(alternative_titles) if anidb_match is not None else existing.get("alternative_titles_json"),
(anidb_match or {}).get("aid") or existing.get("anidb_aid"),
(anidb_match or {}).get("title") or existing.get("anidb_title"),
normalized_show_id,
)
except Exception as exc:
@@ -1681,6 +1514,8 @@ class WatchlistStore:
existing.get("anilist_title"),
existing.get("anilist_site_url"),
existing.get("alternative_titles_json"),
existing.get("anidb_aid"),
existing.get("anidb_title"),
normalized_show_id,
)
@@ -1707,7 +1542,9 @@ class WatchlistStore:
anilist_id = ?,
anilist_title = ?,
anilist_site_url = ?,
alternative_titles_json = ?
alternative_titles_json = ?,
anidb_aid = ?,
anidb_title = ?
WHERE show_id = ?
""",
payload,