#!/usr/bin/env python3 """HTTP handler and route wiring for ani-cli-web.""" import json import mimetypes import os import shutil import traceback from dataclasses import dataclass from http import HTTPStatus from http.server import BaseHTTPRequestHandler from pathlib import Path from urllib.parse import parse_qs, unquote, urlparse from app_support import ( ANI_CLI, APP_NAME, CLIENT_DISCONNECT_ERRORS, MAX_JSON_BODY_BYTES, MODE_CHOICES, VERSION, client_address_is_local, debug_enabled, debug_log, normalize_config, remote_access_allowed, remote_access_token_configured, remote_access_token_matches, ) @dataclass(frozen=True) class HandlerContext: add_to_watchlist: object config_html: str ensure_runtime: object episode_list: object get_config_snapshot: object get_watchlist: object get_watchlist_refresh_status: object http_error: object index_html: str remove_from_watchlist: object runtime_state: object save_runtime_config: object search_anime: object thumbnail_file_path: object update_all_watchlist_statuses: object update_watchlist_category: object update_watchlist_status: object upload_watchlist_thumbnail: object watchlist_html: str def build_handler_class(context): class ConfiguredHandler(Handler): handler_context = context return ConfiguredHandler 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 _context(self): context = getattr(self, "handler_context", None) or getattr(type(self), "handler_context", None) if context is None: raise RuntimeError("Handler context is not configured.") return context def _runtime(self): context = Handler._context(self) context.ensure_runtime(start_workers=True) return context.runtime_state() def _effective_client_host(self): peer_host = self.client_address[0] if self.client_address else "" if not client_address_is_local(peer_host): return peer_host forwarded = ( self.headers.get("X-Forwarded-For") or self.headers.get("x-forwarded-for") or self.headers.get("X-Real-IP") or self.headers.get("x-real-ip") or "" ) if forwarded: first = str(forwarded).split(",", 1)[0].strip().strip("[]") if first: return first forwarded_header = str(self.headers.get("Forwarded") or self.headers.get("forwarded") or "").strip() if forwarded_header: for part in forwarded_header.split(";"): key, _, value = part.partition("=") if key.strip().lower() != "for": continue candidate = value.strip().strip('"') if candidate.startswith("[") and "]" in candidate: candidate = candidate[1 : candidate.index("]")] elif ":" in candidate and candidate.count(":") == 1: candidate = candidate.split(":", 1)[0] candidate = candidate.strip() if candidate: return candidate return peer_host def _remote_token(self): return ( self.headers.get("X-AniCli-Web-Token") or self.headers.get("x-ani-cli-web-token") or Handler._bearer_token(self) or "" ) def _bearer_token(self): header = str(self.headers.get("Authorization") or "").strip() if not header.startswith("Bearer "): return "" return header[len("Bearer ") :].strip() def ensure_client_access(self): client_host = Handler._effective_client_host(self) if client_address_is_local(client_host): return if not remote_access_allowed(): raise PermissionError( "Remote access is disabled. Set ANI_CLI_WEB_ALLOW_REMOTE=1 to allow non-local clients." ) if not remote_access_token_configured(): raise PermissionError( "Remote access requires ANI_CLI_WEB_REMOTE_TOKEN for non-local clients." ) if not remote_access_token_matches(Handler._remote_token(self)): raise PermissionError("Remote access token missing or invalid.") def do_GET(self): parsed = urlparse(self.path) try: self.ensure_client_access() if parsed.path == "/": self.html(Handler._context(self).index_html) elif parsed.path == "/config": self.html(Handler._context(self).config_html) elif parsed.path == "/watchlist": self.html(Handler._context(self).watchlist_html) elif parsed.path == "/api/config": self.json(Handler._context(self).get_config_snapshot()) 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": runtime = Handler._runtime(self) config = runtime["config"] 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": Handler._context(self).search_anime(query, mode)}) elif parsed.path.startswith("/api/anime/") and parsed.path.endswith("/episodes"): runtime = Handler._runtime(self) config = runtime["config"] 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": Handler._context(self).episode_list(show_id, mode)}) elif parsed.path == "/api/queue": runtime = Handler._runtime(self) params = parse_qs(parsed.query) page = (params.get("page") or ["1"])[0] per_page = (params.get("per_page") or ["10"])[0] self.json(runtime["download_queue"].list(page=page, per_page=per_page)) elif parsed.path.startswith("/api/watchlist/thumb/"): runtime = Handler._runtime(self) 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 = runtime["watchlist"].get(show_id) path = Handler._context(self).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"}: Handler._runtime(self) params = parse_qs(parsed.query) page = (params.get("page") or ["1"])[0] per_page = (params.get("per_page") or ["30"])[0] category = (params.get("category") or [""])[0].strip().lower() or None self.json(Handler._context(self).get_watchlist(page=page, per_page=per_page, category=category)) elif parsed.path == "/api/watchlist/refresh-status": Handler._runtime(self) self.json(Handler._context(self).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() if parsed.path == "/api/config": self.update_config(self.body_json()) elif parsed.path == "/api/queue": runtime = Handler._runtime(self) self.json(runtime["download_queue"].add(self.body_json()), HTTPStatus.CREATED) elif parsed.path == "/api/queue/retry-failed": runtime = Handler._runtime(self) self.json(runtime["download_queue"].retry_all_failed()) elif parsed.path == "/api/queue/clear-finished": runtime = Handler._runtime(self) self.json(runtime["download_queue"].clear_finished()) elif parsed.path.startswith("/api/queue/"): runtime = Handler._runtime(self) parts = parsed.path.strip("/").split("/") if len(parts) != 4: self.error(HTTPStatus.NOT_FOUND, "Not found") return _, _, job_id, action = parts actions = { "retry": runtime["download_queue"].retry, "cancel": runtime["download_queue"].cancel, "remove": runtime["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"}: Handler._runtime(self) result = Handler._context(self).add_to_watchlist(self.body_json()) status = HTTPStatus.CREATED if result.get("created") else HTTPStatus.OK self.json(result, status) elif parsed.path == "/api/watchlist/ensure-thumbnails": runtime = Handler._runtime(self) payload = Handler.require_json_object(self, self.body_json()) self.json( runtime["watchlist"].ensure_thumbnails( Handler.optional_list_field(self, payload, "show_ids") or [], limit=payload.get("limit", 6), force=Handler.optional_bool_field(self, payload, "force", default=False), ) ) elif parsed.path == "/api/watchlist/update-status": Handler._runtime(self) payload = Handler.require_fields(self, self.body_json(), "show_id") self.json(Handler._context(self).update_watchlist_status(payload["show_id"], payload.get("mode"))) elif parsed.path == "/api/watchlist/update-category": Handler._runtime(self) payload = Handler.require_fields(self, self.body_json(), "show_id") self.json(Handler._context(self).update_watchlist_category(payload["show_id"], payload.get("category"))) elif parsed.path == "/api/watchlist/update-all": Handler._runtime(self) result = Handler._context(self).update_all_watchlist_statuses() status = HTTPStatus.ACCEPTED if result.get("created") else HTTPStatus.OK self.json(result, status) elif parsed.path == "/api/watchlist/remove": Handler._runtime(self) payload = Handler.require_fields(self, self.body_json(), "show_id") self.json(Handler._context(self).remove_from_watchlist(payload["show_id"])) elif parsed.path == "/api/watchlist/upload-thumbnail": Handler._runtime(self) payload = Handler.require_fields(self, self.body_json(), "show_id", "data_url") self.json(Handler._context(self).upload_watchlist_thumbnail(payload)) else: self.error(HTTPStatus.NOT_FOUND, "Not found") except Exception as exc: self.exception(exc) def update_config(self, payload): config = normalize_config(payload) Path(config["download_dir"]).expanduser().mkdir(parents=True, exist_ok=True) context = Handler._context(self) context.save_runtime_config(config) self.json(context.get_config_snapshot()) def require_fields(self, payload, *names): payload = Handler.require_json_object(self, payload) missing = [name for name in names if not str(payload.get(name) or "").strip()] if missing: if len(missing) == 1: raise ValueError(f"Missing required field: {missing[0]}") raise ValueError(f"Missing required fields: {', '.join(missing)}") return payload def require_json_object(self, payload): if not isinstance(payload, dict): raise ValueError("Expected a JSON object") return payload def optional_list_field(self, payload, name): payload = Handler.require_json_object(self, payload) if name not in payload or payload.get(name) is None: return None value = payload.get(name) if not isinstance(value, list): raise ValueError(f"Field '{name}' must be a JSON array") return value def optional_bool_field(self, payload, name, default=False): payload = Handler.require_json_object(self, payload) if name not in payload or payload.get(name) is None: return default value = payload.get(name) if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False raise ValueError(f"Field '{name}' must be a boolean") 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 Handler._context(self).http_error(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 Handler._context(self).http_error(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, Handler._context(self).http_error): 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: if debug_enabled(): debug_log("handler.exception", path=getattr(self, "path", ""), error=exc) traceback.print_exc() else: debug_log("handler.exception", path=getattr(self, "path", ""), error=exc) self.error(HTTPStatus.INTERNAL_SERVER_ERROR, "Internal server error.") 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