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

Show AllManga alternative watchlist titles

Browse Source
This commit is contained in:
Dymas
2026-05-23 18:34:24 +02:00
parent c6ba7fdbe2
commit 84c2e533d0
7 changed files with 204 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+145 -3
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import base64
import gzip
import html
import json
import mimetypes
import os
@@ -146,6 +147,95 @@ def anime_source_url(show_id, title):
return f"{ALLMANGA_SITE_BASE}/bangumi/{quote(normalized_show_id, safe='')}"
def unique_title_values(values, *exclude_values):
seen = {
normalize_title_key(value)
for value in exclude_values
if normalize_title_key(value)
}
deduped = []
for value in values:
text = str(value or "").strip()
normalized = normalize_title_key(text)
if not text or not normalized or normalized in seen:
continue
seen.add(normalized)
deduped.append(text)
return deduped
def parse_allmanga_alternative_titles(html_text, primary_title=""):
text = str(html_text or "")
primary = str(primary_title or "").strip()
candidates = []
preferred_english = ""
def add_candidate(value, english_preferred=False):
nonlocal preferred_english
cleaned = html.unescape(str(value or "")).replace('\\"', '"').strip()
if not cleaned:
return
candidates.append(cleaned)
if english_preferred and not preferred_english:
preferred_english = cleaned
quoted_key_patterns = (
(r'"english(?:Title|Name)?"\s*:\s*"([^"]+)"', True),
(r'"(?:altEnglish|alternativeEnglish(?:Title|Name)?)"\s*:\s*"([^"]+)"', True),
(r'"romaji(?:Title|Name)?"\s*:\s*"([^"]+)"', False),
(r'"native(?:Title|Name)?"\s*:\s*"([^"]+)"', False),
)
for pattern, english_preferred in quoted_key_patterns:
for match in re.findall(pattern, text, flags=re.IGNORECASE):
add_candidate(match, english_preferred=english_preferred)
array_key_patterns = (
(r'"synonyms"\s*:\s*\[(.*?)\]', False),
(r'"alternative(?:Names|Titles)?"\s*:\s*\[(.*?)\]', False),
(r'"alt(?:Titles|Names)?"\s*:\s*\[(.*?)\]', False),
)
for pattern, english_preferred in array_key_patterns:
for body in re.findall(pattern, text, flags=re.IGNORECASE | re.DOTALL):
for match in re.findall(r'"([^"]+)"', body):
add_candidate(match, english_preferred=english_preferred)
deduped = unique_title_values(candidates, primary)
english_title = ""
for candidate in unique_title_values([preferred_english], primary):
english_title = candidate
break
if not english_title and deduped:
english_title = deduped[0]
return {
"english_title": english_title,
"alt_titles": deduped,
}
def fetch_allmanga_alternative_titles(show_id, title="", timeout=25):
url = anime_source_url(show_id, title)
request = urllib.request.Request(
url,
headers={
"Accept": "text/html,application/xhtml+xml",
"Referer": ALLANIME_REFERER,
"User-Agent": AGENT,
},
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
raw = response.read().decode("utf-8", errors="replace")
except urllib.error.URLError as exc:
raise RuntimeError(f"Could not reach AllManga page: {exc}") from exc
except TimeoutError as exc:
raise RuntimeError("AllManga page request timed out") from exc
parsed = parse_allmanga_alternative_titles(raw, primary_title=title)
if not parsed["alt_titles"] and not parsed["english_title"]:
raise RuntimeError("AllManga page did not expose any alternative titles")
return parsed
def infer_image_extension(url, content_type):
normalized_type = str(content_type or "").split(";", 1)[0].strip().lower()
extension = mimetypes.guess_extension(normalized_type, strict=False)
@@ -1028,6 +1118,8 @@ class WatchlistStore:
dub_latest_episode TEXT,
animeschedule_route TEXT,
animeschedule_title TEXT,
english_title TEXT,
alt_titles_json TEXT NOT NULL DEFAULT '[]',
anidb_aid TEXT,
anidb_title TEXT,
thumbnail_path TEXT,
@@ -1056,6 +1148,10 @@ class WatchlistStore:
conn.execute("ALTER TABLE watchlist ADD COLUMN animeschedule_route TEXT")
if "animeschedule_title" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN animeschedule_title TEXT")
if "english_title" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN english_title TEXT")
if "alt_titles_json" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN alt_titles_json TEXT NOT NULL DEFAULT '[]'")
conn.execute(
"UPDATE watchlist SET category = 'watching' WHERE category IS NULL OR TRIM(category) = ''"
)
@@ -1077,6 +1173,22 @@ class WatchlistStore:
item["dub_progress"] = format_episode_progress(item["dub_count"], item["expected_count"])
item["sub_complete"] = mode_is_complete(item["sub_count"], item["expected_count"], item["airing_status"])
item["dub_complete"] = mode_is_complete(item["dub_count"], item["expected_count"], item["airing_status"])
try:
alt_titles = json.loads(item.get("alt_titles_json") or "[]")
except json.JSONDecodeError:
alt_titles = []
if not isinstance(alt_titles, list):
alt_titles = []
item["alt_titles"] = unique_title_values(alt_titles, item.get("title"))
item["english_title"] = next(iter(unique_title_values([item.get("english_title")], item.get("title"))), "")
display_alt_titles = []
if item["english_title"]:
display_alt_titles.append(item["english_title"])
for candidate in item["alt_titles"]:
if normalize_title_key(candidate) == normalize_title_key(item["english_title"]):
continue
display_alt_titles.append(candidate)
item["display_alt_titles"] = display_alt_titles[:3]
item["finished_badge"] = (
"downloaded" if item["category"] == "finished" and item["downloaded"] else "manual" if item["category"] == "finished" else ""
)
@@ -1097,15 +1209,19 @@ class WatchlistStore:
return item
def _upsert_conn(self, conn, item):
raw_alt_titles = item.get("alt_titles") if isinstance(item.get("alt_titles"), list) else []
if not raw_alt_titles and isinstance(item.get("alt_titles_json"), list):
raw_alt_titles = item.get("alt_titles_json")
normalized_alt_titles = unique_title_values(raw_alt_titles, item.get("title"))
conn.execute(
"""
INSERT INTO watchlist (
show_id, title, category, downloaded, status, status_message, created_at, updated_at,
last_checked, sub_count, dub_count, expected_count, airing_status,
sub_latest_episode, dub_latest_episode,
animeschedule_route, animeschedule_title, anidb_aid, anidb_title,
animeschedule_route, animeschedule_title, english_title, alt_titles_json, anidb_aid, anidb_title,
thumbnail_path, thumbnail_checked_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(show_id) DO UPDATE SET
title = excluded.title,
category = excluded.category,
@@ -1122,6 +1238,8 @@ class WatchlistStore:
dub_latest_episode = excluded.dub_latest_episode,
animeschedule_route = excluded.animeschedule_route,
animeschedule_title = excluded.animeschedule_title,
english_title = excluded.english_title,
alt_titles_json = excluded.alt_titles_json,
anidb_aid = excluded.anidb_aid,
anidb_title = excluded.anidb_title,
thumbnail_path = excluded.thumbnail_path,
@@ -1145,6 +1263,8 @@ class WatchlistStore:
item.get("dub_latest_episode"),
item.get("animeschedule_route"),
item.get("animeschedule_title"),
item.get("english_title"),
json.dumps(normalized_alt_titles),
item.get("anidb_aid"),
item.get("anidb_title"),
item.get("thumbnail_path"),
@@ -1190,6 +1310,8 @@ class WatchlistStore:
"dub_latest_episode": None,
"animeschedule_route": None,
"animeschedule_title": None,
"english_title": None,
"alt_titles": [],
"anidb_aid": None,
"anidb_title": None,
"thumbnail_path": None,
@@ -1315,6 +1437,8 @@ class WatchlistStore:
"dub_latest_episode": None,
"animeschedule_route": None,
"animeschedule_title": None,
"english_title": None,
"alt_titles": [],
"anidb_aid": None,
"anidb_title": None,
"thumbnail_path": None,
@@ -1341,6 +1465,10 @@ class WatchlistStore:
sub_episodes = snapshot["episodes"]["sub"]
dub_episodes = snapshot["episodes"]["dub"]
schedule = None
allmanga_titles = {
"english_title": str(existing.get("english_title") or "").strip(),
"alt_titles": list(existing.get("alt_titles") or []),
}
try:
schedule = resolve_animeschedule_anime(
snapshot["title"] or existing["title"],
@@ -1349,6 +1477,14 @@ class WatchlistStore:
)
except Exception:
schedule = None
try:
allmanga_titles = fetch_allmanga_alternative_titles(
normalized_show_id,
snapshot["title"] or existing["title"],
timeout=request_timeout,
)
except Exception as exc:
debug_log("watchlist.alt_titles.lookup_failed", show_id=normalized_show_id, error=exc)
if self.refresh_stop_event.is_set():
return None
if not self.has_show(normalized_show_id):
@@ -1374,6 +1510,8 @@ 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"),
allmanga_titles.get("english_title") or existing.get("english_title"),
json.dumps(unique_title_values(allmanga_titles.get("alt_titles") or [], snapshot["title"] or existing["title"])),
normalized_show_id,
)
except Exception as exc:
@@ -1395,6 +1533,8 @@ class WatchlistStore:
existing.get("dub_latest_episode"),
existing.get("animeschedule_route"),
existing.get("animeschedule_title"),
existing.get("english_title"),
existing.get("alt_titles_json") or "[]",
normalized_show_id,
)
@@ -1417,7 +1557,9 @@ class WatchlistStore:
sub_latest_episode = ?,
dub_latest_episode = ?,
animeschedule_route = ?,
animeschedule_title = ?
animeschedule_title = ?,
english_title = ?,
alt_titles_json = ?
WHERE show_id = ?
""",
payload,