#!/usr/bin/env python3 import base64 import gzip import ipaddress 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 from web_templates import CONFIG_HTML, INDEX_HTML, PAGE_LINKS, WATCHLIST_HTML, render_page_links, render_sidebar_brand 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.12.0" 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 RUNTIME_LOCK = threading.RLock() QUALITY_CHOICES = {"best", "1080", "1080p", "720", "720p", "480", "480p", "360", "360p", "worst"} MODE_CHOICES = {"sub", "dub"} MAX_LOG_LINES = 240 MAX_JSON_BODY_BYTES = 12 * 1024 * 1024 def env_flag(name, default=False): raw = os.environ.get(name) if raw is None: return default return str(raw).strip().lower() in {"1", "true", "yes", "on"} class HttpError(Exception): def __init__(self, status, message): super().__init__(message) self.status = status self.message = message def debug_enabled(): return DEBUG_MODE or env_flag("ANI_CLI_WEB_DEBUG") def remote_access_allowed(): return env_flag("ANI_CLI_WEB_ALLOW_REMOTE") def background_workers_enabled(): return not env_flag("ANI_CLI_WEB_DISABLE_WORKER") def client_address_is_local(host): text = str(host or "").strip() if not text: return False if text == "localhost": return True try: return ipaddress.ip_address(text).is_loopback except ValueError: return False 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() def project_app_root(): configured = str(os.environ.get("ANI_CLI_WEB_STATE_ROOT") or "").strip() if configured: return Path(configured).expanduser() return PROJECT_ROOT / ".ani-cli-web" PROJECT_APP_ROOT = project_app_root() def ensure_project_app_root(): PROJECT_APP_ROOT.mkdir(parents=True, exist_ok=True) return PROJECT_APP_ROOT 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): return PROJECT_APP_ROOT.joinpath(*parts) def project_state_dir(*parts): return PROJECT_APP_ROOT.joinpath(*parts) 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") STATE_DB_PATH = project_state_path("state.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(): ensure_project_app_root() migrate_legacy_file(CONFIG_PATH, LEGACY_CONFIG_PATH, "config.json") migrate_legacy_file(QUEUE_PATH, LEGACY_QUEUE_PATH, "queue.json") migrate_legacy_file(STATE_DB_PATH, PROJECT_APP_ROOT / "queue.sqlite3", "queue.sqlite3") migrate_legacy_file(STATE_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): path.parent.mkdir(parents=True, exist_ok=True) 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 = None WATCHLIST = None DOWNLOAD_QUEUE = None WATCHLIST_REFRESH = None 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 ANIDB_TITLES_PATH.parent.mkdir(parents=True, exist_ok=True) 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(STATE_DB_PATH, timeout=30) conn.row_factory = sqlite3.Row return conn def _init_db(self): STATE_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 all_show_ids(self): with self.lock, self._connect() as conn: rows = conn.execute("SELECT show_id FROM watchlist ORDER BY title COLLATE NOCASE ASC").fetchall() return [str(row["show_id"]).strip() for row in rows if str(row["show_id"]).strip()] 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): refreshed = [self.refresh(show_id) for show_id in self.all_show_ids()] 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 owns_event = False else: wait_event = threading.Event() self.thumbnail_events[show_id] = wait_event owns_event = True 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: 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"], ), ) 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) finally: if owns_event: with self.lock: inflight = self.thumbnail_events.get(show_id) if inflight is wait_event: inflight.set() self.thumbnail_events.pop(show_id, None) 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} def add_to_watchlist(payload): ensure_runtime() return WATCHLIST.add(payload) def get_watchlist(page=1, per_page=30): ensure_runtime() return WATCHLIST.list(page=page, per_page=per_page) def update_watchlist_status(show_id, _mode=None): ensure_runtime() item = WATCHLIST.refresh(show_id) return {"message": f"Updated {item['title']}.", "item": item} def update_all_watchlist_statuses(): ensure_runtime() return WATCHLIST_REFRESH.start() def get_watchlist_refresh_status(): ensure_runtime() return WATCHLIST_REFRESH.status() def remove_from_watchlist(show_id): ensure_runtime() return WATCHLIST.remove(show_id) def upload_watchlist_thumbnail(payload): ensure_runtime() return WATCHLIST.set_manual_thumbnail(payload["show_id"], payload.get("filename"), payload.get("data_url")) class WatchlistRefreshJobs: def __init__(self, store, start_worker=True): self.store = store self.lock = threading.RLock() self.wakeup = threading.Event() self.pending = [] self.jobs = [] self.current_job_id = None self._init_db() self._restore_interrupted_jobs() self._load_jobs() self.worker = None if start_worker: self.ensure_worker() def _connect(self): conn = sqlite3.connect(STATE_DB_PATH, timeout=30) conn.row_factory = sqlite3.Row return conn def ensure_worker(self): with self.lock: if self.worker is not None and self.worker.is_alive(): return self.worker = threading.Thread(target=self._run, daemon=True) self.worker.start() def _init_db(self): STATE_DB_PATH.parent.mkdir(parents=True, exist_ok=True) with self._connect() as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS watchlist_refresh_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_watchlist_refresh_created_at ON watchlist_refresh_jobs(created_at)") conn.execute("CREATE INDEX IF NOT EXISTS idx_watchlist_refresh_status ON watchlist_refresh_jobs(status)") 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): conn.execute( """ INSERT INTO watchlist_refresh_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 """, ( job["id"], job["status"], job["created_at"], job["updated_at"], json.dumps(job, sort_keys=True), ), ) def _trim_history_conn(self, conn): rows = conn.execute( "SELECT id FROM watchlist_refresh_jobs ORDER BY created_at DESC LIMIT -1 OFFSET 10" ).fetchall() if rows: conn.executemany("DELETE FROM watchlist_refresh_jobs WHERE id = ?", [(row["id"],) for row in rows]) def _save_job_locked(self, job): with self._connect() as conn: self._upsert_job_conn(conn, job) self._trim_history_conn(conn) def _load_jobs(self): with self.lock, self._connect() as conn: rows = conn.execute( "SELECT payload FROM watchlist_refresh_jobs ORDER BY created_at DESC LIMIT 10" ).fetchall() self.jobs = [self._job_from_row(row) for row in rows] def _restore_interrupted_jobs(self): with self.lock, self._connect() as conn: rows = conn.execute( "SELECT payload FROM watchlist_refresh_jobs WHERE status IN ('pending', 'running')" ).fetchall() for row in rows: job = self._job_from_row(row) job["status"] = "interrupted" job["finished_at"] = now_iso() job["updated_at"] = job["finished_at"] job["current_title"] = None job["message"] = "Interrupted by web app restart." self._upsert_job_conn(conn, job) self._trim_history_conn(conn) def _copy_job(self, job): return json.loads(json.dumps(job)) def _find_locked(self, job_id): for job in self.jobs: if job["id"] == job_id: return job return None def status(self): with self.lock: job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None) if job is None: job = self.jobs[0] if self.jobs else None active = bool(job and job["status"] in {"pending", "running"}) return {"job": self._copy_job(job) if job else None, "running": active} def start(self): with self.lock: job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None) if job is not None: return { "message": "Watchlist refresh already queued." if job["status"] == "pending" else "Watchlist refresh already running.", "job": self._copy_job(job) if job else None, "created": False, } now = now_iso() job = { "id": uuid4().hex, "status": "pending", "created_at": now, "updated_at": now, "started_at": None, "finished_at": None, "total": 0, "completed": 0, "errors": 0, "current_title": None, "message": "Queued watchlist refresh.", } self.jobs.insert(0, job) self.jobs = self.jobs[:10] self._save_job_locked(job) self.pending.append(job["id"]) self.wakeup.set() return {"message": "Watchlist refresh started.", "job": self._copy_job(job), "created": True} def _next_pending(self): with self.lock: while self.pending: job_id = self.pending.pop(0) job = self._find_locked(job_id) if job is not None: self.current_job_id = job_id return job return None 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): try: show_ids = self.store.all_show_ids() started_at = now_iso() with self.lock: job["status"] = "running" job["started_at"] = started_at job["updated_at"] = started_at job["total"] = len(show_ids) job["completed"] = 0 job["errors"] = 0 job["current_title"] = None job["message"] = f"Refreshing 0/{len(show_ids)} watchlist entries." self._save_job_locked(job) for index, show_id in enumerate(show_ids, start=1): title = show_id errored = False try: item = self.store.refresh(show_id) title = item.get("title") or title errored = item.get("status") == "error" except Exception: errored = True with self.lock: job["completed"] = index job["errors"] += 1 if errored else 0 job["current_title"] = title job["updated_at"] = now_iso() job["message"] = f"Refreshing {index}/{len(show_ids)}: {title}" self._save_job_locked(job) finished_at = now_iso() with self.lock: job["status"] = "done" job["finished_at"] = finished_at job["updated_at"] = finished_at job["current_title"] = None if job["errors"]: job["message"] = ( f"Refreshed {job['completed']} watchlist entries with {job['errors']} refresh errors." ) else: job["message"] = f"Refreshed {job['completed']} watchlist entries." self._save_job_locked(job) except Exception as exc: finished_at = now_iso() with self.lock: job["status"] = "failed" job["finished_at"] = finished_at job["updated_at"] = finished_at job["current_title"] = None job["message"] = f"Watchlist refresh failed: {exc}" self._save_job_locked(job) finally: with self.lock: self.current_job_id = None class DownloadQueue: def __init__(self, start_worker=True): self.lock = threading.RLock() self.wakeup = threading.Event() self.current_process = None self.current_job_id = None self.current_job = None self._init_db() self._migrate_json_queue() self._restore_interrupted_jobs() self.worker = None if start_worker: self.ensure_worker() def _connect(self): conn = sqlite3.connect(STATE_DB_PATH, timeout=30) conn.row_factory = sqlite3.Row return conn def ensure_worker(self): with self.lock: if self.worker is not None and self.worker.is_alive(): return self.worker = threading.Thread(target=self._run, daemon=True) self.worker.start() def _init_db(self): STATE_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() updated_at = now_iso() job["cancel_requested"] = True job["updated_at"] = updated_at if self.current_job is not None and self.current_job.get("id") == job_id: self.current_job["cancel_requested"] = True self.current_job["updated_at"] = updated_at 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"] self.current_job = job 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.current_job = 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 def build_runtime(start_workers=None): migrate_legacy_state_storage() worker_state = background_workers_enabled() if start_workers is None else bool(start_workers) config = normalize_config(load_json(CONFIG_PATH, DEFAULT_CONFIG)) write_json(CONFIG_PATH, config) watchlist = WatchlistStore() download_queue = DownloadQueue(start_worker=worker_state) watchlist_refresh = WatchlistRefreshJobs(watchlist, start_worker=worker_state) return { "config": config, "watchlist": watchlist, "download_queue": download_queue, "watchlist_refresh": watchlist_refresh, } def initialize_runtime(start_workers=None): global CONFIG, WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH with RUNTIME_LOCK: if CONFIG is not None and WATCHLIST is not None and DOWNLOAD_QUEUE is not None and WATCHLIST_REFRESH is not None: if start_workers: DOWNLOAD_QUEUE.ensure_worker() WATCHLIST_REFRESH.ensure_worker() return { "config": CONFIG, "watchlist": WATCHLIST, "download_queue": DOWNLOAD_QUEUE, "watchlist_refresh": WATCHLIST_REFRESH, } runtime = build_runtime(start_workers=start_workers) CONFIG = runtime["config"] WATCHLIST = runtime["watchlist"] DOWNLOAD_QUEUE = runtime["download_queue"] WATCHLIST_REFRESH = runtime["watchlist_refresh"] return runtime def ensure_runtime(start_workers=False): if CONFIG is None or WATCHLIST is None or DOWNLOAD_QUEUE is None or WATCHLIST_REFRESH is None: return initialize_runtime(start_workers=start_workers) return { "config": CONFIG, "watchlist": WATCHLIST, "download_queue": DOWNLOAD_QUEUE, "watchlist_refresh": WATCHLIST_REFRESH, } 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 ensure_client_access(self): if remote_access_allowed(): return client_host = self.client_address[0] if self.client_address else "" if client_address_is_local(client_host): return raise PermissionError( "Remote access is disabled. Set ANI_CLI_WEB_ALLOW_REMOTE=1 to allow non-local clients." ) def do_GET(self): parsed = urlparse(self.path) try: self.ensure_client_access() ensure_runtime() if parsed.path == "/": self.html(INDEX_HTML) elif parsed.path == "/config": self.html(CONFIG_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/"): show_id = unquote(parsed.path[len("/api/watchlist/thumb/") :]).strip() debug_log("thumbnail.route.request", show_id=show_id, path=parsed.path, query=parsed.query) item = WATCHLIST.get(show_id) path = thumbnail_file_path(item.get("thumbnail_path")) 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)) elif parsed.path == "/api/watchlist/refresh-status": self.json(get_watchlist_refresh_status()) else: self.error(HTTPStatus.NOT_FOUND, "Not found") except Exception as exc: self.exception(exc) def do_POST(self): parsed = urlparse(self.path) try: self.ensure_client_access() ensure_runtime() 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": result = update_all_watchlist_statuses() status = HTTPStatus.ACCEPTED if result.get("created") else HTTPStatus.OK self.json(result, status) 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): try: length = int(self.headers.get("Content-Length", "0") or "0") except ValueError as exc: raise ValueError("Invalid Content-Length header") from exc if length < 0: raise ValueError("Invalid Content-Length header") if length > MAX_JSON_BODY_BYTES: raise HttpError(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "Request body too large.") raw_bytes = self.rfile.read(length) if length else b"{}" if len(raw_bytes) > MAX_JSON_BODY_BYTES: raise HttpError(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "Request body too large.") try: raw = raw_bytes.decode("utf-8") except UnicodeDecodeError as exc: raise ValueError("Request body must be valid UTF-8 JSON.") from exc 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, HttpError): self.error(exc.status, exc.message) elif isinstance(exc, KeyError): self.error(HTTPStatus.NOT_FOUND, str(exc).strip("'")) elif isinstance(exc, PermissionError): self.error(HTTPStatus.FORBIDDEN, str(exc)) 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}") 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:]) initialize_runtime(start_workers=True) 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()