From 009bb35032855cf5f0ecbc32650dc1d3a2bea9f5 Mon Sep 17 00:00:00 2001 From: Zak Bearman Date: Sun, 6 Sep 2026 22:06:30 +1200 Subject: [PATCH] Track repair collection cycles and reconcile request availability --- backend/app/db.py | 73 ++++++- backend/app/routers/requests.py | 56 ++++-- backend/app/services/issue_resolution.py | 133 +----------- backend/app/services/media_repair.py | 162 +++++++++++++++ backend/app/services/snapshot.py | 129 +++++++++++- backend/tests/test_backend_quality.py | 6 +- backend/tests/test_collector_search.py | 2 +- backend/tests/test_repair_pipeline.py | 244 +++++++++++++++++++++++ frontend/UI.md | 2 + frontend/app/requests/[id]/page.tsx | 14 +- scripts/review_repair_pipeline_ui.cjs | 92 +++++++++ 11 files changed, 762 insertions(+), 151 deletions(-) create mode 100644 backend/app/services/media_repair.py create mode 100644 backend/tests/test_repair_pipeline.py create mode 100644 scripts/review_repair_pipeline_ui.cjs diff --git a/backend/app/db.py b/backend/app/db.py index e2d50a8..6268cf7 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -187,6 +187,16 @@ def _has_secure_bootstrap_admin_credentials() -> bool: def init_db() -> None: with _connect() as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS request_repairs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + request_id TEXT NOT NULL, + started_at TEXT NOT NULL, + tracking_json TEXT NOT NULL, + completed_at TEXT, + UNIQUE(request_id, started_at) + ) + """) conn.execute( """ CREATE TABLE IF NOT EXISTS snapshots ( @@ -706,6 +716,59 @@ def init_db() -> None: pass _backfill_auth_providers() ensure_admin_user() + _backfill_request_repairs() + + +def start_request_repair(tracking: Dict[str, Any]) -> None: + """Persist the new collection cycle before a managed file is removed.""" + with _connect() as conn: + conn.execute( + "INSERT OR IGNORE INTO request_repairs (request_id, started_at, tracking_json) VALUES (?, ?, ?)", + (str(tracking["requestId"]), tracking["startedAt"], json.dumps(tracking)), + ) + + +def _backfill_request_repairs() -> None: + # Carry existing issue repairs forward once, without depending on the ticket's + # lifetime. Deleting/closing an issue must not restore stale availability. + with _connect() as conn: + rows = conn.execute(""" + SELECT a.metadata_json FROM portal_item_activity a + JOIN portal_items p ON p.id = a.item_id + WHERE p.kind = 'issue' AND p.status IN ('in_progress', 'blocked') + AND a.event_type IN ('replacement_started', 'missing_search_started') + AND a.id = (SELECT MAX(b.id) FROM portal_item_activity b + WHERE b.item_id = a.item_id + AND b.event_type IN ('replacement_started', 'missing_search_started')) + """).fetchall() + for (raw,) in rows: + try: + tracking = json.loads(raw or "{}").get("repairTracking") + if isinstance(tracking, dict) and tracking.get("requestId") and tracking.get("startedAt"): + start_request_repair(tracking) + except (TypeError, ValueError): + continue + + +def get_request_repairs(request_id: str, *, active_only: bool = True) -> list[Dict[str, Any]]: + with _connect() as conn: + rows = conn.execute( + "SELECT id, tracking_json, completed_at FROM request_repairs WHERE request_id = ?" + + (" AND completed_at IS NULL" if active_only else "") + " ORDER BY id", + (str(request_id),), + ).fetchall() + return [{"id": row[0], **json.loads(row[1]), "completedAt": row[2]} for row in rows] + + +def complete_request_repair(repair_id: int) -> None: + with _connect() as conn: + conn.execute("UPDATE request_repairs SET completed_at = ? WHERE id = ? AND completed_at IS NULL", + (datetime.now(timezone.utc).isoformat(), repair_id)) + + +def active_repair_request_ids() -> set[str]: + with _connect() as conn: + return {row[0] for row in conn.execute("SELECT DISTINCT request_id FROM request_repairs WHERE completed_at IS NULL")} def save_snapshot(snapshot: Snapshot) -> None: @@ -826,15 +889,17 @@ def get_request_download_evidence(request_id: str, limit: int = 100) -> Dict[str described honestly in the UI. """ with _connect() as conn: + cycle = conn.execute("SELECT MAX(started_at) FROM request_repairs WHERE request_id = ?", + (str(request_id),)).fetchone()[0] rows = conn.execute( """ SELECT created_at, payload_json FROM snapshots - WHERE request_id = ? + WHERE request_id = ? AND (? IS NULL OR created_at >= ?) ORDER BY id DESC LIMIT ? """, - (request_id, max(1, min(int(limit or 100), 500))), + (request_id, cycle, cycle, max(1, min(int(limit or 100), 500))), ).fetchall() for created_at, payload_json in rows: @@ -842,6 +907,10 @@ def get_request_download_evidence(request_id: str, limit: int = 100) -> Dict[str payload = json.loads(payload_json) except (TypeError, ValueError): continue + # A poll begun before deletion may finish afterwards. Its wall-clock save + # time alone is not evidence that it belongs to the replacement cycle. + if cycle and (payload.get("raw", {}).get("repairCycle") or "") < cycle: + continue timeline = payload.get("timeline") if isinstance(payload, dict) else None if not isinstance(timeline, list): continue diff --git a/backend/app/routers/requests.py b/backend/app/routers/requests.py index b2a39a2..0d690bf 100644 --- a/backend/app/routers/requests.py +++ b/backend/app/routers/requests.py @@ -51,7 +51,11 @@ from ..db import ( record_seerr_media_failure, clear_seerr_media_failure, get_request_download_evidence, + start_request_repair, + get_request_repairs, + active_repair_request_ids, ) +from ..services.media_repair import current_cycle_torrents from ..models import Snapshot, TriageResult, RequestType from ..services.snapshot import ( _summarize_qbit, @@ -1401,6 +1405,7 @@ def _get_recent_from_cache( status_codes: Optional[list[int]] = None, ) -> List[Dict[str, Any]]: items = _recent_cache.get("items") or [] + repairing = active_repair_request_ids() results = [] since_dt = _parse_iso_datetime(since_iso) for item in items: @@ -1414,6 +1419,8 @@ def _get_recent_from_cache( item_dt = _parse_iso_datetime(candidate) if not item_dt or item_dt < since_dt: continue + if str(item.get("request_id")) in repairing: + item = {**item, "status": 5, "repairing": True} if status_codes and item.get("status") not in status_codes: continue results.append(item) @@ -1862,6 +1869,9 @@ def _record_replacement_activity( message: str, metadata: Optional[Dict[str, Any]] = None, ) -> None: + tracking = (metadata or {}).get("repairTracking") + if event_type in {"replacement_started", "missing_search_started"} and isinstance(tracking, dict): + start_request_repair(tracking) if not issue: return current_status = str(issue.get("status") or "new").strip().lower() @@ -2183,6 +2193,27 @@ async def action_replace_media( collector_id: Optional[int] = None target_episodes: List[Dict[str, int]] = [] jellyfin_baseline: List[Dict[str, Any]] = [] + repair_tracking: Dict[str, Any] = {} + + def record_cycle() -> None: + repair_tracking.update({ + "requestId": request_id, + "actionId": "replace_media", + "mediaType": snapshot.request_type.value, + "collectorId": collector_id, + "originalFileIds": file_ids, + "previousDownloadIds": list(dict.fromkeys( + list(snapshot.raw.get("qbittorrent", {}).get("downloadIds") or []) + + [str(t.get("hash")) for t in snapshot.raw.get("qbittorrent", {}).get("torrents", []) if t.get("hash")] + )), + "episodes": target_episodes, + "jellyfinBaseline": jellyfin_baseline, + "jellyfinFoundAtStart": bool(snapshot.raw.get("jellyfin", {}).get("catalogFound", + snapshot.raw.get("jellyfin", {}).get("found"))), + "startedAt": datetime.now(timezone.utc).isoformat(), + }) + start_request_repair(repair_tracking) + try: if snapshot.request_type == RequestType.movie: movie_id = arr_item.get("id") @@ -2201,6 +2232,7 @@ async def action_replace_media( if not radarr.configured(): raise HTTPException(status_code=400, detail="Radarr is not configured") await radarr.monitor_movie(movie_id, True) + await asyncio.to_thread(record_cycle) await radarr.delete_movie_file(file_ids[0]) await radarr.search(movie_id) elif snapshot.request_type == RequestType.tv: @@ -2263,6 +2295,7 @@ async def action_replace_media( 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) + await asyncio.to_thread(record_cycle) for selected_file_id in file_ids: await sonarr.delete_episode_file(selected_file_id) await sonarr.search_episodes(episode_ids) @@ -2318,17 +2351,7 @@ async def action_replace_media( 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(), - } + "repairTracking": repair_tracking }, ) return { @@ -2688,6 +2711,8 @@ async def get_download_progress( if seerr.configured(): await _ensure_request_access(seerr, int(request_id), user) + repairs = await asyncio.to_thread(get_request_repairs, request_id, active_only=False) + cycle = repairs[-1]["startedAt"] if repairs else None evidence = await asyncio.to_thread(get_request_download_evidence, request_id, 20) historical_torrents = evidence.get("torrents") if isinstance(evidence, dict) else [] hashes: List[str] = [] @@ -2717,7 +2742,7 @@ async def get_download_progress( logger.warning("Live qBittorrent progress failed request_id=%s: %s", request_id, exc) raise HTTPException(status_code=502, detail="Live download progress is temporarily unavailable") from exc - torrents = result if isinstance(result, list) else [] + torrents = current_cycle_torrents(result, cycle) for torrent in torrents: if isinstance(torrent, dict): torrent["progressPercent"] = _torrent_progress(torrent) @@ -2739,6 +2764,8 @@ async def get_download_progress( "summary": message, "torrents": torrents, "updated_at": datetime.now(timezone.utc).isoformat(), + "repairCycle": cycle, + "visible": bool(torrents) or bool(evidence.get("observed")), } @@ -2879,7 +2906,10 @@ async def recent_requests( payload_json=json.dumps(details, ensure_ascii=True), ) status_label = _status_label(status) - if status_label in {"Working on it", "Ready to watch", "Partially ready"}: + if row.get("repairing"): + status = 5 + status_label = "Repair in progress" + elif status_label in {"Working on it", "Ready to watch", "Partially ready"}: is_available = await _request_is_available_in_jellyfin( jellyfin, title, diff --git a/backend/app/services/issue_resolution.py b/backend/app/services/issue_resolution.py index 3573545..2ca5daf 100644 --- a/backend/app/services/issue_resolution.py +++ b/backend/app/services/issue_resolution.py @@ -16,11 +16,10 @@ from ..db import ( 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 +from .media_repair import evaluate_media_repair logger = logging.getLogger(__name__) @@ -138,133 +137,15 @@ def _repair_tracking(item_id: int) -> tuple[Dict[str, Any], list[Dict[str, Any]] 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) + snapshot = await build_snapshot(str(tracking.get("requestId") or "")) 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 + jellyfin = dict(raw.get("jellyfin") or {}) + jellyfin["found"] = jellyfin.get("catalogFound", jellyfin.get("found")) + return await evaluate_media_repair( + tracking, (raw.get("arr") or {}).get("item"), jellyfin, + episodes=(raw.get("arr") or {}).get("episodes"), ) - 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( diff --git a/backend/app/services/media_repair.py b/backend/app/services/media_repair.py new file mode 100644 index 0000000..a6bf252 --- /dev/null +++ b/backend/app/services/media_repair.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any, Dict + +from ..clients.jellyfin import JellyfinClient +from ..clients.sonarr import SonarrClient +from ..runtime import get_runtime_settings + + +def current_cycle_torrents(torrents: Any, cycle: str | None) -> list[Dict[str, Any]]: + """Old seeding jobs are not proof of a replacement download. + + A same-hash retry is valid when it is downloading again or was added anew. + Without a completion/add timestamp, a completed legacy job cannot prove that. + """ + rows = [item for item in torrents if isinstance(item, dict)] if isinstance(torrents, list) else [] + if not cycle: + return rows + cutoff = datetime.fromisoformat(cycle).timestamp() + def belongs(item: Dict[str, Any]) -> bool: + try: + progress = float(item.get("progress", 0)) + completed = float(item.get("completion_on") or 0) + added = float(item.get("added_on") or 0) + except (TypeError, ValueError): + return False + return progress < 1 or max(completed, added) >= cutoff + return [item for item in rows if belongs(item)] + + +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 evaluate_media_repair( + tracking: Dict[str, Any], arr_item: Any, jellyfin: Dict[str, Any], + *, episodes: Any = None, +) -> 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."} + + jellyfin_item = jellyfin.get("item") + 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 not isinstance(arr_item, dict) or arr_item.get("id") != collector_id: + return {"complete": False, "phase": "collecting", "message": "Waiting for the correct collector record."} + + 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 = arr_item.get("hasFile") is not False and 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) + if episodes is None: + 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 not False + 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."} diff --git a/backend/app/services/snapshot.py b/backend/app/services/snapshot.py index ccf5a86..4dea292 100644 --- a/backend/app/services/snapshot.py +++ b/backend/app/services/snapshot.py @@ -19,6 +19,8 @@ from ..db import ( get_request_cache_payload, get_request_cache_by_id, get_request_download_evidence, + get_request_repairs, + complete_request_repair, get_recent_snapshots, get_setting, set_setting, @@ -28,6 +30,7 @@ from ..db import ( ) from ..models import ActionOption, NormalizedState, RequestType, Snapshot, TimelineHop from .collector_search import read_search_status +from .media_repair import current_cycle_torrents, evaluate_media_repair logger = logging.getLogger(__name__) @@ -975,9 +978,83 @@ def _build_presentation( } +def _apply_repair_presentation( + snapshot: Snapshot, repairs: List[Dict[str, Any]], arr_details: Dict[str, Any], + arr_state: str, download: Dict[str, Any], catalog_found: bool, + jellyfin_item: Any, public_url: Optional[str], +) -> None: + """Describe the replacement, without erasing approval or unaffected episodes.""" + imported = all(repair.get("phase") == "indexing" for repair in repairs) + unavailable = arr_state == "error" or any(repair.get("phase") == "unavailable" for repair in repairs) + latest = repairs[-1] + activity = _build_repair_activity( + snapshot, + action={"action_id": latest.get("actionId"), "status": "ok", + "created_at": latest.get("startedAt"), "message": "A new collection cycle was requested."}, + arr_state="available" if imported else arr_state, + arr_details={"availability": {"total": 1, "missing": 0}} if imported else arr_details, + download=download, jellyfin_found=False, + ) or {} + search = (arr_details.get("search") or {}).get("state") + pipeline = {stage["id"]: stage for stage in snapshot.presentation["pipeline"]} + if imported: + label = "Replacement collected — updating Grizzlyflix" + meaning = "The replacement has been imported. Waiting for Grizzlyflix to index the updated file." + snapshot.state = NormalizedState.importing + pipeline["download"].update(state="complete", summary="The replacement has been imported.", torrents=[], visible=False) + pipeline["available"].update(label="Updating Grizzlyflix", state="active", stateLabel="Indexing", summary=meaning) + snapshot.presentation["nextStep"] = { + "title": "Wait for the updated file", "description": "This page will update when Grizzlyflix confirms the replacement.", "actionIds": [], + } + elif unavailable: + label = "Repair status temporarily unavailable" + meaning = "Magent cannot verify the replacement right now. The old library entry is not confirmation that the repair is complete." + snapshot.state = NormalizedState.unknown + pipeline["available"].update(state="waiting", stateLabel="Unconfirmed", summary="Waiting for the replacement to be verified.") + snapshot.presentation["nextStep"] = {"title": "Recheck the request", "description": "Magent will retry automatically. You can also use Recheck request.", "actionIds": []} + activity.update(state="attention", headline=label, message=meaning) + elif download.get("visible"): + label = activity.get("headline", "Replacement in progress") + meaning = activity.get("message", "Magent is tracking the replacement download.") + snapshot.state = NormalizedState.importing if download.get("state") == "completed" else NormalizedState.downloading + else: + label = "Searching for a replacement" if search == "searching" else "Replacement search queued" if search == "queued" else "Waiting for a replacement" + meaning = "The affected content is being replaced. " + { + "searching": "The collector is looking for a suitable release.", + "queued": "The collector has queued the search.", + "idle": "No download has started and the collector is not currently searching.", + }.get(search, "Magent cannot currently confirm the search status.") + snapshot.state = NormalizedState.searching if search in {"searching", "queued"} else NormalizedState.added_to_arr + pipeline["download"].update(label="Replacement download", state="waiting", stateLabel="Pending", + summary="Waiting for a replacement download to start.", torrents=[], visible=False) + activity.update(state="searching" if search in {"searching", "queued"} else "waiting", headline=label, message=meaning) + previous_activity = snapshot.presentation.get("repairActivity") or {} + if search not in {"searching", "queued"} and previous_activity.get("state") == "attention" and (previous_activity.get("updatedAt") or "") >= latest["startedAt"]: + label, meaning = "Repair needs attention", str(previous_activity.get("message") or meaning) + activity.update(state="attention", headline=label, message=meaning) + snapshot.presentation["status"] = {"label": label, "meaning": meaning} + snapshot.presentation["repairActivity"] = activity + if not imported: + pipeline["available"].update(summary="The affected content will be available after the replacement is imported and indexed.") + + counts = arr_details.get("availability") or {} + targets = {episode.get("id") for repair in repairs for episode in repair.get("episodes", [])} + # For a series, keep a route to unaffected episodes without claiming that the + # repaired ones are ready (even while a stale series entry remains indexed). + has_unaffected = snapshot.request_type == RequestType.tv and int(counts.get("available") or 0) > (len(targets) if imported else 0) + if has_unaffected and catalog_found and isinstance(jellyfin_item, dict) and jellyfin_item.get("Id"): + link = f"{public_url.rstrip('/')}/web/index.html#!/details?id={quote(str(jellyfin_item['Id']))}" if public_url else None + pipeline["available"].update(label="Partially available", state="partial", stateLabel="Repair in progress", + summary="Other collected episodes remain available. The selected episodes are being replaced." if not imported else "Other episodes remain available. Waiting for Grizzlyflix to index the repaired episodes.", link=link) + snapshot.raw["jellyfin"].update(partial=True, link=link) + + async def build_snapshot(request_id: str) -> Snapshot: timeline = [] runtime = get_runtime_settings() + repair_records = await asyncio.to_thread(get_request_repairs, request_id, active_only=False) + repair_cycle = repair_records[-1]["startedAt"] if repair_records else None + active_repairs = [record for record in repair_records if not record.get("completedAt")] jellyseerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) @@ -1129,6 +1206,7 @@ async def build_snapshot(request_id: str) -> Snapshot: arr_details: Dict[str, Any] = {} arr_item = None arr_queue = None + episodes = None media_status = jelly_request.get("media", {}).get("status") try: media_status_code = int(media_status) if media_status is not None else None @@ -1168,6 +1246,8 @@ async def build_snapshot(request_id: str) -> Snapshot: "state": await read_search_status(sonarr, RequestType.tv, series_id, episodes) } arr_details["availability"] = _episode_availability(episodes) + counts = arr_details["availability"] + arr_state = "available" if counts.get("total", 0) > 0 and not counts.get("missing") else "added" missing_by_season = _missing_episode_numbers_by_season(episodes) if missing_by_season: arr_details["missingEpisodes"] = missing_by_season @@ -1276,7 +1356,7 @@ async def build_snapshot(request_id: str) -> Snapshot: jellyfin_item = item break - if jellyfin_available and arr_state == "missing" and runtime.jellyfin_sync_to_arr: + if jellyfin_available and not active_repairs and arr_state == "missing" and runtime.jellyfin_sync_to_arr: arr_details["note"] = "Found in Jellyfin but not tracked in Sonarr/Radarr." if snapshot.request_type == RequestType.movie: if runtime.radarr_quality_profile_id and runtime.radarr_root_folder: @@ -1317,6 +1397,30 @@ async def build_snapshot(request_id: str) -> Snapshot: except Exception: pass + catalog_found = jellyfin_available + pending_repairs = [] + for repair in active_repairs: + try: + evidence = await evaluate_media_repair( + repair, arr_item, {"found": catalog_found, "item": jellyfin_item}, episodes=episodes, + ) + except Exception: + logger.warning("Unable to verify replacement request_id=%s", request_id) + evidence = {"complete": False, "phase": "unavailable"} + if evidence.get("complete"): + await asyncio.to_thread(complete_request_repair, repair["id"]) + else: + pending_repairs.append({**repair, "phase": evidence.get("phase")}) + repair_imported = bool(pending_repairs) and all(r["phase"] == "indexing" for r in pending_repairs) + if pending_repairs: + # Jellyfin can retain the original item while its replacement is missing. + jellyfin_available = False + if arr_state == "available" and not repair_imported: + arr_state = "added" + elif snapshot.request_type == RequestType.movie and arr_state == "added": + # Also reconcile externally removed files, not only Magent repairs. + jellyfin_available = False + qbit_state = "not_started" qbit_message = "No download attempt has been observed." download_ids = _download_ids(_queue_records(arr_queue)) @@ -1333,6 +1437,17 @@ async def build_snapshot(request_id: str) -> Snapshot: request_tag = f"magent-{request_id}" torrents = await qbittorrent.get_torrents_by_tag(request_tag) torrent_list = torrents if isinstance(torrents, list) else [] + unfiltered_torrents = torrent_list + torrent_list = current_cycle_torrents(torrent_list, repair_cycle) + discarded_hashes = {str(t.get("hash") or "").lower() for t in unfiltered_torrents if t not in torrent_list} + if repair_cycle and not download_history.get("observed"): + current_hashes = {str(t.get("hash") or "").lower() for t in torrent_list} + discarded_hashes.update( + str(h).lower() for repair in active_repairs for h in repair.get("previousDownloadIds", []) + if str(h).lower() not in current_hashes + ) + download_ids = [h for h in download_ids if h.lower() not in discarded_hashes] + download_visible = bool(download_ids) or bool(download_history.get("observed")) for torrent in torrent_list: if isinstance(torrent, dict): torrent["progressPercent"] = _torrent_progress(torrent) @@ -1476,8 +1591,7 @@ async def build_snapshot(request_id: str) -> Snapshot: ) ) - download_ids = _download_ids(_queue_records(arr_queue)) - if download_ids and qbittorrent.configured(): + if download_ids and qbittorrent.configured() and qbit_state == "paused": actions.append( ActionOption( id="resume_torrent", @@ -1506,12 +1620,15 @@ async def build_snapshot(request_id: str) -> Snapshot: if jellyfin_available and not is_partial: snapshot.actions = [] snapshot.raw = { + "repairCycle": repair_cycle, "jellyseerr": jelly_request, "arr": { "item": arr_item, "queue": arr_queue, + "episodes": episodes, }, "jellyfin": { + "catalogFound": catalog_found, "publicUrl": runtime.jellyfin_public_url, "found": jellyfin_available, "available": jellyfin_available and snapshot.state in { @@ -1550,6 +1667,12 @@ async def build_snapshot(request_id: str) -> Snapshot: ) if repair_activity: snapshot.presentation["repairActivity"] = repair_activity + snapshot.presentation["repairCycle"] = repair_cycle + if pending_repairs: + _apply_repair_presentation( + snapshot, pending_repairs, arr_details, arr_state, download_presentation, + catalog_found, jellyfin_item, runtime.jellyfin_public_url, + ) status_presentation = snapshot.presentation.get("status") if isinstance(status_presentation, dict) and status_presentation.get("meaning"): snapshot.state_reason = str(status_presentation["meaning"]) diff --git a/backend/tests/test_backend_quality.py b/backend/tests/test_backend_quality.py index c5f419a..22bf1e3 100644 --- a/backend/tests/test_backend_quality.py +++ b/backend/tests/test_backend_quality.py @@ -317,7 +317,7 @@ class RequestCacheTests(unittest.TestCase): self.assertEqual(requests_router._cache_get(key), {"id": 123}) -class RequestVisibilityTests(unittest.TestCase): +class RequestVisibilityTests(TempDatabaseMixin, unittest.TestCase): def test_non_admin_snapshot_excludes_advanced_identifying_data(self) -> None: snapshot = Snapshot( request_id="3925", @@ -991,7 +991,7 @@ class SnapshotIdentityTests(unittest.TestCase): self.assertEqual(snapshot.year, 2016) -class LiveDownloadProgressTests(unittest.IsolatedAsyncioTestCase): +class LiveDownloadProgressTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase): async def test_live_download_progress_uses_saved_hash_and_current_qbittorrent_value(self) -> None: runtime = SimpleNamespace( jellyseerr_base_url=None, @@ -1499,7 +1499,7 @@ class AuthFlowTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase): self.assertEqual(context.exception.detail, "recipient_email is required and must be a valid email address.") -class MediaReplacementTests(unittest.IsolatedAsyncioTestCase): +class MediaReplacementTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase): def test_failed_repair_marks_linked_issue_as_blocked(self) -> None: issue = {"id": 12, "status": "in_progress"} with ( diff --git a/backend/tests/test_collector_search.py b/backend/tests/test_collector_search.py index 11406bf..92df2a7 100644 --- a/backend/tests/test_collector_search.py +++ b/backend/tests/test_collector_search.py @@ -146,7 +146,7 @@ class SearchSnapshotIntegrationTests(unittest.IsolatedAsyncioTestCase): "QBittorrentClient": SimpleNamespace(configured=lambda: False), "SonarrClient": collector, "RadarrClient": collector, "ProwlarrClient": SimpleNamespace(get_health=AsyncMock(return_value=[])), - "get_request_download_evidence": {}, "_latest_repair_action": None, "save_snapshot": None, + "get_request_download_evidence": {}, "get_request_repairs": [], "_latest_repair_action": None, "save_snapshot": None, } for name, value in mocks.items(): stack.enter_context(patch.object(snapshot_service, name, return_value=value)) diff --git a/backend/tests/test_repair_pipeline.py b/backend/tests/test_repair_pipeline.py new file mode 100644 index 0000000..25489a8 --- /dev/null +++ b/backend/tests/test_repair_pipeline.py @@ -0,0 +1,244 @@ +"""Replacement-cycle regressions. All collectors/downloads are fixtures.""" +from contextlib import ExitStack +from datetime import datetime, timedelta, timezone +import json +from types import SimpleNamespace +import unittest +from unittest.mock import AsyncMock, patch + +from backend.app import db +from backend.app.config import settings +from backend.app.models import NormalizedState, RequestType, Snapshot +from backend.app.routers import requests as requests_router +from backend.app.services import snapshot as service, media_repair +from backend.tests.test_backend_quality import TempDatabaseMixin + + +class RepairPipelineTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase): + def setUp(self): + super().setUp() + self.cycle = datetime.now(timezone.utc).isoformat() + self.item = {"id": 12, "title": "Example", "hasFile": False} + self.jf = {"Id": "jf-1", "Name": "Example", "Type": "Movie", "ProviderIds": {"Tmdb": "123"}, "Etag": "old"} + self.episodes = [ + {"id": 109, "seasonNumber": 5, "episodeNumber": 9, "hasFile": False, "episodeFileId": 0}, + {"id": 110, "seasonNumber": 5, "episodeNumber": 10, "hasFile": True, "episodeFileId": 42}, + ] + self.torrents = [] + self.queue = [] + self.commands = [] + self.jf_episodes = [{"Id": "ep9", "ParentIndexNumber": 5, "IndexNumber": 9, "Etag": "old"}] + self.media_type = RequestType.movie + self.fail_collector = False + + def start(self, media_type=RequestType.movie): + self.media_type = media_type + if media_type == RequestType.tv: + self.jf.update(Type="Series", ProviderIds={"Tvdb": "456"}) + tracking = { + "requestId": "12", "startedAt": self.cycle, "actionId": "replace_media", + "collectorId": 12, "mediaType": media_type.value, "originalFileIds": [40], + "previousDownloadIds": ["old"], + "episodes": [{"id": 109, "seasonNumber": 5, "episodeNumber": 9}] if media_type == RequestType.tv else [], + "jellyfinFoundAtStart": True, + "jellyfinBaseline": [{"Id": "ep9", "Etag": "old", "seasonNumber": 5, "episodeNumber": 9}] if media_type == RequestType.tv else [{"Id": "jf-1", "Etag": "old"}], + } + db.start_request_repair(tracking) + return tracking + + async def snapshot(self): + runtime = settings.model_copy(update={"requests_data_source": "prefer_cache", "jellyfin_public_url": "https://media.test"}) + lookup = AsyncMock(side_effect=RuntimeError("offline")) if self.fail_collector else AsyncMock(return_value=[self.item]) + collector = SimpleNamespace( + get_movie_by_tmdb_id=lookup, get_series_by_tvdb_id=lookup, + get_episodes=AsyncMock(return_value=self.episodes), get_queue=AsyncMock(return_value={"records": self.queue}), + get=AsyncMock(return_value=self.commands), + ) + jellyfin = SimpleNamespace(configured=lambda: True, search_items=AsyncMock(return_value={"Items": [self.jf]}), + get_series_episodes=AsyncMock(return_value=self.jf_episodes)) + with ExitStack() as stack: + mocks = { + "get_runtime_settings": runtime, + "get_request_cache_payload": {"id": 12, "type": self.media_type.value, "status": 4, + "media": {"title": "Example", "tmdbId": 123, "tvdbId": 456}}, + "get_request_cache_by_id": None, + "JellyseerrClient": SimpleNamespace(configured=lambda: False), "JellyfinClient": jellyfin, + "QBittorrentClient": SimpleNamespace(configured=lambda: True, + get_torrents_by_hashes=AsyncMock(return_value=self.torrents), get_torrents_by_tag=AsyncMock(return_value=self.torrents)), + "SonarrClient": collector, "RadarrClient": collector, + "ProwlarrClient": SimpleNamespace(get_health=AsyncMock(return_value=[])), + "_latest_repair_action": None, + } + for name, value in mocks.items(): + stack.enter_context(patch.object(service, name, return_value=value)) + stack.enter_context(patch.object(service, "_maybe_refresh_jellyfin", new=AsyncMock())) + stack.enter_context(patch.object(media_repair, "JellyfinClient", return_value=jellyfin)) + return await service.build_snapshot("12") + + @staticmethod + def stage(snapshot, name): + return next(stage for stage in snapshot.presentation["pipeline"] if stage["id"] == name) + + async def test_movie_old_catalog_and_completed_torrent_do_not_complete_repair(self): + self.start() + self.torrents = [{"hash": "old", "progress": 1, "state": "uploading", "added_on": 1, "completion_on": 2}] + self.queue = [{"movieId": 12, "downloadId": "old"}] + snapshot = await self.snapshot() + self.assertEqual(self.stage(snapshot, "download")["stateLabel"], "Pending") + self.assertEqual(self.stage(snapshot, "available")["state"], "waiting") + self.assertEqual(snapshot.presentation["status"]["label"], "Waiting for a replacement") + self.assertFalse(snapshot.presentation["download"]["visible"]) + self.assertTrue(snapshot.raw["jellyfin"]["catalogFound"]) + self.assertFalse(snapshot.raw["jellyfin"]["available"]) + self.assertIn("search_auto", [a.id for a in snapshot.actions]) + for name in ["requested", "approved"]: + self.assertEqual(self.stage(snapshot, name)["state"], "complete") + + async def test_movie_repair_search_download_import_index_and_complete(self): + self.start() + self.commands = [{"name": "MoviesSearch", "status": "started", "body": {"movieIds": [12]}}] + searching = await self.snapshot() + self.assertEqual(searching.presentation["status"]["label"], "Searching for a replacement") + self.commands = [] + self.torrents = [{"hash": "new", "progress": .32, "state": "downloading"}] + downloading = await self.snapshot() + self.assertEqual(downloading.state, NormalizedState.downloading) + self.assertEqual(self.stage(downloading, "download")["torrents"][0]["progressPercent"], 32) + self.assertNotIn("resume_torrent", [a.id for a in downloading.actions]) + self.item.update(hasFile=True, movieFile={"id": 41}) + imported = await self.snapshot() + self.assertEqual(self.stage(imported, "available")["stateLabel"], "Indexing") + self.assertEqual(self.stage(imported, "download")["state"], "complete") + self.assertEqual(len(db.get_request_repairs("12")), 1) + self.jf["Etag"] = "new" + self.torrents = [] + completed = await self.snapshot() + self.assertEqual(completed.state, NormalizedState.completed) + self.assertEqual(db.get_request_repairs("12"), []) + self.assertEqual(completed.presentation["status"]["label"], "Available to watch") + + async def test_old_queue_record_without_torrent_is_not_new_download_attempt(self): + self.start() + self.queue = [{"movieId": 12, "downloadId": "old"}] + snapshot = await self.snapshot() + self.assertEqual(self.stage(snapshot, "download")["stateLabel"], "Pending") + self.assertFalse(snapshot.presentation["download"]["visible"]) + + async def test_same_original_file_cannot_confirm_replacement(self): + self.start() + self.item.update(hasFile=True, movieFile={"id": 40}) + self.jf["Etag"] = "new" + snapshot = await self.snapshot() + self.assertNotEqual(snapshot.state, NormalizedState.completed) + self.assertEqual(self.stage(snapshot, "download")["state"], "waiting") + + async def test_tv_preserves_unaffected_episodes_and_verifies_exact_replacement(self): + self.start(RequestType.tv) + self.item["statistics"] = {"episodeFileCount": 2, "totalEpisodeCount": 2} # stale summary + pending = await self.snapshot() + self.assertEqual(self.stage(pending, "library")["missing"], 1) + self.assertEqual(self.stage(pending, "available")["state"], "partial") + self.assertEqual(self.stage(pending, "download")["stateLabel"], "Pending") + self.episodes[0].update(hasFile=True, episodeFileId=43) + imported = await self.snapshot() + self.assertEqual(imported.state, NormalizedState.importing) + self.assertEqual(self.stage(imported, "available")["state"], "partial") + self.assertEqual(len(db.get_request_repairs("12")), 1) + self.jf_episodes[0]["Etag"] = "new" + completed = await self.snapshot() + self.assertEqual(completed.state, NormalizedState.completed) + self.assertEqual(db.get_request_repairs("12"), []) + + async def test_collector_outage_does_not_restore_old_availability(self): + self.start() + self.fail_collector = True + snapshot = await self.snapshot() + self.assertNotEqual(snapshot.state, NormalizedState.completed) + self.assertEqual(snapshot.presentation["status"]["label"], "Repair status temporarily unavailable") + + async def test_external_movie_removal_reconciles_old_jellyfin_entry(self): + snapshot = await self.snapshot() + self.assertNotEqual(snapshot.state, NormalizedState.completed) + self.assertFalse(snapshot.raw["jellyfin"]["available"]) + + async def test_history_from_previous_cycle_and_late_old_poll_are_ignored(self): + old = Snapshot(request_id="12", title="Example", state=NormalizedState.completed, + timeline=[{"service": "qBittorrent", "status": "completed", "details": {"torrents": [{"hash": "old"}]}}]) + db.save_snapshot(old) + self.start() + self.assertFalse(db.get_request_download_evidence("12")["observed"]) + old.state_reason = "A pre-repair poll returned late" + db.save_snapshot(old) + self.assertFalse(db.get_request_download_evidence("12")["observed"]) + self.torrents = [{"hash": "new", "progress": .5, "state": "downloading"}] + await self.snapshot() + self.assertTrue(db.get_request_download_evidence("12")["observed"]) + + def test_same_hash_redownload_and_new_completed_job_are_kept(self): + old = {"hash": "same", "progress": 1, "added_on": 1, "completion_on": 2} + retry = {**old, "progress": .3} + fresh = {**old, "completion_on": datetime.now(timezone.utc).timestamp() + 1} + self.assertEqual(media_repair.current_cycle_torrents([old, retry, fresh], self.cycle), [retry, fresh]) + + def test_repair_cycle_survives_restart_and_list_does_not_say_ready(self): + self.start() + db.init_db() + self.assertEqual(len(db.get_request_repairs("12")), 1) + with patch.dict(requests_router._recent_cache, {"items": [{"request_id": 12, "status": 4, "requested_by_id": 7}]}): + self.assertEqual(requests_router._get_recent_from_cache(None, 7, 10, 0, None, [4]), []) + rows = requests_router._get_recent_from_cache(None, 7, 10, 0, None, [5]) + self.assertEqual(rows[0]["status"], 5) + + async def test_live_poll_ignores_old_completed_download(self): + self.start() + runtime = settings.model_copy(update={"jellyseerr_base_url": None, "jellyseerr_api_key": None}) + qbit = SimpleNamespace(configured=lambda: True, get_torrents_by_tag=AsyncMock(return_value=[{"progress": 1, "hash": "old", "state": "uploading"}])) + with patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object(requests_router, "QBittorrentClient", return_value=qbit): + progress = await requests_router.get_download_progress("12", {"role": "user"}) + self.assertEqual(progress["state"], "not_started") + self.assertFalse(progress["visible"]) + self.assertEqual(progress["repairCycle"], self.cycle) + + async def test_failed_search_after_deletion_keeps_cycle_and_pending_pipeline(self): + before = Snapshot(request_id="12", title="Example", request_type=RequestType.movie, + raw={"arr": {"item": {"id": 12, "hasFile": True, "movieFile": {"id": 40}}}, + "jellyfin": {"found": True, "item": self.jf}}) + async def delete(_): + self.assertEqual(len(db.get_request_repairs("12")), 1, "Must persist before removal") + radarr = SimpleNamespace(configured=lambda: True, monitor_movie=AsyncMock(), + delete_movie_file=AsyncMock(side_effect=delete), search=AsyncMock(side_effect=RuntimeError("search failed"))) + with ExitStack() as stack: + for name, value in {"_user_can_use_search_auto": True, "_linked_issue_for_replacement": None, + "JellyseerrClient": SimpleNamespace(configured=lambda: False), "RadarrClient": radarr}.items(): + stack.enter_context(patch.object(requests_router, name, return_value=value)) + stack.enter_context(patch.object(requests_router, "build_snapshot", new=AsyncMock(return_value=before))) + with self.assertRaises(requests_router.HTTPException): + await requests_router.action_replace_media("12", {"file_ids": [40], "confirmed": True}, {"role": "admin"}) + self.assertEqual(len(db.get_request_repairs("12")), 1) + snapshot = await self.snapshot() + self.assertEqual(self.stage(snapshot, "download")["state"], "waiting") + + def test_existing_issue_tracking_is_migrated_once_and_survives_ticket_deletion(self): + tracking = {"requestId": "12", "startedAt": self.cycle, "actionId": "replace_media", "collectorId": 12, + "mediaType": "movie", "originalFileIds": [40]} + issue = db.create_portal_item(kind="issue", title="Repair", description="Replace movie", status="in_progress", + created_by_username="reporter", created_by_id=None, issue_type="broken_media") + db.add_portal_item_activity(issue["id"], event_type="replacement_started", actor_username="reporter", + actor_role="user", message="Repair requested", metadata_json=json.dumps({"repairTracking": tracking})) + db.init_db() + db.init_db() + self.assertEqual(len(db.get_request_repairs("12")), 1) + db.delete_portal_item(issue["id"]) + self.assertEqual(len(db.get_request_repairs("12")), 1) + + async def test_multiple_repairs_wait_for_every_target_not_just_latest(self): + first = self.start(RequestType.tv) + second = {**first, "startedAt": (datetime.now(timezone.utc) + timedelta(seconds=1)).isoformat(), + "episodes": [{"id": 110, "seasonNumber": 5, "episodeNumber": 10}], "originalFileIds": [42], + "jellyfinFoundAtStart": False, "jellyfinBaseline": []} + db.start_request_repair(second) + self.episodes[1].update(episodeFileId=43) + self.jf_episodes.append({"Id": "ep10", "ParentIndexNumber": 5, "IndexNumber": 10}) + snapshot = await self.snapshot() + self.assertNotEqual(snapshot.state, NormalizedState.completed) + self.assertEqual([r["originalFileIds"] for r in db.get_request_repairs("12")], [[40]]) diff --git a/frontend/UI.md b/frontend/UI.md index b10f540..84981b7 100644 --- a/frontend/UI.md +++ b/frontend/UI.md @@ -7,6 +7,7 @@ - Keep primary actions, secondary controls and destructive actions visually distinct. Do not fade or uppercase every span inside a button: cards also use buttons, often with nested text. - Keep technical IDs and pipeline labels monospace. Use sentence case for ordinary labels and descriptions. - Preserve the six-stage request pipeline: three columns on desktop, two on tablet, one on narrow screens. Issue reports remain a right-hand column on desktop and stack on smaller screens. +- A media repair starts a new collection cycle: preserve Requested/Approved and unaffected TV episodes, but do not reuse an old torrent or Jellyfin entry to mark the replacement ready. Show Pending → Downloading → Indexing → Available from collector/file evidence. Subtitle-only repairs must not reset video availability. - Desktop navigation stays at the top; mobile navigation stays at the bottom. Dialogs must remain clear of both. - The guided issue form uses `portal/IssueFlowStep.tsx`: show one expanded step, collapse completed answers into Change rows, and keep repairs behind the final submit action. Movies use the selected title's managed file directly; only TV needs a season/episode picker. Multi-select controls must expose `aria-pressed` and a visible selected state. @@ -17,6 +18,7 @@ Build the frontend before reviewing. The scripts in `scripts/` run using Node an - `review_layout_ui.cjs`: page alignment, consistent headings, overflow, redirects, pipeline layout, invite tabs, and fixture-only recovery forms. - `review_account_ui.cjs`: fixture-only login and profile interaction checks. - `review_issue_flow_ui.cjs`: fixture-only movie/TV issue flow, multi-device report payloads, subtitle routing, permissions, and collapsed-step navigation. +- `review_repair_pipeline_ui.cjs`: fixture-only movie/TV replacement stages, old-cycle progress rejection, desktop/mobile layout and automatic availability transitions. - `review_settings_ui.cjs`: settings state, region-only saves, secret preservation, responsive controls and issue dialog placement. The layout/settings reviews accept `REVIEW_BASE`, `REVIEW_LIVE_BASE`, `REVIEW_PLAYWRIGHT`, and `REVIEW_DIR`. Provide an authorised short-lived session through `REVIEW_SESSION` as `{ "name": "cookie-name", "token": "..." }` in the process environment, never in a committed file. Live writes are blocked; submission checks use fixtures. Screenshots may contain account information and must stay outside the repository. diff --git a/frontend/app/requests/[id]/page.tsx b/frontend/app/requests/[id]/page.tsx index e064093..e2eec0f 100644 --- a/frontend/app/requests/[id]/page.tsx +++ b/frontend/app/requests/[id]/page.tsx @@ -68,6 +68,7 @@ type Snapshot = { actions: RequestAction[] artwork?: { poster_url?: string; backdrop_url?: string } presentation?: { + repairCycle?: string | null status?: { label?: string; meaning?: string } download?: { visible?: boolean @@ -121,6 +122,8 @@ type ActionHistory = { } type LiveDownloadProgress = { + repairCycle?: string | null + visible?: boolean request_id: string state: string summary: string @@ -205,15 +208,20 @@ const formatDuration = (duration?: number | null) => { const mergeLiveDownload = (current: Snapshot, live: LiveDownloadProgress): Snapshot => { if (String(current.request_id) !== String(live.request_id)) return current + if ((current.presentation?.repairCycle ?? null) !== (live.repairCycle ?? null)) return current + if (['COMPLETED', 'AVAILABLE'].includes(current.state)) return current + const downloadStage = current.presentation?.pipeline?.find((stage) => stage.id === 'download') + if (downloadStage?.state === 'complete' && downloadStage.visible === false) return current + const visible = live.visible ?? live.state !== 'not_started' const stageState = live.state === 'completed' ? 'complete' : ['missing', 'error'].includes(live.state) ? 'attention' - : 'active' + : live.state === 'not_started' ? 'waiting' : 'active' const presentation = current.presentation ?? {} const pipeline = (presentation.pipeline ?? fallbackPipeline(current)).map((stage) => stage.id === 'download' - ? { ...stage, state: stageState, summary: live.summary, visible: true, torrents: live.torrents } + ? { ...stage, state: stageState, stateLabel: undefined, summary: live.summary, visible, torrents: live.torrents } : stage ) return { @@ -222,7 +230,7 @@ const mergeLiveDownload = (current: Snapshot, live: LiveDownloadProgress): Snaps ...presentation, download: { ...(presentation.download ?? {}), - visible: true, + visible, state: live.state, summary: live.summary, torrents: live.torrents, diff --git a/scripts/review_repair_pipeline_ui.cjs b/scripts/review_repair_pipeline_ui.cjs new file mode 100644 index 0000000..4e70787 --- /dev/null +++ b/scripts/review_repair_pipeline_ui.cjs @@ -0,0 +1,92 @@ +// Fixture-only regression: never sends repairs, deletes files, or starts downloads. +const assert = require('node:assert/strict') +const { chromium } = require(process.env.REVIEW_PLAYWRIGHT || 'playwright') +const base = process.env.REVIEW_BASE || 'http://127.0.0.1:3101' +const cycle = '2026-09-06T01:00:00+00:00' + +const fixture = (state, partial = false) => { + const ready = state === 'complete' + const downloading = state === 'downloading' + const indexing = state === 'indexing' + const title = ready ? 'Available to watch' : indexing ? 'Replacement collected — updating Grizzlyflix' : downloading ? 'Replacement download in progress' : 'Waiting for a replacement' + const torrents = downloading ? [{ hash: 'new', name: 'Example replacement', state: 'downloading', progress: .32 }] : [] + return { + request_id: '12', title: 'Repair review', request_type: partial ? 'tv' : 'movie', + state: ready ? 'COMPLETED' : downloading ? 'DOWNLOADING' : indexing ? 'IMPORTING' : 'ADDED_TO_ARR', timeline: [], actions: [], + presentation: { + repairCycle: cycle, status: { label: title, meaning: title }, + download: { visible: downloading, state: downloading ? 'downloading' : 'not_started', torrents }, + nextStep: { title: ready ? 'Ready to watch' : 'Tracking your replacement', description: 'This page updates automatically.', actionIds: [] }, + repairActivity: { visible: !ready, state, headline: title, message: 'Tracking the affected content.', steps: [] }, + pipeline: [ + { id: 'requested', label: 'Requested', state: 'complete', summary: 'Request received' }, + { id: 'approved', label: 'Approved', state: 'complete', summary: 'Approved for collection' }, + { id: 'library', label: 'Library collection', state: ready || indexing ? 'complete' : 'active', searchStatus: 'idle', summary: 'Checking the collector' }, + { id: 'search', label: 'Release search', state: ready || indexing || downloading ? 'complete' : 'waiting', summary: 'Tracking collection' }, + { id: 'download', label: 'Replacement download', state: ready || indexing ? 'complete' : downloading ? 'active' : 'waiting', stateLabel: downloading ? 'Active' : ready || indexing ? 'Complete' : 'Pending', visible: downloading, torrents, summary: ready || indexing ? 'The replacement has been imported.' : downloading ? 'Downloading (1 active).' : 'Waiting for a replacement download to start.' }, + { id: 'available', label: partial && !ready ? 'Partially available' : ready ? 'Available to watch' : indexing ? 'Updating Grizzlyflix' : 'Media server', state: ready ? 'complete' : partial ? 'partial' : indexing ? 'active' : 'waiting', stateLabel: ready ? 'Ready' : partial ? 'Repair in progress' : indexing ? 'Indexing' : 'Waiting', summary: partial && !ready ? 'Other episodes remain available. The selected episodes are being replaced.' : ready ? 'Ready to watch.' : 'Waiting for the updated file.', link: ready || partial ? 'https://media.test/web/index.html#!/details?id=example' : null }, + ], + }, + } +} + +;(async () => { + const browser = await chromium.launch({ headless: true }) + try { + const context = await browser.newContext() + await context.addCookies([{ name: 'magent_logged_in', value: '1', url: base }]) + let mode = 'waiting', partial = false, liveMode = 'old', livePolls = 0, snapshots = 0 + const mutations = [], errors = [] + await context.route('**/api/**', (route) => { + const req = route.request(), path = new URL(req.url()).pathname + if (req.method() !== 'GET') mutations.push(path) + const reply = (json) => route.fulfill({ json }) + if (path === '/api/auth/me') return reply({ username: 'Member', role: 'user' }) + if (path.endsWith('/snapshot')) { snapshots++; return reply(fixture(mode, partial)) } + if (path.endsWith('/download-progress')) { + livePolls++ + return reply({ request_id: '12', repairCycle: liveMode === 'old' ? null : cycle, visible: true, + state: liveMode === 'old' ? 'completed' : 'downloading', summary: liveMode === 'old' ? 'Stale completed torrent' : 'Downloading (1 active).', + torrents: [{ hash: 'new', name: 'Example replacement', state: 'downloading', progress: liveMode === 'old' ? 1 : .48 }], updated_at: new Date().toISOString() }) + } + if (path.includes('/events/stream')) return route.fulfill({ contentType: 'text/event-stream', body: ': fixture\n\n' }) + if (path.includes('/branding/')) return route.fulfill({ status: 404 }) + return reply({ navigation: { showRequests: true }, requests: [], services: [] }) + }) + const page = await context.newPage() + page.on('pageerror', (error) => errors.push(error.message)) + const download = page.locator('.request-stage').filter({ has: page.getByRole('heading', { name: 'Replacement download', exact: true }) }) + for (const width of [1440, 390]) { + await page.setViewportSize({ width, height: 1000 }) + for (partial of [false, true]) { + for (mode of ['waiting', 'downloading', 'indexing', 'complete']) { + await page.goto(base + '/requests/12') + await download.waitFor() + assert.equal(await page.evaluate(() => document.documentElement.scrollWidth > innerWidth), false) + if (mode === 'waiting') await download.getByText('Pending', { exact: true }).waitFor() + if (!partial && mode === 'waiting') assert.equal(await page.getByRole('link', { name: /Open on media server/ }).count(), 0) + if (partial && mode !== 'complete') await page.getByRole('heading', { name: 'Partially available', exact: true }).waitFor() + if (process.env.REVIEW_DIR) await page.screenshot({ path: `${process.env.REVIEW_DIR}/repair-${width}-${partial ? 'tv' : 'movie'}-${mode}.png` }) + } + } + } + partial = false; mode = 'downloading'; liveMode = 'old'; livePolls = 0 + await page.goto(base + '/requests/12') + // Wait for a real scheduled live poll, rather than a manual data injection. + await page.waitForResponse((response) => response.url().includes('/download-progress'), { timeout: 6000 }) + assert.ok(livePolls > 0) + assert.equal(await download.getByText('Stale completed torrent', { exact: true }).count(), 0) + await download.getByText('32% complete', { exact: true }).waitFor() + liveMode = 'current' + await download.getByText('48% complete', { exact: true }).waitFor({ timeout: 6000 }) + const before = snapshots + mode = 'indexing' + await page.getByRole('heading', { name: 'Updating Grizzlyflix', exact: true }).waitFor({ timeout: 9000 }) + assert.ok(snapshots > before) + mode = 'complete' + await page.getByRole('heading', { name: 'Available to watch', exact: true }).waitFor({ timeout: 9000 }) + assert.deepEqual(mutations, []) + assert.deepEqual(errors, []) + console.log('PASS: movie/TV desktop/mobile repair states; old-cycle polls ignored; real-time progress and automatic indexing → available transition; no API writes.') + } finally { await browser.close() } +})().catch((error) => { console.error(error); process.exitCode = 1 })