Add anipy-cli download fallback

Add an optional config toggle that retries failed ani-cli downloads with anipy-cli, expose fallback runtime status in the config page, and update release docs and tests for the new behavior.
This commit is contained in:
Dymas
2026-07-09 14:11:40 +02:00
parent 332ec670fc
commit d8ed64bd64
8 changed files with 319 additions and 61 deletions
+38 -7
View File
@@ -23,6 +23,7 @@ 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"
ANIPY_CLI = os.environ.get("ANIPY_CLI_BIN") or shutil.which("anipy-cli") or "anipy-cli"
APP_NAME = "ani-cli-web"
VERSION_FILE = PROJECT_ROOT / "VERSION"
CHANGELOG_FILE = PROJECT_ROOT / "CHANGELOG.md"
@@ -290,6 +291,7 @@ DEFAULT_CONFIG = {
"download_dir": default_download_dir(),
"mode": os.environ.get("ANI_CLI_MODE", "sub"),
"quality": os.environ.get("ANI_CLI_QUALITY", "best"),
"anipy_cli_fallback_enabled": False,
"watchlist_auto_refresh_enabled": False,
"watchlist_auto_refresh_minutes": WATCHLIST_AUTO_REFRESH_DEFAULT_MINUTES,
"watchlist_refresh_delay_seconds": WATCHLIST_REFRESH_DELAY_DEFAULT_SECONDS,
@@ -445,6 +447,7 @@ def normalize_config(data):
if quality.endswith("p") and quality[:-1].isdigit():
quality = quality[:-1]
download_dir = str(config.get("download_dir") or DEFAULT_CONFIG["download_dir"])
anipy_cli_fallback_enabled = config.get("anipy_cli_fallback_enabled")
auto_refresh_enabled = config.get("watchlist_auto_refresh_enabled")
auto_refresh_minutes = config.get("watchlist_auto_refresh_minutes")
refresh_delay_seconds = config.get("watchlist_refresh_delay_seconds")
@@ -461,6 +464,10 @@ def normalize_config(data):
auth_username = str(config.get("auth_username") or "").strip()
auth_password = str(config.get("auth_password") or "").strip()
if isinstance(anipy_cli_fallback_enabled, str):
anipy_cli_fallback_enabled = anipy_cli_fallback_enabled.strip().lower() in {"1", "true", "yes", "on"}
else:
anipy_cli_fallback_enabled = bool(anipy_cli_fallback_enabled)
if isinstance(auto_refresh_enabled, str):
auto_refresh_enabled = auto_refresh_enabled.strip().lower() in {"1", "true", "yes", "on"}
else:
@@ -494,6 +501,7 @@ def normalize_config(data):
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["anipy_cli_fallback_enabled"] = anipy_cli_fallback_enabled
config["watchlist_auto_refresh_enabled"] = auto_refresh_enabled
config["watchlist_auto_refresh_minutes"] = auto_refresh_minutes
config["watchlist_refresh_delay_seconds"] = refresh_delay_seconds
@@ -961,6 +969,13 @@ def strip_control(value):
return re.sub(r"\x1b\[[0-9;?]*[A-Za-z]", "", str(value)).replace("\r", "\n")
def cli_executable_exists(command):
text = str(command or "").strip()
if not text:
return False
return bool(shutil.which(text) or (Path(text).exists() and os.access(text, os.X_OK)))
def printable_command(command):
return " ".join(sh_quote(part) for part in command)
@@ -1081,8 +1096,8 @@ def finalize_library_files(job):
episode_offset = normalize_episode_offset(job.get("episode_offset"))
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),
[path for path in staging_dir.rglob("*") if path.is_file()],
key=lambda path: (path.stat().st_mtime, str(path.relative_to(staging_dir)).lower()),
)
if not files:
raise RuntimeError("No downloaded files were found in the staging folder")
@@ -1103,10 +1118,7 @@ def finalize_library_files(job):
shutil.move(str(source), str(destination))
moved.append(str(destination))
try:
staging_dir.rmdir()
except OSError:
pass
shutil.rmtree(staging_dir, ignore_errors=True)
return moved
@@ -1153,7 +1165,26 @@ def build_job(payload, config):
}
def command_for_job(job):
def anipy_search_spec(job):
query = re.sub(r"\s+", " ", str(job.get("query") or "").replace(":", " ")).strip()
mode = "dub" if str(job.get("mode") or "").strip().lower() == "dub" else "sub"
return f"{query}:{job['episodes']}:{mode}"
def command_for_job(job, backend="ani-cli", download_path=None):
normalized_backend = str(backend or "ani-cli").strip().lower()
if normalized_backend == "anipy-cli":
command = [
ANIPY_CLI,
"-D",
"-s",
anipy_search_spec(job),
"-q",
job["quality"],
]
if download_path:
command.extend(["-l", str(download_path)])
return command
command = [
ANI_CLI,
"-d",