#!/usr/bin/env python3
import base64
import gzip
import json
import mimetypes
import os
import re
import shutil
import signal
import sqlite3
import subprocess
import threading
import time
import unicodedata
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET
from binascii import Error as BinasciiError
from datetime import datetime, timezone
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import sys
from urllib.parse import parse_qs, quote, unquote, urljoin, urlparse
from uuid import uuid4
PROJECT_ROOT = Path(__file__).resolve().parent
ANI_CLI = os.environ.get("ANI_CLI_BIN") or shutil.which("ani-cli") or "ani-cli"
APP_NAME = "ani-cli-web"
VERSION = "0.8.13"
ALLANIME_BASE = "allanime.day"
ALLANIME_API = f"https://api.{ALLANIME_BASE}"
ALLANIME_REFERER = "https://allmanga.to"
ALLMANGA_SITE_BASE = "https://allmanga.to"
ANIMESCHEDULE_API = "https://animeschedule.net/api/v3"
ANIMESCHEDULE_IMAGE_BASE = "https://img.animeschedule.net/production/assets/public/img/"
ANIDB_TITLES_URL = "https://anidb.net/api/anime-titles.xml.gz"
ANIDB_ANIME_PAGE = "https://anidb.net/anime/{aid}"
AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/121.0"
CLIENT_DISCONNECT_ERRORS = (BrokenPipeError, ConnectionResetError, ConnectionAbortedError)
DEBUG_MODE = False
QUALITY_CHOICES = {"best", "1080", "1080p", "720", "720p", "480", "480p", "360", "360p", "worst"}
MODE_CHOICES = {"sub", "dub"}
MAX_LOG_LINES = 240
def debug_enabled():
return DEBUG_MODE or str(os.environ.get("ANI_CLI_WEB_DEBUG") or "").strip().lower() in {"1", "true", "yes", "on"}
def debug_log(event, **fields):
if not debug_enabled():
return
parts = [f"[debug] {event}"]
for key, value in fields.items():
text = str(value)
if len(text) > 240:
text = text[:237] + "..."
parts.append(f"{key}={text}")
print(" | ".join(parts), flush=True)
def default_download_dir():
if os.environ.get("ANI_CLI_DOWNLOAD_DIR"):
return os.environ["ANI_CLI_DOWNLOAD_DIR"]
home_downloads = Path.home() / "Downloads" / "Anime"
if os.access(home_downloads.parent if home_downloads.parent.exists() else Path.home(), os.W_OK):
return str(home_downloads)
return str(PROJECT_ROOT / "downloads")
DEFAULT_CONFIG = {
"download_dir": default_download_dir(),
"mode": os.environ.get("ANI_CLI_MODE", "sub"),
"quality": os.environ.get("ANI_CLI_QUALITY", "best"),
}
def now_iso():
return datetime.now(timezone.utc).isoformat()
PROJECT_APP_ROOT = PROJECT_ROOT / ".ani-cli-web"
PROJECT_APP_ROOT.mkdir(parents=True, exist_ok=True)
def legacy_state_root():
return Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state")) / APP_NAME
def legacy_config_root():
return Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / APP_NAME
def project_state_path(*parts):
path = PROJECT_APP_ROOT.joinpath(*parts)
path.parent.mkdir(parents=True, exist_ok=True)
return path
def project_state_dir(*parts):
path = PROJECT_APP_ROOT.joinpath(*parts)
path.mkdir(parents=True, exist_ok=True)
return path
def migrate_legacy_file(project_path, legacy_path, label):
if project_path.exists() or not legacy_path.exists():
return project_path
project_path.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.move(str(legacy_path), str(project_path))
debug_log("state.migrated", label=label, from_path=legacy_path, to_path=project_path)
except OSError as exc:
debug_log("state.migrate_failed", label=label, from_path=legacy_path, to_path=project_path, error=exc)
return project_path
def migrate_legacy_dir(project_dir, legacy_dir, label):
if not legacy_dir.exists():
return project_dir
project_dir.mkdir(parents=True, exist_ok=True)
moved = 0
for child in legacy_dir.iterdir():
target = project_dir / child.name
if target.exists():
if child.is_dir() and target.is_dir():
migrate_legacy_dir(target, child, f"{label}/{child.name}")
continue
try:
shutil.move(str(child), str(target))
moved += 1
except OSError as exc:
debug_log("state.migrate_failed", label=label, from_path=child, to_path=target, error=exc)
try:
legacy_dir.rmdir()
except OSError:
pass
if moved:
debug_log("state.migrated", label=label, from_path=legacy_dir, to_path=project_dir, entries=moved)
return project_dir
LEGACY_STATE_ROOT = legacy_state_root()
LEGACY_CONFIG_ROOT = legacy_config_root()
LEGACY_CONFIG_PATH = LEGACY_CONFIG_ROOT / "config.json"
LEGACY_QUEUE_PATH = LEGACY_STATE_ROOT / "queue.json"
LEGACY_QUEUE_DB_PATH = LEGACY_STATE_ROOT / "queue.sqlite3"
LEGACY_STAGING_ROOT = LEGACY_STATE_ROOT / "staging"
LEGACY_WATCHLIST_JSON_PATH = LEGACY_STATE_ROOT / "watchlist.json"
LEGACY_THUMBNAIL_ROOT = LEGACY_STATE_ROOT / "thumbnails"
LEGACY_ANIDB_TITLES_PATH = LEGACY_STATE_ROOT / "anidb-anime-titles.xml.gz"
CONFIG_PATH = project_state_path("config.json")
QUEUE_PATH = project_state_path("queue.json")
QUEUE_DB_PATH = project_state_path("queue.sqlite3")
STAGING_ROOT = project_state_dir("staging")
WATCHLIST_JSON_PATH = project_state_path("watchlist.json")
THUMBNAIL_ROOT = project_state_dir("thumbnails")
THUMBNAIL_RETRY_SECONDS = 15 * 60
ANIDB_TITLES_PATH = project_state_path("anidb-anime-titles.xml.gz")
ANIDB_TITLES_CACHE = {"mtime": None, "entries": None}
def migrate_legacy_state_storage():
migrate_legacy_file(CONFIG_PATH, LEGACY_CONFIG_PATH, "config.json")
migrate_legacy_file(QUEUE_PATH, LEGACY_QUEUE_PATH, "queue.json")
migrate_legacy_file(QUEUE_DB_PATH, LEGACY_QUEUE_DB_PATH, "queue.sqlite3")
migrate_legacy_file(WATCHLIST_JSON_PATH, LEGACY_WATCHLIST_JSON_PATH, "watchlist.json")
migrate_legacy_file(ANIDB_TITLES_PATH, LEGACY_ANIDB_TITLES_PATH, "anidb-anime-titles.xml.gz")
migrate_legacy_dir(STAGING_ROOT, LEGACY_STAGING_ROOT, "staging")
migrate_legacy_dir(THUMBNAIL_ROOT, LEGACY_THUMBNAIL_ROOT, "thumbnails")
migrate_legacy_state_storage()
def load_json(path, fallback):
try:
with path.open("r", encoding="utf-8") as handle:
data = json.load(handle)
return data if isinstance(data, type(fallback)) else fallback
except (FileNotFoundError, json.JSONDecodeError, OSError):
return fallback
def write_json(path, data):
tmp = path.with_suffix(path.suffix + ".tmp")
with tmp.open("w", encoding="utf-8") as handle:
json.dump(data, handle, indent=2, sort_keys=True)
tmp.replace(path)
def normalize_config(data):
config = dict(DEFAULT_CONFIG)
if isinstance(data, dict):
config.update({key: data[key] for key in DEFAULT_CONFIG if key in data})
mode = str(config.get("mode", "sub")).lower()
quality = str(config.get("quality", "best")).lower()
if quality.endswith("p") and quality[:-1].isdigit():
quality = quality[:-1]
download_dir = str(config.get("download_dir") or DEFAULT_CONFIG["download_dir"])
config["mode"] = mode if mode in MODE_CHOICES else "sub"
config["quality"] = quality if quality in QUALITY_CHOICES else "best"
config["download_dir"] = str(Path(download_dir).expanduser())
return config
CONFIG = normalize_config(load_json(CONFIG_PATH, DEFAULT_CONFIG))
write_json(CONFIG_PATH, CONFIG)
def graph_request(query, variables):
payload = json.dumps({"variables": variables, "query": query}).encode("utf-8")
request = urllib.request.Request(
f"{ALLANIME_API}/api",
data=payload,
headers={
"Content-Type": "application/json",
"Referer": ALLANIME_REFERER,
"User-Agent": AGENT,
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=25) as response:
raw = response.read().decode("utf-8")
except urllib.error.URLError as exc:
raise RuntimeError(f"Could not reach AllAnime: {exc}") from exc
except TimeoutError as exc:
raise RuntimeError("AllAnime request timed out") from exc
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise RuntimeError("AllAnime returned an unreadable response") from exc
if data.get("errors"):
message = data["errors"][0].get("message", "AllAnime returned an error")
raise RuntimeError(message)
return data.get("data", {})
def parse_iso_timestamp(value):
text = str(value or "").strip()
if not text:
return None
try:
return datetime.fromisoformat(text.replace("Z", "+00:00"))
except ValueError:
return None
def thumbnail_retry_due(checked_at):
timestamp = parse_iso_timestamp(checked_at)
if not timestamp:
return True
age = datetime.now(timezone.utc) - timestamp.astimezone(timezone.utc)
return age.total_seconds() >= THUMBNAIL_RETRY_SECONDS
def thumbnail_file_path(name):
filename = Path(str(name or "")).name
if not filename:
return None
project_path = THUMBNAIL_ROOT / filename
if project_path.exists():
return project_path
legacy_path = LEGACY_THUMBNAIL_ROOT / filename
if legacy_path.exists():
THUMBNAIL_ROOT.mkdir(parents=True, exist_ok=True)
try:
shutil.move(str(legacy_path), str(project_path))
debug_log("thumbnail.cache.migrated", from_path=legacy_path, to_path=project_path)
return project_path
except OSError as exc:
debug_log("thumbnail.cache.migrate_failed", from_path=legacy_path, to_path=project_path, error=exc)
return legacy_path
return project_path
def cover_route(show_id):
return f"/api/watchlist/thumb/{quote(str(show_id).strip(), safe='')}"
def anime_source_url(show_id, title):
normalized_show_id = str(show_id or "").strip()
if not normalized_show_id:
return f"{ALLMANGA_SITE_BASE}/search-anime?tr=sub&cty=ALL&query={quote(str(title or '').strip(), safe='')}"
return f"{ALLMANGA_SITE_BASE}/bangumi/{quote(normalized_show_id, safe='')}"
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)
if extension == ".jpe":
extension = ".jpg"
if extension:
return extension
suffix = Path(urlparse(str(url or "")).path).suffix.lower()
if suffix in {".jpg", ".jpeg", ".png", ".webp", ".gif"}:
return suffix
return ".jpg"
def save_thumbnail_bytes(show_id, source_name, content_type, content):
if not content:
raise RuntimeError("Thumbnail content was empty")
THUMBNAIL_ROOT.mkdir(parents=True, exist_ok=True)
stem = sanitize_path_component(show_id, "anime")
extension = infer_image_extension(source_name, content_type)
final_path = THUMBNAIL_ROOT / f"{stem}{extension}"
tmp_path = final_path.with_suffix(final_path.suffix + ".tmp")
with tmp_path.open("wb") as handle:
handle.write(content)
tmp_path.replace(final_path)
for existing in THUMBNAIL_ROOT.glob(f"{stem}.*"):
if existing != final_path and existing.is_file():
existing.unlink(missing_ok=True)
debug_log("thumbnail.cache.write", show_id=show_id, path=final_path, bytes=len(content), content_type=content_type)
return final_path.name
def decode_image_data_url(data_url):
text = str(data_url or "").strip()
match = re.match(r"^data:(image/[-a-zA-Z0-9.+]+);base64,(.+)$", text, re.DOTALL)
if not match:
raise ValueError("Thumbnail upload must be a base64 image data URL.")
content_type = match.group(1).strip().lower()
try:
content = base64.b64decode(match.group(2), validate=True)
except (ValueError, BinasciiError) as exc:
raise ValueError("Thumbnail upload data was not valid base64.") from exc
if len(content) > 8 * 1024 * 1024:
raise ValueError("Thumbnail upload is too large. Keep it under 8 MB.")
return content_type, content
def normalize_title_key(value):
text = unicodedata.normalize("NFKD", str(value or ""))
text = "".join(char for char in text if not unicodedata.combining(char))
text = text.lower()
text = re.sub(r"\((19|20)\d{2}\)", " ", text)
text = re.sub(r"[^a-z0-9]+", " ", text)
return re.sub(r"\s+", " ", text).strip()
def base_title_variants(value):
text = str(value or "").strip()
variants = []
if text:
variants.append(text)
patterns = [
r"\s+\((19|20)\d{2}\)\s*$",
r"\s+season\s+\d+\s*$",
r"\s+part\s+\d+\s*$",
r"\s+cour\s+\d+\s*$",
r"\s+\d+(st|nd|rd|th)\s+season\s*$",
]
changed = True
current = text
while current and changed:
changed = False
for pattern in patterns:
trimmed = re.sub(pattern, "", current, flags=re.IGNORECASE).strip(" -:")
if trimmed and trimmed != current:
current = trimmed
if current not in variants:
variants.append(current)
changed = True
return variants
def title_match_variants(value):
variants = set()
for raw_value in base_title_variants(value):
normalized = normalize_title_key(raw_value)
if normalized:
variants.add(normalized)
stripped = re.sub(r"\b(19|20)\d{2}\b", " ", normalized)
stripped = re.sub(r"\s+", " ", stripped).strip()
if stripped:
variants.add(stripped)
return variants
def title_lookup_queries(value):
queries = []
for raw_value in base_title_variants(value):
text = str(raw_value or "").strip()
if text and text not in queries:
queries.append(text)
return queries
def animeschedule_request(path, params=None):
params = params or {}
query = "&".join(f"{quote(str(key), safe='')}={quote(str(value), safe='')}" for key, value in params.items())
url = f"{ANIMESCHEDULE_API}{path}"
if query:
url = f"{url}?{query}"
debug_log("animeschedule.request", url=url, params=params)
request = urllib.request.Request(
url,
headers={
"Accept": "application/json",
"User-Agent": AGENT,
},
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=25) as response:
raw = response.read().decode("utf-8")
debug_log(
"animeschedule.response",
url=url,
status=getattr(response, "status", "unknown"),
bytes=len(raw.encode("utf-8")),
snippet=raw[:200],
)
except urllib.error.URLError as exc:
debug_log("animeschedule.error", url=url, error=exc)
raise RuntimeError(f"Could not reach AnimeSchedule: {exc}") from exc
except TimeoutError as exc:
debug_log("animeschedule.timeout", url=url)
raise RuntimeError("AnimeSchedule request timed out") from exc
try:
return json.loads(raw)
except json.JSONDecodeError as exc:
raise RuntimeError("AnimeSchedule returned unreadable JSON") from exc
def collect_animeschedule_entries(payload):
entries = []
def walk(node):
if isinstance(node, dict):
route = str(node.get("route") or "").strip()
image_route = str(node.get("imageVersionRoute") or "").strip()
title = str(node.get("title") or "").strip()
names = node.get("names") if isinstance(node.get("names"), dict) else {}
if route and (title or names):
entries.append(
{
"route": route,
"title": title,
"imageVersionRoute": image_route,
"names": names,
}
)
for value in node.values():
walk(value)
elif isinstance(node, list):
for value in node:
walk(value)
walk(payload)
return entries
def animeschedule_title_candidates(entry):
values = []
if entry.get("title"):
values.append(entry["title"])
names = entry.get("names") if isinstance(entry.get("names"), dict) else {}
for key in ("english", "romaji", "native", "abbreviation"):
text = str(names.get(key) or "").strip()
if text:
values.append(text)
synonyms = names.get("synonyms") if isinstance(names.get("synonyms"), list) else []
for value in synonyms:
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_animeschedule_match(query_variants, entry):
best = None
for candidate in animeschedule_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 fetch_animeschedule_cover_by_route(route, fallback_title=""):
debug_log("thumbnail.source.route_lookup", source="AnimeSchedule", route=route, fallback_title=fallback_title)
payload = animeschedule_request(f"/anime/{quote(str(route).strip(), safe='')}")
if not isinstance(payload, dict):
raise RuntimeError("AnimeSchedule anime response was not an object")
image_route = str(payload.get("imageVersionRoute") or "").strip()
if not image_route:
raise RuntimeError("AnimeSchedule anime did not provide an image route")
debug_log(
"thumbnail.source.route_match",
source="AnimeSchedule",
route=str(payload.get("route") or route).strip(),
image_route=image_route,
title=str(payload.get("title") or fallback_title or "").strip(),
)
return {
"route": str(payload.get("route") or route).strip(),
"title": str(payload.get("title") or fallback_title or "").strip(),
"url": urljoin(ANIMESCHEDULE_IMAGE_BASE, image_route),
}
def fetch_animeschedule_cover(title):
query_variants = title_match_variants(title)
if not query_variants:
raise RuntimeError("Missing title for AnimeSchedule lookup")
queries = title_lookup_queries(title)
debug_log("thumbnail.source.title_lookup", source="AnimeSchedule", title=title, variants=sorted(query_variants), queries=queries)
best = None
candidate_count = 0
last_error = None
for query in queries:
try:
payload = animeschedule_request("/anime", {"q": query, "st": "alphabetic"})
except Exception as exc:
last_error = exc
debug_log("thumbnail.source.query_failed", source="AnimeSchedule", title=title, query=query, error=exc)
continue
for entry in collect_animeschedule_entries(payload):
candidate_count += 1
score = score_animeschedule_match(query_variants, entry)
if score is None:
continue
if best is None or score > best["score"]:
best = {
"route": entry["route"],
"title": next(iter(animeschedule_title_candidates(entry)), entry.get("title") or str(title)),
"imageVersionRoute": str(entry.get("imageVersionRoute") or "").strip(),
"score": score,
}
if best is not None:
break
if not best:
debug_log("thumbnail.source.no_match", source="AnimeSchedule", title=title, candidates=candidate_count, queries=queries)
if last_error is not None:
raise last_error
raise RuntimeError("AnimeSchedule lookup did not find a matching anime")
debug_log(
"thumbnail.source.best_match",
source="AnimeSchedule",
title=title,
route=best["route"],
matched_title=best["title"],
image_route=best["imageVersionRoute"],
score=best["score"],
candidates=candidate_count,
)
if best["imageVersionRoute"]:
return {
"route": best["route"],
"title": str(best["title"]).strip(),
"url": urljoin(ANIMESCHEDULE_IMAGE_BASE, best["imageVersionRoute"]),
}
return fetch_animeschedule_cover_by_route(best["route"], fallback_title=best["title"])
def anidb_titles_cache_due():
if not ANIDB_TITLES_PATH.exists():
return True
age = time.time() - ANIDB_TITLES_PATH.stat().st_mtime
return age >= 24 * 60 * 60
def ensure_anidb_titles_dump(force=False):
if not force and not anidb_titles_cache_due():
return ANIDB_TITLES_PATH
debug_log("anidb.titles.request", url=ANIDB_TITLES_URL, force=force)
request = urllib.request.Request(
ANIDB_TITLES_URL,
headers={
"Accept": "application/gzip, application/octet-stream;q=0.9, */*;q=0.5",
"User-Agent": AGENT,
},
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
payload = response.read()
debug_log(
"anidb.titles.response",
status=getattr(response, "status", "unknown"),
bytes=len(payload),
content_type=response.headers.get("Content-Type", ""),
)
except urllib.error.URLError as exc:
debug_log("anidb.titles.error", error=exc)
if ANIDB_TITLES_PATH.exists():
return ANIDB_TITLES_PATH
raise RuntimeError(f"Could not download AniDB titles dump: {exc}") from exc
except TimeoutError as exc:
if ANIDB_TITLES_PATH.exists():
return ANIDB_TITLES_PATH
raise RuntimeError("AniDB titles dump download timed out") from exc
if not payload:
if ANIDB_TITLES_PATH.exists():
return ANIDB_TITLES_PATH
raise RuntimeError("AniDB titles dump was empty")
tmp_path = ANIDB_TITLES_PATH.with_suffix(".xml.gz.tmp")
tmp_path.write_bytes(payload)
tmp_path.replace(ANIDB_TITLES_PATH)
return ANIDB_TITLES_PATH
def iter_anidb_titles():
path = ensure_anidb_titles_dump()
mtime = path.stat().st_mtime if path.exists() else None
if ANIDB_TITLES_CACHE["entries"] is not None and ANIDB_TITLES_CACHE["mtime"] == mtime:
for entry in ANIDB_TITLES_CACHE["entries"]:
yield entry
return
entries = []
with gzip.open(path, "rb") as handle:
context = ET.iterparse(handle, events=("end",))
for _, elem in context:
if elem.tag != "anime":
continue
aid = str(elem.attrib.get("aid") or "").strip()
if not aid:
elem.clear()
continue
for title_elem in elem.findall("title"):
title = str(title_elem.text or "").strip()
if not title:
continue
entries.append(
{
"aid": aid,
"title": title,
"type": str(title_elem.attrib.get("type") or "").strip().lower(),
"lang": str(title_elem.attrib.get("{http://www.w3.org/XML/1998/namespace}lang") or "").strip().lower(),
}
)
elem.clear()
ANIDB_TITLES_CACHE["mtime"] = mtime
ANIDB_TITLES_CACHE["entries"] = entries
for entry in entries:
yield entry
def score_anidb_title_match(query_variants, candidate):
normalized = normalize_title_key(candidate["title"])
if not normalized:
return None
score = None
if normalized in query_variants:
score = 100
else:
for variant in query_variants:
if variant and (normalized in variant or variant in normalized):
score = max(score or 0, 72 - abs(len(normalized) - len(variant)))
if score is None:
return None
type_bonus = {
"official": 12,
"main": 10,
"short": 6,
"syn": 4,
}.get(candidate["type"], 0)
lang_bonus = {
"en": 5,
"x-jat": 4,
"ja": 2,
}.get(candidate["lang"], 0)
return score + type_bonus + lang_bonus
def resolve_anidb_aid(title):
query_variants = title_match_variants(title)
if not query_variants:
raise RuntimeError("Missing title for AniDB lookup")
best = None
candidate_count = 0
for candidate in iter_anidb_titles():
candidate_count += 1
score = score_anidb_title_match(query_variants, candidate)
if score is None:
continue
if best is None or score > best["score"]:
best = {
"aid": candidate["aid"],
"title": candidate["title"],
"score": score,
}
if not best:
debug_log("thumbnail.source.no_match", source="AniDB", title=title, candidates=candidate_count)
raise RuntimeError("AniDB title lookup did not find a matching anime")
debug_log(
"thumbnail.source.best_match",
source="AniDB",
title=title,
aid=best["aid"],
matched_title=best["title"],
score=best["score"],
candidates=candidate_count,
)
return best
def fetch_anidb_cover(title):
match = resolve_anidb_aid(title)
return fetch_anidb_cover_by_aid(match["aid"], fallback_title=match["title"])
def fetch_anidb_cover_by_aid(aid, fallback_title=""):
page_url = ANIDB_ANIME_PAGE.format(aid=aid)
debug_log("thumbnail.source.route_lookup", source="AniDB", aid=aid, fallback_title=fallback_title, page_url=page_url)
request = urllib.request.Request(
page_url,
headers={
"Accept": "text/html,application/xhtml+xml",
"Accept-Language": "en-US,en;q=0.9",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
"Referer": "https://anidb.net/",
"User-Agent": AGENT,
},
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=25) as response:
html = response.read().decode("utf-8", errors="replace")
debug_log(
"anidb.page.response",
aid=aid,
status=getattr(response, "status", "unknown"),
bytes=len(html.encode("utf-8")),
snippet=html[:200],
)
except urllib.error.URLError as exc:
debug_log("anidb.page.error", aid=aid, error=exc)
raise RuntimeError(f"Could not reach AniDB anime page: {exc}") from exc
except TimeoutError as exc:
raise RuntimeError("AniDB anime page request timed out") from exc
candidates = [
r'((?:https?:)?//cdn-[^/"\']+\.anidb\.net/images/main/[^"\'\s>]+)',
r'((?:https?:)?//anidb\.net/images/main/[^"\'\s>]+)',
r'(["\'])(/images/main/[^"\']+)\1',
r'(["\'])(/[^"\']*images/main/[^"\']+)\1',
]
image_url = None
for pattern in candidates:
match = re.search(pattern, html, re.IGNORECASE)
if not match:
continue
image_url = match.group(1) if match.lastindex == 1 else match.group(2)
break
if not image_url:
debug_log("anidb.page.no_image", aid=aid)
raise RuntimeError("AniDB page did not expose a cover image")
if image_url.startswith("//"):
image_url = f"https:{image_url}"
elif image_url.startswith("/"):
image_url = urljoin(page_url, image_url)
debug_log("anidb.page.image", aid=aid, image_url=image_url)
return {
"aid": str(aid).strip(),
"title": str(fallback_title or "").strip(),
"url": image_url,
}
def cache_thumbnail_image(show_id, image_url):
THUMBNAIL_ROOT.mkdir(parents=True, exist_ok=True)
debug_log("thumbnail.download.request", show_id=show_id, image_url=image_url)
request = urllib.request.Request(
image_url,
headers={
"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",
"Referer": "https://anidb.net/",
"User-Agent": AGENT,
},
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=25) as response:
content = response.read()
content_type = response.headers.get("Content-Type", "")
debug_log(
"thumbnail.download.response",
show_id=show_id,
status=getattr(response, "status", "unknown"),
bytes=len(content),
content_type=content_type,
final_url=getattr(response, "url", image_url),
)
except urllib.error.URLError as exc:
debug_log("thumbnail.download.error", show_id=show_id, image_url=image_url, error=exc)
raise RuntimeError(f"Could not download cover image: {exc}") from exc
except TimeoutError as exc:
debug_log("thumbnail.download.timeout", show_id=show_id, image_url=image_url)
raise RuntimeError("Cover image download timed out") from exc
if not content:
raise RuntimeError("Cover image response was empty")
stem = sanitize_path_component(show_id, "anime")
extension = infer_image_extension(image_url, content_type)
final_path = THUMBNAIL_ROOT / f"{stem}{extension}"
tmp_path = final_path.with_suffix(final_path.suffix + ".tmp")
with tmp_path.open("wb") as handle:
handle.write(content)
tmp_path.replace(final_path)
for existing in THUMBNAIL_ROOT.glob(f"{stem}.*"):
if existing != final_path and existing.is_file():
existing.unlink(missing_ok=True)
debug_log("thumbnail.cache.write", show_id=show_id, path=final_path, bytes=len(content), content_type=content_type)
return final_path.name
def search_anime(query, mode):
search_gql = (
"query( $search: SearchInput $limit: Int $page: Int "
"$translationType: VaildTranslationTypeEnumType $countryOrigin: VaildCountryOriginEnumType ) "
"{ shows( search: $search limit: $limit page: $page translationType: $translationType "
"countryOrigin: $countryOrigin ) { edges { _id name availableEpisodes __typename } }}"
)
data = graph_request(
search_gql,
{
"search": {"allowAdult": False, "allowUnknown": False, "query": query},
"limit": 40,
"page": 1,
"translationType": mode,
"countryOrigin": "ALL",
},
)
shows = (((data.get("shows") or {}).get("edges")) or [])
results = []
for show in shows:
episodes = (show.get("availableEpisodes") or {}).get(mode)
if not episodes:
continue
title = str(show.get("name") or "").replace('\\"', '"').strip()
if not show.get("_id") or not title:
continue
index = len(results) + 1
results.append(
{
"id": show["_id"],
"title": title,
"episodes": episodes,
"index": index,
"query": query,
"mode": mode,
}
)
return results
def numeric_episode_key(value):
parts = re.findall(r"\d+|\D+", str(value))
key = []
for part in parts:
key.append((0, int(part)) if part.isdigit() else (1, part))
return key
def fetch_show_episode_snapshot(show_id):
query = "query ($showId: String!) { show(_id: $showId) { _id name availableEpisodesDetail } }"
data = graph_request(query, {"showId": show_id})
show = data.get("show") or {}
detail = show.get("availableEpisodesDetail") or {}
episodes = {}
for mode in MODE_CHOICES:
values = [str(item).strip() for item in detail.get(mode, []) if str(item).strip()]
episodes[mode] = sorted(values, key=numeric_episode_key)
return {
"show_id": str(show.get("_id") or show_id),
"title": str(show.get("name") or "").replace('\\"', '"').strip(),
"episodes": episodes,
}
def episode_list(show_id, mode):
normalized_mode = str(mode or "").strip().lower()
if normalized_mode not in MODE_CHOICES:
raise ValueError("Mode must be sub or dub")
return fetch_show_episode_snapshot(show_id)["episodes"][normalized_mode]
class WatchlistStore:
def __init__(self):
self.lock = threading.RLock()
self.thumbnail_events = {}
self._init_db()
self._migrate_legacy_json()
def _connect(self):
conn = sqlite3.connect(QUEUE_DB_PATH, timeout=30)
conn.row_factory = sqlite3.Row
return conn
def _init_db(self):
QUEUE_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with self._connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS watchlist (
show_id TEXT PRIMARY KEY,
title TEXT NOT NULL,
status TEXT NOT NULL,
status_message TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
last_checked TEXT,
sub_count INTEGER NOT NULL DEFAULT 0,
dub_count INTEGER NOT NULL DEFAULT 0,
sub_latest_episode TEXT,
dub_latest_episode TEXT,
animeschedule_route TEXT,
animeschedule_title TEXT,
anidb_aid TEXT,
anidb_title TEXT,
thumbnail_path TEXT,
thumbnail_checked_at TEXT
)
"""
)
columns = {row["name"] for row in conn.execute("PRAGMA table_info(watchlist)").fetchall()}
if "thumbnail_path" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN thumbnail_path TEXT")
if "thumbnail_checked_at" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN thumbnail_checked_at TEXT")
if "anidb_aid" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN anidb_aid TEXT")
if "anidb_title" not in columns:
conn.execute("ALTER TABLE watchlist ADD COLUMN anidb_title TEXT")
if "animeschedule_route" not in columns:
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")
conn.execute("CREATE INDEX IF NOT EXISTS idx_watchlist_title ON watchlist(title)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_watchlist_updated_at ON watchlist(updated_at DESC)")
def _row_to_item(self, row):
item = dict(row)
item["sub_count"] = int(item.get("sub_count") or 0)
item["dub_count"] = int(item.get("dub_count") or 0)
item["total_count"] = item["sub_count"] + item["dub_count"]
item["source_url"] = anime_source_url(item.get("show_id"), item.get("title"))
thumb_path = thumbnail_file_path(item.get("thumbnail_path"))
item["thumbnail_ready"] = bool(thumb_path and thumb_path.exists())
item["thumbnail_url"] = cover_route(item.get("show_id"))
if item.get("animeschedule_route"):
title = str(item.get("animeschedule_title") or item.get("title") or "").strip() or "unknown title"
item["thumbnail_debug"] = f"AnimeSchedule: {item['animeschedule_route']} - {title}"
elif item.get("anidb_aid"):
title = str(item.get("anidb_title") or item.get("title") or "").strip() or "unknown title"
item["thumbnail_debug"] = f"AniDB: {item['anidb_aid']} - {title}"
elif item["thumbnail_ready"]:
item["thumbnail_debug"] = "Thumbnail source: manual upload"
else:
item["thumbnail_debug"] = "Thumbnail source: unresolved"
return item
def _upsert_conn(self, conn, item):
conn.execute(
"""
INSERT INTO watchlist (
show_id, title, status, status_message, created_at, updated_at,
last_checked, sub_count, dub_count, sub_latest_episode, dub_latest_episode,
animeschedule_route, animeschedule_title, anidb_aid, anidb_title,
thumbnail_path, thumbnail_checked_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(show_id) DO UPDATE SET
title = excluded.title,
status = excluded.status,
status_message = excluded.status_message,
updated_at = excluded.updated_at,
last_checked = excluded.last_checked,
sub_count = excluded.sub_count,
dub_count = excluded.dub_count,
sub_latest_episode = excluded.sub_latest_episode,
dub_latest_episode = excluded.dub_latest_episode,
animeschedule_route = excluded.animeschedule_route,
animeschedule_title = excluded.animeschedule_title,
anidb_aid = excluded.anidb_aid,
anidb_title = excluded.anidb_title,
thumbnail_path = excluded.thumbnail_path,
thumbnail_checked_at = excluded.thumbnail_checked_at
""",
(
item["show_id"],
item["title"],
item["status"],
item["status_message"],
item["created_at"],
item["updated_at"],
item["last_checked"],
int(item.get("sub_count") or 0),
int(item.get("dub_count") or 0),
item.get("sub_latest_episode"),
item.get("dub_latest_episode"),
item.get("animeschedule_route"),
item.get("animeschedule_title"),
item.get("anidb_aid"),
item.get("anidb_title"),
item.get("thumbnail_path"),
item.get("thumbnail_checked_at"),
),
)
def _migrate_legacy_json(self):
if not WATCHLIST_JSON_PATH.exists():
return
legacy = load_json(WATCHLIST_JSON_PATH, {})
if not legacy:
WATCHLIST_JSON_PATH.rename(WATCHLIST_JSON_PATH.with_suffix(".json.migrated"))
return
aggregated = {}
for value in legacy.values():
if not isinstance(value, dict):
continue
show_id = str(value.get("show_id") or "").strip()
if not show_id:
continue
mode = str(value.get("mode") or "").strip().lower()
if mode not in MODE_CHOICES:
continue
item = aggregated.setdefault(
show_id,
{
"show_id": show_id,
"title": str(value.get("title") or "Unknown Anime").strip() or "Unknown Anime",
"status": str(value.get("status") or "tracked"),
"status_message": str(value.get("status_message") or "Migrated from legacy watchlist."),
"created_at": str(value.get("added_at") or now_iso()),
"updated_at": now_iso(),
"last_checked": value.get("last_checked"),
"sub_count": 0,
"dub_count": 0,
"sub_latest_episode": None,
"dub_latest_episode": None,
"animeschedule_route": None,
"animeschedule_title": None,
"anidb_aid": None,
"anidb_title": None,
"thumbnail_path": None,
"thumbnail_checked_at": None,
},
)
episodes = value.get("episodes_available") if isinstance(value.get("episodes_available"), list) else []
count = len(episodes)
latest = episodes[-1] if episodes else None
if mode == "sub":
item["sub_count"] = max(item["sub_count"], count)
item["sub_latest_episode"] = latest or item["sub_latest_episode"]
else:
item["dub_count"] = max(item["dub_count"], count)
item["dub_latest_episode"] = latest or item["dub_latest_episode"]
if item["title"] == "Unknown Anime" and str(value.get("title") or "").strip():
item["title"] = str(value.get("title")).strip()
with self.lock, self._connect() as conn:
for item in aggregated.values():
self._upsert_conn(conn, item)
WATCHLIST_JSON_PATH.rename(WATCHLIST_JSON_PATH.with_suffix(".json.migrated"))
def list(self, page=1, per_page=30):
page = max(1, int(page or 1))
per_page = min(60, max(1, int(per_page or 30)))
offset = (page - 1) * per_page
with self.lock, self._connect() as conn:
total = conn.execute("SELECT COUNT(*) AS count FROM watchlist").fetchone()["count"]
totals = conn.execute(
"""
SELECT
COALESCE(SUM(sub_count), 0) AS sub_total,
COALESCE(SUM(dub_count), 0) AS dub_total
FROM watchlist
"""
).fetchone()
rows = conn.execute(
"SELECT * FROM watchlist ORDER BY title COLLATE NOCASE ASC LIMIT ? OFFSET ?",
(per_page, offset),
).fetchall()
items = [self._row_to_item(row) for row in rows]
pages = max(1, (total + per_page - 1) // per_page)
return {
"items": items,
"total": total,
"page": min(page, pages),
"pages": pages,
"per_page": per_page,
"summary": {
"shows": int(total),
"sub": int(totals["sub_total"] or 0),
"dub": int(totals["dub_total"] or 0),
},
}
def get(self, show_id):
with self.lock, self._connect() as conn:
row = conn.execute("SELECT * FROM watchlist WHERE show_id = ?", (str(show_id).strip(),)).fetchone()
if not row:
raise KeyError("Anime not found in watchlist.")
return self._row_to_item(row)
def add(self, payload):
show_id = str(payload.get("show_id") or "").strip()
title = str(payload.get("title") or "Unknown Anime").strip() or "Unknown Anime"
if not show_id:
raise ValueError("Missing show_id.")
with self.lock, self._connect() as conn:
row = conn.execute("SELECT * FROM watchlist WHERE show_id = ?", (show_id,)).fetchone()
if row:
item = self._row_to_item(row)
return {"message": f"Already tracking {item['title']}.", "item": item, "created": False}
now = now_iso()
item = {
"show_id": show_id,
"title": title,
"status": "tracked",
"status_message": "Waiting for the first manual refresh.",
"created_at": now,
"updated_at": now,
"last_checked": None,
"sub_count": 0,
"dub_count": 0,
"sub_latest_episode": None,
"dub_latest_episode": None,
"animeschedule_route": None,
"animeschedule_title": None,
"anidb_aid": None,
"anidb_title": None,
"thumbnail_path": None,
"thumbnail_checked_at": None,
}
self._upsert_conn(conn, item)
refreshed = self.refresh(show_id)
return {"message": f"Added {refreshed['title']} to the watchlist.", "item": refreshed, "created": True}
def refresh(self, show_id):
existing = self.get(show_id)
now = now_iso()
try:
snapshot = fetch_show_episode_snapshot(show_id)
sub_episodes = snapshot["episodes"]["sub"]
dub_episodes = snapshot["episodes"]["dub"]
item = {
"show_id": existing["show_id"],
"title": snapshot["title"] or existing["title"],
"status": "updated",
"status_message": f"Sub: {len(sub_episodes)} episodes available. Dub: {len(dub_episodes)} episodes available.",
"created_at": existing["created_at"],
"updated_at": now,
"last_checked": now,
"sub_count": len(sub_episodes),
"dub_count": len(dub_episodes),
"sub_latest_episode": sub_episodes[-1] if sub_episodes else None,
"dub_latest_episode": dub_episodes[-1] if dub_episodes else None,
"animeschedule_route": existing.get("animeschedule_route"),
"animeschedule_title": existing.get("animeschedule_title"),
"anidb_aid": existing.get("anidb_aid"),
"anidb_title": existing.get("anidb_title"),
"thumbnail_path": existing.get("thumbnail_path"),
"thumbnail_checked_at": existing.get("thumbnail_checked_at"),
}
except Exception as exc:
item = {
"show_id": existing["show_id"],
"title": existing["title"],
"status": "error",
"status_message": f"Could not refresh episodes: {exc}",
"created_at": existing["created_at"],
"updated_at": now,
"last_checked": now,
"sub_count": existing["sub_count"],
"dub_count": existing["dub_count"],
"sub_latest_episode": existing.get("sub_latest_episode"),
"dub_latest_episode": existing.get("dub_latest_episode"),
"animeschedule_route": existing.get("animeschedule_route"),
"animeschedule_title": existing.get("animeschedule_title"),
"anidb_aid": existing.get("anidb_aid"),
"anidb_title": existing.get("anidb_title"),
"thumbnail_path": existing.get("thumbnail_path"),
"thumbnail_checked_at": existing.get("thumbnail_checked_at"),
}
with self.lock, self._connect() as conn:
self._upsert_conn(conn, item)
try:
self.ensure_thumbnail(show_id)
except Exception:
pass
return self.get(show_id)
def refresh_all(self):
items = self.list(page=1, per_page=10000)["items"]
refreshed = [self.refresh(item["show_id"]) for item in items]
return {"count": len(refreshed), "items": refreshed, "message": f"Refreshed {len(refreshed)} watchlist entries."}
def remove(self, show_id):
existing = self.get(show_id)
thumb_path = thumbnail_file_path(existing.get("thumbnail_path"))
with self.lock, self._connect() as conn:
cursor = conn.execute("DELETE FROM watchlist WHERE show_id = ?", (str(show_id).strip(),))
if cursor.rowcount < 1:
raise KeyError("Anime not found in watchlist.")
self.thumbnail_events.pop(str(show_id).strip(), None)
if thumb_path and thumb_path.exists():
thumb_path.unlink(missing_ok=True)
return {"ok": True, "message": f"Removed {existing['title']} from the watchlist.", "item": existing}
def set_manual_thumbnail(self, show_id, filename, data_url):
existing = self.get(show_id)
content_type, content = decode_image_data_url(data_url)
cached_name = save_thumbnail_bytes(existing["show_id"], filename or existing["title"], content_type, content)
checked_at = now_iso()
with self.lock, self._connect() as conn:
conn.execute(
"""
UPDATE watchlist
SET thumbnail_path = ?, thumbnail_checked_at = ?, updated_at = ?
WHERE show_id = ?
""",
(cached_name, checked_at, checked_at, existing["show_id"]),
)
item = self.get(show_id)
return {"message": f"Uploaded thumbnail for {item['title']}.", "item": item}
def ensure_thumbnail(self, show_id, force=False):
item = self.get(show_id)
current_path = thumbnail_file_path(item.get("thumbnail_path"))
current_exists = bool(current_path and current_path.exists())
debug_log(
"thumbnail.ensure.start",
show_id=show_id,
force=force,
title=item.get("title"),
cached_path=current_path,
cached_exists=current_exists,
thumbnail_ready=item.get("thumbnail_ready"),
route=item.get("animeschedule_route"),
aid=item.get("anidb_aid"),
)
if current_exists and not force:
debug_log("thumbnail.ensure.cached", show_id=show_id, path=current_path)
return current_path
if not force and current_path is None and not thumbnail_retry_due(item.get("thumbnail_checked_at")):
debug_log("thumbnail.ensure.skipped_retry_window", show_id=show_id, checked_at=item.get("thumbnail_checked_at"))
return None
if current_path and not current_exists:
debug_log("thumbnail.ensure.missing_cached_file", show_id=show_id, expected_path=current_path)
with self.lock:
event = self.thumbnail_events.get(show_id)
if event:
wait_event = event
else:
wait_event = threading.Event()
self.thumbnail_events[show_id] = wait_event
event = None
if event:
wait_event.wait(timeout=20)
refreshed = self.get(show_id)
refreshed_path = thumbnail_file_path(refreshed.get("thumbnail_path"))
debug_log("thumbnail.ensure.waited_for_inflight", show_id=show_id, resolved_path=refreshed_path)
return refreshed_path if refreshed_path and refreshed_path.exists() else None
cached_name = None
cover = None
checked_at = now_iso()
try:
stored_route = str(item.get("animeschedule_route") or "").strip()
stored_schedule_title = str(item.get("animeschedule_title") or item.get("title") or "").strip()
if stored_route:
try:
debug_log("thumbnail.ensure.try_route", source="AnimeSchedule", show_id=show_id, route=stored_route)
cover = fetch_animeschedule_cover_by_route(stored_route, fallback_title=stored_schedule_title)
except Exception:
debug_log("thumbnail.ensure.route_failed", source="AnimeSchedule", show_id=show_id, route=stored_route)
if not force:
raise
if cover is None:
try:
debug_log("thumbnail.ensure.try_title", source="AnimeSchedule", show_id=show_id, title=item["title"])
cover = fetch_animeschedule_cover(item["title"])
except Exception:
debug_log("thumbnail.ensure.title_failed", source="AnimeSchedule", show_id=show_id, title=item["title"])
cover = None
stored_aid = str(item.get("anidb_aid") or "").strip()
stored_title = str(item.get("anidb_title") or item.get("title") or "").strip()
if cover is None and stored_aid:
try:
debug_log("thumbnail.ensure.try_route", source="AniDB", show_id=show_id, aid=stored_aid)
cover = fetch_anidb_cover_by_aid(stored_aid, fallback_title=stored_title)
except Exception:
debug_log("thumbnail.ensure.route_failed", source="AniDB", show_id=show_id, aid=stored_aid)
if not force:
raise
if cover is None:
debug_log("thumbnail.ensure.try_title", source="AniDB", show_id=show_id, title=item["title"])
cover = fetch_anidb_cover(item["title"])
debug_log("thumbnail.ensure.cover_selected", show_id=show_id, source="AnimeSchedule" if cover.get("route") else "AniDB", cover=cover)
cached_name = cache_thumbnail_image(item["show_id"], cover["url"])
except Exception as exc:
debug_log("thumbnail.ensure.failed", show_id=show_id, error=exc)
cached_name = None
checked_at = item.get("thumbnail_checked_at")
with self.lock, self._connect() as conn:
conn.execute(
"""
UPDATE watchlist
SET thumbnail_path = ?, thumbnail_checked_at = ?,
animeschedule_route = ?, animeschedule_title = ?,
anidb_aid = ?, anidb_title = ?
WHERE show_id = ?
""",
(
cached_name,
checked_at,
(cover or {}).get("route"),
(cover or {}).get("title") if (cover or {}).get("route") else item.get("animeschedule_title"),
(cover or {}).get("aid"),
(cover or {}).get("title") if (cover or {}).get("aid") else item.get("anidb_title"),
item["show_id"],
),
)
inflight = self.thumbnail_events.get(show_id)
if inflight is wait_event:
inflight.set()
self.thumbnail_events.pop(show_id, None)
debug_log(
"thumbnail.ensure.finish",
show_id=show_id,
cached_name=cached_name,
route=(cover or {}).get("route"),
aid=(cover or {}).get("aid"),
checked_at=checked_at,
)
return thumbnail_file_path(cached_name)
def ensure_thumbnails(self, show_ids, limit=6, force=False):
normalized = []
for show_id in show_ids:
text = str(show_id or "").strip()
if text and text not in normalized:
normalized.append(text)
updated = []
for show_id in normalized[: max(1, min(int(limit or 6), 10))]:
item = self.get(show_id)
if not force and item.get("thumbnail_ready"):
updated.append(item)
continue
try:
self.ensure_thumbnail(show_id, force=force)
except Exception:
pass
updated.append(self.get(show_id))
return {"items": updated}
WATCHLIST = WatchlistStore()
def add_to_watchlist(payload):
return WATCHLIST.add(payload)
def get_watchlist(page=1, per_page=30):
return WATCHLIST.list(page=page, per_page=per_page)
def update_watchlist_status(show_id, _mode=None):
item = WATCHLIST.refresh(show_id)
return {"message": f"Updated {item['title']}.", "item": item}
def update_all_watchlist_statuses():
return WATCHLIST.refresh_all()
def remove_from_watchlist(show_id):
return WATCHLIST.remove(show_id)
def upload_watchlist_thumbnail(payload):
return WATCHLIST.set_manual_thumbnail(payload["show_id"], payload.get("filename"), payload.get("data_url"))
class DownloadQueue:
def __init__(self):
self.lock = threading.RLock()
self.wakeup = threading.Event()
self.current_process = None
self.current_job_id = None
self._init_db()
self._migrate_json_queue()
self._restore_interrupted_jobs()
self.worker = threading.Thread(target=self._run, daemon=True)
self.worker.start()
def _connect(self):
conn = sqlite3.connect(QUEUE_DB_PATH, timeout=30)
conn.row_factory = sqlite3.Row
return conn
def _init_db(self):
QUEUE_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with self._connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS jobs (
id TEXT PRIMARY KEY,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
payload TEXT NOT NULL
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_created_at ON jobs(created_at)")
def _migrate_json_queue(self):
if not QUEUE_PATH.exists():
return
jobs = load_json(QUEUE_PATH, [])
if not jobs:
QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated"))
return
with self.lock, self._connect() as conn:
existing = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"]
if existing:
return
for job in jobs:
if isinstance(job, dict) and job.get("id"):
self._upsert_job_conn(conn, job)
QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated"))
def _restore_interrupted_jobs(self):
with self.lock, self._connect() as conn:
rows = conn.execute("SELECT payload FROM jobs WHERE status = 'running'").fetchall()
for row in rows:
job = self._job_from_row(row)
job["status"] = "pending"
job["pid"] = None
job.setdefault("log", []).append("Restarted after web app restart.")
self._upsert_job_conn(conn, job)
def _job_from_row(self, row):
payload = row["payload"] if isinstance(row, sqlite3.Row) else row[0]
return json.loads(payload)
def _upsert_job_conn(self, conn, job):
clean = dict(job)
clean.pop("process", None)
conn.execute(
"""
INSERT INTO jobs (id, status, created_at, updated_at, payload)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
status = excluded.status,
updated_at = excluded.updated_at,
payload = excluded.payload
""",
(
clean["id"],
clean.get("status", "pending"),
clean.get("created_at") or now_iso(),
clean.get("updated_at") or now_iso(),
json.dumps(clean, sort_keys=True),
),
)
def _save_job_locked(self, job):
with self._connect() as conn:
self._upsert_job_conn(conn, job)
def list(self, page=1, per_page=10):
page = max(1, int(page or 1))
per_page = min(50, max(1, int(per_page or 10)))
offset = (page - 1) * per_page
with self.lock, self._connect() as conn:
total = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"]
rows = conn.execute(
"SELECT payload FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?",
(per_page, offset),
).fetchall()
pages = max(1, (total + per_page - 1) // per_page)
return {
"jobs": [self._job_from_row(row) for row in rows],
"page": page,
"per_page": per_page,
"total": total,
"pages": pages,
}
def add(self, payload):
job = build_job(payload)
with self.lock:
self._save_job_locked(job)
self.wakeup.set()
return job
def retry(self, job_id):
with self.lock:
job = self._find(job_id)
if job["status"] == "running":
raise ValueError("Running jobs cannot be retried")
job["status"] = "pending"
job["exit_code"] = None
job["pid"] = None
job["started_at"] = None
job["finished_at"] = None
job["updated_at"] = now_iso()
job["log"] = []
self._save_job_locked(job)
self.wakeup.set()
return job
def retry_all_failed(self):
count = 0
with self.lock, self._connect() as conn:
rows = conn.execute("SELECT payload FROM jobs WHERE status = 'failed'").fetchall()
for row in rows:
job = self._job_from_row(row)
job["status"] = "pending"
job["exit_code"] = None
job["pid"] = None
job["started_at"] = None
job["finished_at"] = None
job["updated_at"] = now_iso()
job["log"] = []
self._upsert_job_conn(conn, job)
count += 1
if count:
self.wakeup.set()
return {"ok": True, "count": count}
def cancel(self, job_id):
with self.lock:
job = self._find(job_id)
if job["status"] == "pending":
job["status"] = "canceled"
job["updated_at"] = now_iso()
self._save_job_locked(job)
return job
if job["status"] != "running":
return job
if self.current_job_id == job_id and self.current_process:
try:
os.killpg(os.getpgid(self.current_process.pid), signal.SIGTERM)
except OSError:
self.current_process.terminate()
job["cancel_requested"] = True
job["updated_at"] = now_iso()
self._save_job_locked(job)
return job
raise ValueError("Could not find the running process")
def remove(self, job_id):
with self.lock:
job = self._find(job_id)
if job["status"] == "running":
raise ValueError("Running jobs must be canceled before removing")
with self._connect() as conn:
conn.execute("DELETE FROM jobs WHERE id = ?", (job_id,))
return {"ok": True}
def clear_finished(self):
with self.lock, self._connect() as conn:
cursor = conn.execute("DELETE FROM jobs WHERE status IN ('done', 'canceled')")
return {"ok": True, "count": cursor.rowcount}
def _find(self, job_id):
with self._connect() as conn:
row = conn.execute("SELECT payload FROM jobs WHERE id = ?", (job_id,)).fetchone()
if row:
return self._job_from_row(row)
raise KeyError("Job not found")
def _next_pending(self):
with self.lock:
with self._connect() as conn:
row = conn.execute(
"SELECT payload FROM jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1"
).fetchone()
if row:
return self._job_from_row(row)
return None
def _append_log(self, job, line):
text = strip_control(line).strip()
if not text:
return
with self.lock:
job.setdefault("log", []).append(text)
job["log"] = job["log"][-MAX_LOG_LINES:]
job["updated_at"] = now_iso()
self._save_job_locked(job)
def _run(self):
while True:
job = self._next_pending()
if not job:
self.wakeup.wait(2)
self.wakeup.clear()
continue
self._run_job(job)
def _run_job(self, job):
command = command_for_job(job)
staging_dir = job_staging_dir(job)
target_dir = job_output_dir(job)
env = os.environ.copy()
env.update(
{
"ANI_CLI_DOWNLOAD_DIR": str(staging_dir),
"ANI_CLI_MODE": job["mode"],
"ANI_CLI_QUALITY": job["quality"],
"TERM": env.get("TERM", "xterm-256color"),
}
)
staging_dir.mkdir(parents=True, exist_ok=True)
target_dir.mkdir(parents=True, exist_ok=True)
with self.lock:
job["status"] = "running"
job["started_at"] = now_iso()
job["updated_at"] = job["started_at"]
job["command"] = printable_command(command)
job["staging_dir"] = str(staging_dir)
job["target_dir"] = str(target_dir)
job["log"] = [
f"Starting: {job['command']}",
f"Staging in: {job['staging_dir']}",
f"Final folder: {job['target_dir']}",
]
job["cancel_requested"] = False
self._save_job_locked(job)
process = subprocess.Popen(
command,
cwd=str(PROJECT_ROOT),
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
start_new_session=True,
)
with self.lock:
self.current_process = process
self.current_job_id = job["id"]
job["pid"] = process.pid
self._save_job_locked(job)
assert process.stdout is not None
for line in process.stdout:
self._append_log(job, line)
exit_code = process.wait()
finalize_error = None
moved_files = []
if exit_code == 0 and not job.get("cancel_requested"):
try:
moved_files = finalize_library_files(job)
except Exception as exc:
finalize_error = exc
with self.lock:
canceled = job.get("cancel_requested")
job["exit_code"] = 1 if finalize_error else exit_code
job["pid"] = None
job["finished_at"] = now_iso()
job["updated_at"] = job["finished_at"]
if canceled:
job["status"] = "canceled"
job.setdefault("log", []).append("Canceled.")
elif finalize_error:
job["status"] = "failed"
job.setdefault("log", []).append(f"Library rename failed: {finalize_error}")
elif exit_code == 0:
job["status"] = "done"
for path in moved_files:
job.setdefault("log", []).append(f"Saved: {path}")
job.setdefault("log", []).append("Download completed.")
else:
job["status"] = "failed"
job.setdefault("log", []).append(f"Download failed with exit code {exit_code}.")
self.current_process = None
self.current_job_id = None
self._save_job_locked(job)
def strip_control(value):
return re.sub(r"\x1b\[[0-9;?]*[A-Za-z]", "", str(value)).replace("\r", "\n")
def printable_command(command):
return " ".join(sh_quote(part) for part in command)
def sh_quote(value):
if re.match(r"^[A-Za-z0-9_./:=+-]+$", value):
return value
return "'" + value.replace("'", "'\"'\"'") + "'"
def validate_episode_spec(value):
text = str(value or "").strip()
if not text:
raise ValueError("Choose at least one episode")
if not re.match(r"^[0-9.\-\s]+$", text):
raise ValueError("Episodes can contain numbers, dots, spaces, and ranges")
return re.sub(r"\s+", " ", text)
def sanitize_path_component(value, fallback="Anime"):
text = str(value or "").strip()
text = re.sub(r'[\\/:*?"<>|]+', "-", text)
text = re.sub(r"\s+", " ", text).strip(" .")
return text or fallback
def normalize_season(value):
text = str(value or "1").strip()
if not re.match(r"^[0-9]+$", text):
raise ValueError("Season must be a positive number")
number = int(text, 10)
if number < 1:
raise ValueError("Season must be a positive number")
return str(number)
def job_output_dir(job):
anime_name = sanitize_path_component(job.get("anime_name") or job.get("title"))
return Path(job["download_dir"]).expanduser() / anime_name / f"Season {season_label(job)}"
def job_staging_dir(job):
return STAGING_ROOT / job["id"]
def season_label(job):
return f"{int(job.get('season') or 1):02d}"
def episode_label(value):
text = str(value or "").strip()
match = re.match(r"^0*([0-9]+)(.*)$", text)
if match:
number = int(match.group(1) or "0", 10)
suffix = re.sub(r"[^0-9A-Za-z.]+", "-", match.group(2)).strip("-")
return f"{number:02d}{suffix}"
return re.sub(r"[^0-9A-Za-z.]+", "-", text).strip("-") or "00"
def extract_episode_from_filename(path):
match = re.search(r"\bEpisode\s+([0-9][0-9A-Za-z.\-]*)\s*$", path.stem)
return match.group(1) if match else None
def unique_destination(path):
if not path.exists():
return path
counter = 2
while True:
candidate = path.with_name(f"{path.stem} ({counter}){path.suffix}")
if not candidate.exists():
return candidate
counter += 1
def finalize_library_files(job):
staging_dir = job_staging_dir(job)
target_dir = job_output_dir(job)
library_name = sanitize_path_component(job.get("anime_name") or job.get("title"))
season = season_label(job)
target_dir.mkdir(parents=True, exist_ok=True)
files = sorted(
[path for path in staging_dir.iterdir() if path.is_file()],
key=lambda path: (path.stat().st_mtime, path.name),
)
if not files:
raise RuntimeError("No downloaded files were found in the staging folder")
moved = []
fallback_episode = 1
for source in files:
ep = extract_episode_from_filename(source)
if ep is None:
ep = str(fallback_episode)
fallback_episode += 1
final_name = f"{library_name} - S{season}E{episode_label(ep)}{source.suffix}"
destination = unique_destination(target_dir / final_name)
shutil.move(str(source), str(destination))
moved.append(str(destination))
try:
staging_dir.rmdir()
except OSError:
pass
return moved
def build_job(payload):
if not isinstance(payload, dict):
raise ValueError("Expected a JSON object")
query = str(payload.get("query") or payload.get("title") or "").strip()
title = str(payload.get("title") or query).strip()
anime_name = sanitize_path_component(payload.get("anime_name") or title)
if not query:
raise ValueError("Missing search query")
try:
index = int(payload.get("index", 1))
except (TypeError, ValueError) as exc:
raise ValueError("Invalid result index") from exc
if index < 1:
raise ValueError("Result index must be positive")
mode = str(payload.get("mode") or CONFIG["mode"]).lower()
quality = str(payload.get("quality") or CONFIG["quality"]).lower()
download_dir = str(payload.get("download_dir") or CONFIG["download_dir"]).strip()
if mode not in MODE_CHOICES:
raise ValueError("Mode must be sub or dub")
if quality not in QUALITY_CHOICES:
raise ValueError("Unknown quality")
return {
"id": uuid4().hex,
"title": title,
"anime_name": anime_name,
"season": normalize_season(payload.get("season")),
"query": query,
"result_index": index,
"mode": mode,
"quality": quality,
"episodes": validate_episode_spec(payload.get("episodes")),
"download_dir": str(Path(download_dir).expanduser()),
"status": "pending",
"created_at": now_iso(),
"updated_at": now_iso(),
"started_at": None,
"finished_at": None,
"exit_code": None,
"pid": None,
"log": [],
}
def command_for_job(job):
command = [
ANI_CLI,
"-d",
"-q",
job["quality"],
"-e",
job["episodes"],
"-S",
str(job["result_index"]),
]
if job["mode"] == "dub":
command.append("--dub")
command.append(job["query"])
return command
DOWNLOAD_QUEUE = DownloadQueue()
def dependency_status():
checks = ["curl", "sed", "grep", "openssl", "fzf", "aria2c", "ffmpeg", "yt-dlp"]
result = {name: bool(shutil.which(name)) for name in checks}
result["ani-cli"] = bool(shutil.which(ANI_CLI) or (Path(ANI_CLI).exists() and os.access(ANI_CLI, os.X_OK)))
return result
class Handler(BaseHTTPRequestHandler):
server_version = "AniCliWeb/1.0"
def do_GET(self):
parsed = urlparse(self.path)
try:
if parsed.path == "/":
self.html(INDEX_HTML)
elif parsed.path == "/watchlist":
self.html(WATCHLIST_HTML)
elif parsed.path == "/api/config":
self.json(CONFIG)
elif parsed.path == "/api/version":
self.json({"name": APP_NAME, "version": VERSION})
elif parsed.path == "/api/dependencies":
self.json(dependency_status())
elif parsed.path == "/api/search":
params = parse_qs(parsed.query)
query = (params.get("q") or [""])[0].strip()
mode = (params.get("mode") or [CONFIG["mode"]])[0].lower()
if not query:
raise ValueError("Search query is required")
if mode not in MODE_CHOICES:
raise ValueError("Mode must be sub or dub")
self.json({"results": search_anime(query, mode)})
elif parsed.path.startswith("/api/anime/") and parsed.path.endswith("/episodes"):
show_id = unquote(parsed.path[len("/api/anime/") : -len("/episodes")])
params = parse_qs(parsed.query)
mode = (params.get("mode") or [CONFIG["mode"]])[0].lower()
if mode not in MODE_CHOICES:
raise ValueError("Mode must be sub or dub")
self.json({"episodes": episode_list(show_id, mode)})
elif parsed.path == "/api/queue":
params = parse_qs(parsed.query)
page = (params.get("page") or ["1"])[0]
per_page = (params.get("per_page") or ["10"])[0]
self.json(DOWNLOAD_QUEUE.list(page=page, per_page=per_page))
elif parsed.path.startswith("/api/watchlist/thumb/"):
params = parse_qs(parsed.query)
show_id = unquote(parsed.path[len("/api/watchlist/thumb/") :]).strip()
force = (params.get("force") or ["0"])[0] == "1"
debug_log("thumbnail.route.request", show_id=show_id, force=force, path=parsed.path, query=parsed.query)
path = WATCHLIST.ensure_thumbnail(show_id, force=force)
if not path or not path.exists():
debug_log("thumbnail.route.miss", show_id=show_id, resolved_path=path)
self.error(HTTPStatus.NOT_FOUND, "Thumbnail not found")
return
debug_log("thumbnail.route.hit", show_id=show_id, resolved_path=path)
self.file(path)
elif parsed.path in {"/api/watchlist", "/api/watchlist/get"}:
params = parse_qs(parsed.query)
page = (params.get("page") or ["1"])[0]
per_page = (params.get("per_page") or ["30"])[0]
return self.json(get_watchlist(page=page, per_page=per_page))
else:
self.error(HTTPStatus.NOT_FOUND, "Not found")
except Exception as exc:
self.exception(exc)
def do_POST(self):
parsed = urlparse(self.path)
try:
if parsed.path == "/api/config":
self.update_config(self.body_json())
elif parsed.path == "/api/queue":
self.json(DOWNLOAD_QUEUE.add(self.body_json()), HTTPStatus.CREATED)
elif parsed.path == "/api/queue/retry-failed":
self.json(DOWNLOAD_QUEUE.retry_all_failed())
elif parsed.path == "/api/queue/clear-finished":
self.json(DOWNLOAD_QUEUE.clear_finished())
elif parsed.path.startswith("/api/queue/"):
parts = parsed.path.strip("/").split("/")
if len(parts) != 4:
self.error(HTTPStatus.NOT_FOUND, "Not found")
return
_, _, job_id, action = parts
actions = {
"retry": DOWNLOAD_QUEUE.retry,
"cancel": DOWNLOAD_QUEUE.cancel,
"remove": DOWNLOAD_QUEUE.remove,
}
if action not in actions:
self.error(HTTPStatus.NOT_FOUND, "Not found")
return
self.json(actions[action](job_id))
elif parsed.path in {"/api/watchlist", "/api/watchlist/add"}:
self.json(add_to_watchlist(self.body_json()), HTTPStatus.CREATED)
elif parsed.path == "/api/watchlist/ensure-thumbnails":
payload = self.body_json()
self.json(
WATCHLIST.ensure_thumbnails(
payload.get("show_ids") or [],
limit=payload.get("limit", 6),
force=bool(payload.get("force")),
)
)
elif parsed.path == "/api/watchlist/update-status":
payload = self.body_json()
self.json(update_watchlist_status(payload["show_id"], payload.get("mode")))
elif parsed.path == "/api/watchlist/update-all":
self.json(update_all_watchlist_statuses())
elif parsed.path == "/api/watchlist/remove":
payload = self.body_json()
self.json(remove_from_watchlist(payload["show_id"]))
elif parsed.path == "/api/watchlist/upload-thumbnail":
payload = self.body_json()
self.json(upload_watchlist_thumbnail(payload))
else:
self.error(HTTPStatus.NOT_FOUND, "Not found")
except Exception as exc:
self.exception(exc)
def update_config(self, payload):
global CONFIG
config = normalize_config(payload)
Path(config["download_dir"]).expanduser().mkdir(parents=True, exist_ok=True)
CONFIG = config
write_json(CONFIG_PATH, CONFIG)
self.json(CONFIG)
def body_json(self):
length = int(self.headers.get("Content-Length", "0") or "0")
raw = self.rfile.read(length).decode("utf-8") if length else "{}"
try:
return json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError("Invalid JSON body") from exc
def html(self, content):
data = content.encode("utf-8")
self.write_response_bytes(data, HTTPStatus.OK, {"Content-Type": "text/html; charset=utf-8"})
def file(self, path):
payload = Path(path).read_bytes()
content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
debug_log("file.serve", path=path, content_type=content_type, bytes=len(payload))
self.write_response_bytes(
payload,
HTTPStatus.OK,
{
"Content-Type": content_type,
"Cache-Control": "public, max-age=86400",
},
)
def json(self, payload, status=HTTPStatus.OK):
data = json.dumps(payload).encode("utf-8")
self.write_response_bytes(data, status, {"Content-Type": "application/json"})
def write_response_bytes(self, payload, status, headers):
try:
self.send_response(status)
for key, value in headers.items():
self.send_header(key, value)
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
except CLIENT_DISCONNECT_ERRORS:
return
def error(self, status, message):
try:
self.json({"error": message}, status)
except CLIENT_DISCONNECT_ERRORS:
return
def exception(self, exc):
if isinstance(exc, CLIENT_DISCONNECT_ERRORS):
return
if isinstance(exc, KeyError):
self.error(HTTPStatus.NOT_FOUND, str(exc).strip("'"))
elif isinstance(exc, ValueError):
self.error(HTTPStatus.BAD_REQUEST, str(exc))
else:
self.error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
def log_message(self, fmt, *args):
print(f"{self.address_string()} - {fmt % args}")
INDEX_HTML = r"""
ani-cli web
Select a result
Episodes will appear here.
Defaults
Download queue
"""
WATCHLIST_HTML = r"""
ani-cli web - watchlist
Tracked shows
Compact cards with local AnimeSchedule cover caching when artwork is available.
0Tracked shows
0Sub episodes
0Dub episodes
Watchlist library
Compact poster grid for faster scanning.
"""
def server_host():
return os.environ.get("ANI_CLI_WEB_HOST", "127.0.0.1").strip() or "127.0.0.1"
def server_port():
raw = str(os.environ.get("ANI_CLI_WEB_PORT", "8421")).strip() or "8421"
try:
port = int(raw, 10)
except ValueError as exc:
raise ValueError("ANI_CLI_WEB_PORT must be a valid integer") from exc
if not 1 <= port <= 65535:
raise ValueError("ANI_CLI_WEB_PORT must be between 1 and 65535")
return port
def parse_runtime_flags(argv):
global DEBUG_MODE
remaining = []
for arg in argv:
if arg in {"--debug", "---debug"}:
DEBUG_MODE = True
continue
remaining.append(arg)
return remaining
def main():
parse_runtime_flags(sys.argv[1:])
host = server_host()
port = server_port()
server = ThreadingHTTPServer((host, port), Handler)
debug_log("server.start", version=VERSION, host=host, port=port, argv=sys.argv[1:])
print(f"Serving {APP_NAME} {VERSION} on http://{host}:{port}/")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nShutting down server.")
finally:
server.server_close()
if __name__ == "__main__":
main()