daniels
f9eb633e19
Only treat configured manual storage as Importarr-owned and map SAB container paths to local paths.
63 lines
2.9 KiB
Python
63 lines
2.9 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, 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)
|