Automate repair completion confirmation

2026-09-01 22:37:23 +12:00
parent ded794a819
commit 0b59289a2e
4 changed files with 413 additions and 6 deletions
@@ -12,15 +12,20 @@ from ..db import (
    add_portal_item_activity,
    get_portal_item,
    get_user_by_username,
+    list_portal_item_activity,
    list_portal_items,
    update_portal_item,
 )
+from ..clients.jellyfin import JellyfinClient
+from ..clients.sonarr import SonarrClient
 from ..runtime import get_runtime_settings
 from .invite_email import resolve_user_delivery_email, send_generic_email
+from .snapshot import build_snapshot


 logger = logging.getLogger(__name__)
 _SYSTEM_USER = "Magent"
+_MEDIA_REPAIR_STARTED_EVENTS = {"replacement_started", "missing_search_started"}


 def _now() -> datetime:
@@ -111,6 +116,157 @@ def _activity(
    )


+def _activity_metadata(entry: Dict[str, Any]) -> Dict[str, Any]:
+    raw = entry.get("metadata_json")
+    if not isinstance(raw, str) or not raw.strip():
+        return {}
+    try:
+        parsed = json.loads(raw)
+    except (TypeError, ValueError):
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
+def _repair_tracking(item_id: int) -> tuple[Dict[str, Any], list[Dict[str, Any]]]:
+    activity = list_portal_item_activity(item_id, limit=500)
+    for entry in reversed(activity):
+        if str(entry.get("event_type") or "") not in _MEDIA_REPAIR_STARTED_EVENTS:
+            continue
+        tracking = _activity_metadata(entry).get("repairTracking")
+        if isinstance(tracking, dict):
+            return dict(tracking), activity
+    return {}, activity
+
+
+def _positive_ints(value: Any) -> list[int]:
+    if not isinstance(value, list):
+        return []
+    return [
+        int(item)
+        for item in value
+        if isinstance(item, int) and not isinstance(item, bool) and item > 0
+    ]
+
+
+def _media_signature(item: Any) -> Dict[str, str]:
+    if not isinstance(item, dict):
+        return {}
+    result: Dict[str, str] = {}
+    for key in ("Id", "Etag", "Path", "DateCreated", "MediaSources"):
+        value = item.get(key)
+        if isinstance(value, (dict, list)):
+            result[key] = json.dumps(value, separators=(",", ":"), sort_keys=True)
+        elif value is not None and str(value).strip():
+            result[key] = str(value).strip()
+    return result
+
+
+def _signature_changed(current: Dict[str, str], baseline: Dict[str, Any]) -> bool:
+    previous = _media_signature(baseline)
+    if not previous:
+        return True
+    return any(current.get(key) and current.get(key) != value for key, value in previous.items())
+
+
+async def _media_repair_evidence(tracking: Dict[str, Any]) -> Dict[str, Any]:
+    request_id = str(tracking.get("requestId") or "").strip()
+    action_id = str(tracking.get("actionId") or "").strip()
+    media_type = str(tracking.get("mediaType") or "").strip().lower()
+    collector_id = tracking.get("collectorId")
+    if not request_id.isdigit() or media_type not in {"movie", "tv"} or not isinstance(collector_id, int):
+        return {"complete": False, "phase": "invalid", "message": "Repair tracking information is incomplete."}
+
+    snapshot = await build_snapshot(request_id)
+    raw = snapshot.raw if isinstance(snapshot.raw, dict) else {}
+    arr = raw.get("arr") if isinstance(raw.get("arr"), dict) else {}
+    arr_item = arr.get("item") if isinstance(arr, dict) else None
+    jellyfin = raw.get("jellyfin") if isinstance(raw.get("jellyfin"), dict) else {}
+    jellyfin_item = jellyfin.get("item") if isinstance(jellyfin, dict) else None
+    original_file_ids = set(_positive_ints(tracking.get("originalFileIds")))
+    baselines = tracking.get("jellyfinBaseline")
+    baselines = [item for item in baselines if isinstance(item, dict)] if isinstance(baselines, list) else []
+    found_at_start = tracking.get("jellyfinFoundAtStart") is True
+
+    if media_type == "movie":
+        movie_file = arr_item.get("movieFile") if isinstance(arr_item, dict) else None
+        current_file_id = movie_file.get("id") if isinstance(movie_file, dict) else None
+        imported = isinstance(current_file_id, int) and current_file_id > 0 and current_file_id not in original_file_ids
+        if not imported:
+            return {"complete": False, "phase": "collecting", "message": "Waiting for Radarr to import the repaired movie file."}
+        if not jellyfin.get("found") or not isinstance(jellyfin_item, dict):
+            return {"complete": False, "phase": "indexing", "message": "Radarr imported the repaired movie. Waiting for Jellyfin to index it."}
+        current_signature = _media_signature(jellyfin_item)
+        if action_id == "replace_media" and found_at_start:
+            if not baselines or not _signature_changed(current_signature, baselines[0]):
+                return {"complete": False, "phase": "indexing", "message": "Radarr imported the repaired movie. Waiting for Jellyfin to refresh the existing title."}
+        return {"complete": True, "phase": "complete", "message": "Radarr imported the repaired movie and Jellyfin has indexed the updated file."}
+
+    target_rows = tracking.get("episodes")
+    targets = [item for item in target_rows if isinstance(item, dict)] if isinstance(target_rows, list) else []
+    target_ids = {
+        int(item["id"])
+        for item in targets
+        if isinstance(item.get("id"), int) and int(item["id"]) > 0
+    }
+    target_pairs = {
+        (int(item["seasonNumber"]), int(item["episodeNumber"]))
+        for item in targets
+        if isinstance(item.get("seasonNumber"), int) and isinstance(item.get("episodeNumber"), int)
+    }
+    if not target_ids or not target_pairs:
+        return {"complete": False, "phase": "invalid", "message": "No exact Sonarr episodes were recorded for this repair."}
+
+    runtime = get_runtime_settings()
+    sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key)
+    episodes = await sonarr.get_episodes(collector_id)
+    episode_map = {
+        int(item["id"]): item
+        for item in episodes
+        if isinstance(item, dict) and isinstance(item.get("id"), int)
+    } if isinstance(episodes, list) else {}
+    imported = all(
+        episode_id in episode_map
+        and (
+            episode_map[episode_id].get("hasFile") is True
+            or (
+                isinstance(episode_map[episode_id].get("episodeFileId"), int)
+                and episode_map[episode_id]["episodeFileId"] > 0
+            )
+        )
+        and episode_map[episode_id].get("episodeFileId") not in original_file_ids
+        for episode_id in target_ids
+    )
+    if not imported:
+        return {"complete": False, "phase": "collecting", "message": "Waiting for Sonarr to import every repaired episode."}
+
+    jellyfin_series_id = jellyfin_item.get("Id") if isinstance(jellyfin_item, dict) else None
+    jellyfin_client = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
+    if not jellyfin.get("found") or not jellyfin_series_id or not jellyfin_client.configured():
+        return {"complete": False, "phase": "indexing", "message": "Sonarr imported the repaired episodes. Waiting for Jellyfin to index them."}
+    jellyfin_episodes = await jellyfin_client.get_series_episodes(str(jellyfin_series_id))
+    current_by_pair = {
+        (int(item["ParentIndexNumber"]), int(item["IndexNumber"])): item
+        for item in jellyfin_episodes
+        if isinstance(item.get("ParentIndexNumber"), int) and isinstance(item.get("IndexNumber"), int)
+    }
+    if not all(pair in current_by_pair for pair in target_pairs):
+        return {"complete": False, "phase": "indexing", "message": "Sonarr imported the repaired episodes. Waiting for the exact episodes to appear in Jellyfin."}
+    if action_id == "replace_media" and found_at_start:
+        baseline_by_pair = {
+            (int(item["seasonNumber"]), int(item["episodeNumber"])): item
+            for item in baselines
+            if isinstance(item.get("seasonNumber"), int) and isinstance(item.get("episodeNumber"), int)
+        }
+        if any(pair not in baseline_by_pair for pair in target_pairs):
+            return {"complete": False, "phase": "indexing", "message": "Sonarr imported the repaired episodes. Waiting for Jellyfin to confirm the existing entries changed."}
+        if not all(
+            _signature_changed(_media_signature(current_by_pair[pair]), baseline_by_pair[pair])
+            for pair in target_pairs
+        ):
+            return {"complete": False, "phase": "indexing", "message": "Sonarr imported the repaired episodes. Waiting for Jellyfin to refresh every affected episode."}
+    return {"complete": True, "phase": "complete", "message": "Sonarr imported every repaired episode and Jellyfin has indexed the updated files."}
+
+
 def _close_issue(
    item: Dict[str, Any],
    *,
@@ -333,6 +489,54 @@ def respond_to_issue_confirmation(
    return updated


+async def process_active_media_repairs() -> Dict[str, int]:
+    items = list_portal_items(kind="issue", status="in_progress", limit=500)
+    result = {"checked": 0, "waiting": 0, "completed": 0, "failed": 0}
+    for item in items:
+        tracking, activity = _repair_tracking(int(item["id"]))
+        if not tracking:
+            continue
+        result["checked"] += 1
+        try:
+            evidence = await _media_repair_evidence(tracking)
+            if evidence.get("complete"):
+                _activity(
+                    int(item["id"]),
+                    "repair_verified",
+                    str(evidence.get("message") or "Magent verified the repaired media in Jellyfin."),
+                    metadata={
+                        "requestId": tracking.get("requestId"),
+                        "actionId": tracking.get("actionId"),
+                    },
+                )
+                await begin_issue_confirmation(
+                    int(item["id"]),
+                    actor_username=_SYSTEM_USER,
+                    actor_role="system",
+                )
+                result["completed"] += 1
+                continue
+
+            result["waiting"] += 1
+            if evidence.get("phase") == "indexing" and not any(
+                str(entry.get("event_type") or "") == "repair_imported"
+                for entry in activity
+            ):
+                _activity(
+                    int(item["id"]),
+                    "repair_imported",
+                    str(evidence.get("message") or "The repaired file was imported and is waiting for Jellyfin."),
+                    metadata={
+                        "requestId": tracking.get("requestId"),
+                        "actionId": tracking.get("actionId"),
+                    },
+                )
+        except Exception:
+            result["failed"] += 1
+            logger.exception("automatic media repair check failed item_id=%s", item.get("id"))
+    return result
+
+
 async def process_due_issue_confirmations(now: Optional[datetime] = None) -> Dict[str, int]:
    current = (now or _now()).astimezone(timezone.utc)
    items = list_portal_items(kind="issue", status="awaiting_confirmation", limit=500)
@@ -368,6 +572,9 @@ async def process_due_issue_confirmations(now: Optional[datetime] = None) -> Dic
 async def run_issue_confirmation_loop() -> None:
    while True:
        try:
+            repair_result = await process_active_media_repairs()
+            if repair_result["completed"] or repair_result["failed"]:
+                logger.info("automatic media repair sweep complete result=%s", repair_result)
            result = await process_due_issue_confirmations()
            if result["contacted"] or result["closed"] or result["failed"]:
                logger.info("issue confirmation sweep complete result=%s", result)
@@ -375,4 +582,4 @@ async def run_issue_confirmation_loop() -> None:
            raise
        except Exception:
            logger.exception("issue confirmation sweep failed")
-        await asyncio.sleep(15 * 60)
+        await asyncio.sleep(60)