Initial productized Importarr service

importarr/readiness.py

@@ -0,0 +1,49 @@
+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) -> Readiness:
+    nzo_id = str(item.get("nzo_id") or item.get("nzoid") or "")
+    if 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 status == "Failed":
+        return Readiness("failed", "SAB history reports failure")
+    if 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)
+    return Readiness("ready", "SAB completed in owned category with final storage", storage)