From 0b59289a2eda8322f67cd237866c898ed1297941 Mon Sep 17 00:00:00 2001 From: Zak Bearman Date: Tue, 1 Sep 2026 22:37:23 +1200 Subject: [PATCH] Automate repair completion confirmation --- backend/app/clients/jellyfin.py | 21 +++ backend/app/routers/requests.py | 102 ++++++++++- backend/app/services/issue_resolution.py | 209 ++++++++++++++++++++++- backend/tests/test_backend_quality.py | 87 +++++++++- 4 files changed, 413 insertions(+), 6 deletions(-) diff --git a/backend/app/clients/jellyfin.py b/backend/app/clients/jellyfin.py index 0e5a478..220dc04 100644 --- a/backend/app/clients/jellyfin.py +++ b/backend/app/clients/jellyfin.py @@ -192,6 +192,7 @@ class JellyfinClient(ApiClient): "SearchTerm": term, "IncludeItemTypes": ",".join(item_types or []), "Recursive": "true", + "Fields": "Path,MediaSources", "Limit": limit, } headers = self._emby_headers() @@ -219,6 +220,26 @@ class JellyfinClient(ApiClient): ) raise + async def get_series_episodes(self, series_id: str) -> list[Dict[str, Any]]: + if not self.base_url or not self.api_key or not str(series_id).strip(): + return [] + url = f"{self.base_url}/Items" + params = { + "ParentId": str(series_id).strip(), + "IncludeItemTypes": "Episode", + "Recursive": "true", + "Fields": "Path,ProviderIds,MediaSources", + "Limit": 10000, + } + async with httpx.AsyncClient(timeout=20.0) as client: + response = await client.get(url, headers=self._emby_headers(), params=params) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + return [] + items = payload.get("Items") or payload.get("items") or [] + return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else [] + async def get_system_info(self) -> Optional[Dict[str, Any]]: if not self.base_url or not self.api_key: return None diff --git a/backend/app/routers/requests.py b/backend/app/routers/requests.py index 40e3ae8..b2a39a2 100644 --- a/backend/app/routers/requests.py +++ b/backend/app/routers/requests.py @@ -1803,6 +1803,16 @@ def _replacement_quality_name(file_data: Dict[str, Any]) -> Optional[str]: return str(value).strip() if value is not None and str(value).strip() else None +def _jellyfin_media_signature(item: Any) -> Dict[str, Any]: + if not isinstance(item, dict): + return {} + return { + key: item.get(key) + for key in ("Id", "Etag", "Path", "DateCreated", "MediaSources") + if item.get(key) is not None + } + + def _replacement_file_payload( file_data: Dict[str, Any], *, @@ -1850,6 +1860,7 @@ def _record_replacement_activity( user: Dict[str, str], event_type: str, message: str, + metadata: Optional[Dict[str, Any]] = None, ) -> None: if not issue: return @@ -1871,6 +1882,11 @@ def _record_replacement_activity( actor_username=str(user.get("username") or "unknown"), actor_role=str(user.get("role") or "user"), message=message, + metadata_json=( + json.dumps(metadata, separators=(",", ":"), sort_keys=True) + if metadata + else None + ), ) @@ -2164,6 +2180,9 @@ async def action_replace_media( collector = "Sonarr" if snapshot.request_type == RequestType.tv else "Radarr" target_names: List[str] = [] + collector_id: Optional[int] = None + target_episodes: List[Dict[str, int]] = [] + jellyfin_baseline: List[Dict[str, Any]] = [] try: if snapshot.request_type == RequestType.movie: movie_id = arr_item.get("id") @@ -2172,6 +2191,11 @@ async def action_replace_media( raise HTTPException(status_code=409, detail="Radarr does not report a replaceable movie file") if len(file_ids) != 1 or movie_file.get("id") != file_ids[0]: raise HTTPException(status_code=409, detail="The selected movie file is no longer current") + collector_id = movie_id + jellyfin_baseline = [ + _jellyfin_media_signature(snapshot.raw.get("jellyfin", {}).get("item")) + ] + jellyfin_baseline = [item for item in jellyfin_baseline if item] target_names = [_replacement_file_name(movie_file)] radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) if not radarr.configured(): @@ -2197,15 +2221,46 @@ async def action_replace_media( ] if isinstance(episode_files, list) else [] if len(selected_files) != len(file_ids): raise HTTPException(status_code=409, detail="One or more selected episode files are no longer current") - episode_ids = [ - episode.get("id") + target_episodes = [ + { + "id": int(episode["id"]), + "seasonNumber": int(episode["seasonNumber"]), + "episodeNumber": int(episode["episodeNumber"]), + } for episode in episodes if isinstance(episode, dict) and episode.get("episodeFileId") in file_ids and isinstance(episode.get("id"), int) + and isinstance(episode.get("seasonNumber"), int) + and isinstance(episode.get("episodeNumber"), int) ] if isinstance(episodes, list) else [] + episode_ids = [episode["id"] for episode in target_episodes] if not episode_ids: raise HTTPException(status_code=409, detail="Sonarr could not match episodes to this file") + collector_id = series_id + jellyfin_series = snapshot.raw.get("jellyfin", {}).get("item") + jellyfin_series_id = jellyfin_series.get("Id") if isinstance(jellyfin_series, dict) else None + if jellyfin_series_id: + try: + jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) + jellyfin_episodes = await jellyfin.get_series_episodes(str(jellyfin_series_id)) + target_pairs = { + (episode["seasonNumber"], episode["episodeNumber"]) + for episode in target_episodes + } + jellyfin_baseline = [ + { + "seasonNumber": int(item["ParentIndexNumber"]), + "episodeNumber": int(item["IndexNumber"]), + **_jellyfin_media_signature(item), + } + for item in jellyfin_episodes + if isinstance(item.get("ParentIndexNumber"), int) + and isinstance(item.get("IndexNumber"), int) + and (item["ParentIndexNumber"], item["IndexNumber"]) in target_pairs + ] + except Exception: + logger.warning("Could not capture Jellyfin episode baseline request_id=%s", request_id) target_names = [_replacement_file_name(file_data) for file_data in selected_files] await sonarr.monitor_episodes(episode_ids, True) for selected_file_id in file_ids: @@ -2262,6 +2317,19 @@ async def action_replace_media( user=user, event_type="replacement_started", message=message, + metadata={ + "repairTracking": { + "requestId": request_id, + "actionId": "replace_media", + "mediaType": snapshot.request_type.value, + "collectorId": collector_id, + "originalFileIds": file_ids, + "episodes": target_episodes, + "jellyfinBaseline": jellyfin_baseline, + "jellyfinFoundAtStart": bool(snapshot.raw.get("jellyfin", {}).get("found")), + "startedAt": datetime.now(timezone.utc).isoformat(), + } + }, ) return { "status": "ok", @@ -2318,6 +2386,7 @@ async def action_search_missing_media( if not isinstance(arr_item, dict) or not isinstance(arr_item.get("id"), int): raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr") collector_id = int(arr_item["id"]) + target_episodes: List[Dict[str, int]] = [] try: if snapshot.request_type == RequestType.movie: radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) @@ -2358,6 +2427,17 @@ async def action_search_missing_media( ) ] if searched_ids: + target_episodes = [ + { + "id": int(episode_map[episode_id]["id"]), + "seasonNumber": int(episode_map[episode_id]["seasonNumber"]), + "episodeNumber": int(episode_map[episode_id]["episodeNumber"]), + } + for episode_id in searched_ids + if isinstance(episode_map.get(episode_id), dict) + and isinstance(episode_map[episode_id].get("seasonNumber"), int) + and isinstance(episode_map[episode_id].get("episodeNumber"), int) + ] await sonarr.monitor_episodes(searched_ids, True) await sonarr.search_episodes(searched_ids) message = f"Sonarr started searching for {len(searched_ids)} selected missing episode(s)." @@ -2400,7 +2480,23 @@ async def action_search_missing_media( save_action, request_id, "search_missing", "Search for missing content", "ok", message ) _record_replacement_activity( - linked_issue, user=user, event_type="missing_search_started", message=message + linked_issue, + user=user, + event_type="missing_search_started", + message=message, + metadata={ + "repairTracking": { + "requestId": request_id, + "actionId": "search_missing", + "mediaType": snapshot.request_type.value, + "collectorId": collector_id, + "originalFileIds": [], + "episodes": target_episodes, + "jellyfinBaseline": [], + "jellyfinFoundAtStart": bool(snapshot.raw.get("jellyfin", {}).get("found")), + "startedAt": datetime.now(timezone.utc).isoformat(), + } + }, ) return {"status": "ok", "message": message, "episode_ids": searched_ids} diff --git a/backend/app/services/issue_resolution.py b/backend/app/services/issue_resolution.py index ab6a36e..3573545 100644 --- a/backend/app/services/issue_resolution.py +++ b/backend/app/services/issue_resolution.py @@ -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) diff --git a/backend/tests/test_backend_quality.py b/backend/tests/test_backend_quality.py index 907ca22..c5f419a 100644 --- a/backend/tests/test_backend_quality.py +++ b/backend/tests/test_backend_quality.py @@ -1575,6 +1575,8 @@ class MediaReplacementTests(unittest.IsolatedAsyncioTestCase): radarr.search.assert_awaited_once_with(44) add_activity.assert_called_once() self.assertEqual(add_activity.call_args.kwargs["event_type"], "replacement_started") + self.assertIn('"repairTracking"', add_activity.call_args.kwargs["metadata_json"]) + self.assertIn('"originalFileIds":[77]', add_activity.call_args.kwargs["metadata_json"]) update_issue.assert_called_once_with(12, status="in_progress", issue_resolved_at=None) async def test_tv_replacement_options_return_only_safe_file_details(self) -> None: @@ -1709,8 +1711,8 @@ class MediaReplacementTests(unittest.IsolatedAsyncioTestCase): {"id": 89, "relativePath": "S01E02.mkv"}, ]), get_episodes=AsyncMock(return_value=[ - {"id": 101, "episodeFileId": 88}, - {"id": 102, "episodeFileId": 89}, + {"id": 101, "episodeFileId": 88, "seasonNumber": 1, "episodeNumber": 1}, + {"id": 102, "episodeFileId": 89, "seasonNumber": 1, "episodeNumber": 2}, ]), monitor_episodes=AsyncMock(return_value={"monitored": True}), delete_episode_file=AsyncMock(return_value=None), @@ -2182,6 +2184,87 @@ class IssueResolutionWorkflowTests(TempDatabaseMixin, unittest.IsolatedAsyncioTe self.assertIsNotNone(closed["issue_resolved_at"]) self.assertEqual(db.list_portal_item_activity(int(issue["id"]))[-1]["event_type"], "resolution_confirmed") + async def test_replacement_waits_for_jellyfin_to_refresh_existing_movie(self) -> None: + tracking = { + "requestId": "144", + "actionId": "replace_media", + "mediaType": "movie", + "collectorId": 12, + "originalFileIds": [40], + "episodes": [], + "jellyfinFoundAtStart": True, + "jellyfinBaseline": [{"Id": "jf-1", "Etag": "old", "Path": "/movies/test.mkv"}], + } + unchanged = Snapshot( + request_id="144", + title="Test movie", + raw={ + "arr": {"item": {"id": 12, "hasFile": True, "movieFile": {"id": 41}}}, + "jellyfin": { + "found": True, + "item": {"Id": "jf-1", "Etag": "old", "Path": "/movies/test.mkv"}, + }, + }, + ) + refreshed = unchanged.model_copy(deep=True) + refreshed.raw["jellyfin"]["item"]["Etag"] = "new" + + with patch.object(issue_resolution, "build_snapshot", new=AsyncMock(return_value=unchanged)): + waiting = await issue_resolution._media_repair_evidence(tracking) + with patch.object(issue_resolution, "build_snapshot", new=AsyncMock(return_value=refreshed)): + complete = await issue_resolution._media_repair_evidence(tracking) + + self.assertFalse(waiting["complete"]) + self.assertEqual(waiting["phase"], "indexing") + self.assertTrue(complete["complete"]) + + async def test_verified_media_repair_starts_confirmation_without_admin(self) -> None: + issue = self._create_issue() + db.add_portal_item_activity( + int(issue["id"]), + event_type="replacement_started", + actor_username="reporter", + actor_role="user", + message="Radarr started the replacement.", + metadata_json=( + '{"repairTracking":{"requestId":"144","actionId":"replace_media",' + '"mediaType":"movie","collectorId":12,"originalFileIds":[40],' + '"episodes":[],"jellyfinFoundAtStart":true,' + '"jellyfinBaseline":[{"Id":"jf-1","Etag":"old"}]}}' + ), + ) + + with ( + patch.object( + issue_resolution, + "_media_repair_evidence", + new=AsyncMock( + return_value={ + "complete": True, + "phase": "complete", + "message": "Radarr imported the repaired movie and Jellyfin indexed it.", + } + ), + ), + patch.object( + issue_resolution, + "begin_issue_confirmation", + new=AsyncMock(return_value={"status": "awaiting_confirmation"}), + ) as begin_confirmation, + ): + result = await issue_resolution.process_active_media_repairs() + + self.assertEqual(result["completed"], 1) + begin_confirmation.assert_awaited_once_with( + int(issue["id"]), + actor_username="Magent", + actor_role="system", + ) + self.assertEqual( + db.list_portal_item_activity(int(issue["id"]))[-1]["event_type"], + "repair_verified", + ) + async def test_zero_confirmation_emails_closes_issue_immediately(self) -> None: issue = self._create_issue() with patch.object(issue_resolution, "_workflow_settings", return_value=(0, 1, "days")):