51 lines
2.2 KiB
Python
51 lines
2.2 KiB
Python
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) -> 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")
|
|
if str(item.get("category") or "") != category:
|
|
return Readiness("ignored", "SAB category is not owned by Importarr")
|
|
status = str(item.get("status") or "")
|
|
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'}")
|
|
storage_value = str(item.get("storage") or "")
|
|
if not storage_value:
|
|
return Readiness("unknown", "SAB completed item has no final storage")
|
|
storage = Path(storage_value).resolve()
|
|
root = download_root.resolve()
|
|
if storage != root and root not in storage.parents:
|
|
return Readiness("ignored", "SAB storage is outside configured download root", storage)
|
|
if has_transient_part(storage):
|
|
return Readiness("processing", "SAB storage path contains transient unpack/admin marker", storage)
|
|
reason = "forced despite SAB status" if force_status and status != "Completed" else "SAB completed in owned category with final storage"
|
|
return Readiness("ready", reason, storage)
|