459 lines
15 KiB
Python
459 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""Shared constants and helper utilities for ani-cli-web."""
|
|
|
|
import hmac
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
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.31.1"
|
|
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"}
|
|
WATCHLIST_CATEGORY_LABELS = {
|
|
"watching": "Watching",
|
|
"planned": "Planned",
|
|
"finished": "Finished",
|
|
"dropped": "Dropped",
|
|
}
|
|
WATCHLIST_CATEGORY_CHOICES = tuple(WATCHLIST_CATEGORY_LABELS.keys())
|
|
MAX_LOG_LINES = 240
|
|
MAX_JSON_BODY_BYTES = 12 * 1024 * 1024
|
|
WATCHLIST_REFRESH_REQUEST_TIMEOUT_SECONDS = 8
|
|
WORKER_SHUTDOWN_TIMEOUT_SECONDS = 10
|
|
WATCHLIST_AUTO_REFRESH_DEFAULT_MINUTES = 60
|
|
WATCHLIST_AUTO_REFRESH_MIN_MINUTES = 1
|
|
WATCHLIST_AUTO_REFRESH_MAX_MINUTES = 7 * 24 * 60
|
|
|
|
|
|
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"}
|
|
|
|
|
|
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 remote_access_token():
|
|
return str(os.environ.get("ANI_CLI_WEB_REMOTE_TOKEN") or "").strip()
|
|
|
|
|
|
def remote_access_token_configured():
|
|
return bool(remote_access_token())
|
|
|
|
|
|
def remote_access_token_matches(value):
|
|
provided = str(value or "").strip()
|
|
expected = remote_access_token()
|
|
return bool(expected and provided and hmac.compare_digest(provided, expected))
|
|
|
|
|
|
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"),
|
|
"watchlist_auto_refresh_enabled": False,
|
|
"watchlist_auto_refresh_minutes": WATCHLIST_AUTO_REFRESH_DEFAULT_MINUTES,
|
|
}
|
|
|
|
|
|
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,
|
|
"title_index": None,
|
|
"token_index": None,
|
|
"match_cache": {},
|
|
}
|
|
|
|
|
|
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")
|
|
|
|
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"])
|
|
auto_refresh_enabled = config.get("watchlist_auto_refresh_enabled")
|
|
auto_refresh_minutes = config.get("watchlist_auto_refresh_minutes")
|
|
|
|
if isinstance(auto_refresh_enabled, str):
|
|
auto_refresh_enabled = auto_refresh_enabled.strip().lower() in {"1", "true", "yes", "on"}
|
|
else:
|
|
auto_refresh_enabled = bool(auto_refresh_enabled)
|
|
|
|
try:
|
|
auto_refresh_minutes = int(str(auto_refresh_minutes).strip() or WATCHLIST_AUTO_REFRESH_DEFAULT_MINUTES)
|
|
except (TypeError, ValueError):
|
|
auto_refresh_minutes = WATCHLIST_AUTO_REFRESH_DEFAULT_MINUTES
|
|
auto_refresh_minutes = max(
|
|
WATCHLIST_AUTO_REFRESH_MIN_MINUTES,
|
|
min(WATCHLIST_AUTO_REFRESH_MAX_MINUTES, auto_refresh_minutes),
|
|
)
|
|
|
|
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())
|
|
config["watchlist_auto_refresh_enabled"] = auto_refresh_enabled
|
|
config["watchlist_auto_refresh_minutes"] = auto_refresh_minutes
|
|
return config
|
|
|
|
|
|
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, config):
|
|
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,
|
|
"show_id": str(payload.get("show_id") or "").strip(),
|
|
"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
|