from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any TRANSIENT_PARTS = {"_UNPACK_", "__UNPACK__", "_FAILED_", "_ADMIN_"} NOT_READY_STATUSES = { "Queued", "QuickCheck", "Verifying", "Repairing", "Fetching", "Extracting", "Moving", "Running", "Downloading", "Paused", "Propagating", } @dataclass(frozen=True) class Readiness: state: str reason: str storage: Path | None = None @property def ready(self) -> bool: return self.state == "ready" def has_transient_part(path: Path) -> bool: return any(part in TRANSIENT_PARTS or any(token in part for token in TRANSIENT_PARTS) for part in path.parts) def classify_history_item(item: dict[str, Any], active_nzo_ids: set[str], category: str, download_root: Path, force_status: bool = False, sab_storage_root: Path | None = None) -> Readiness: nzo_id = str(item.get("nzo_id") or item.get("nzoid") or "") if not force_status and nzo_id and nzo_id in active_nzo_ids: return Readiness("processing", "SAB job is still present in queue") item_category = str(item.get("category") or item.get("cat") or "") status = str(item.get("status") or "") storage_value = str(item.get("storage") or "") storage = Path(storage_value).resolve() if storage_value else None root = download_root.resolve() sab_root = (sab_storage_root or download_root).resolve() storage_in_local_root = bool(storage and (storage == root or root in storage.parents)) storage_in_sab_root = bool(storage and (storage == sab_root or sab_root in storage.parents)) storage_in_root = storage_in_local_root or storage_in_sab_root if item_category != category and not storage_in_root: return Readiness("ignored", "SAB category/storage is not owned by Importarr", storage) if not force_status and status == "Failed": return Readiness("failed", "SAB history reports failure") if not force_status and (status in NOT_READY_STATUSES or status != "Completed"): return Readiness("processing", f"SAB status is {status or 'unknown'}") if storage is None: return Readiness("unknown", "SAB completed item has no final storage") if not storage_in_root: return Readiness("ignored", "SAB storage is outside configured download root", storage) if storage_in_sab_root and not storage_in_local_root: storage = root / storage.relative_to(sab_root) if has_transient_part(storage): return Readiness("processing", "SAB storage path contains transient unpack/admin marker", storage) if force_status and status != "Completed": reason = "forced despite SAB status" elif item_category != category: reason = "SAB completed inside Importarr download root" else: reason = "SAB completed in owned category with final storage" return Readiness("ready", reason, storage)