from typing import Any, Dict, List, Optional, Tuple import asyncio import httpx import json import logging import os import time from urllib.parse import quote from datetime import datetime, timezone, timedelta from fastapi import APIRouter, HTTPException, Depends from ..clients.jellyseerr import JellyseerrClient from ..clients.jellyfin import JellyfinClient from ..clients.qbittorrent import QBittorrentClient from ..clients.radarr import RadarrClient from ..clients.sonarr import SonarrClient from ..clients.bazarr import BazarrClient from ..ai.triage import triage_snapshot from ..auth import get_current_user from ..runtime import get_runtime_settings from .images import cache_tmdb_image, is_tmdb_cached from ..db import ( add_portal_item_activity, get_portal_item, save_action, get_recent_actions, get_recent_snapshots, get_cached_requests, get_cached_requests_since, get_cached_request_by_media_id, get_request_cache_lookup, get_request_cache_payload, get_request_cache_last_updated, get_request_cache_count, get_request_cache_payloads, get_request_cache_payloads_missing, repair_request_cache_titles, prune_duplicate_requests_cache, upsert_request_cache, upsert_request_cache_many, upsert_artwork_cache_status, upsert_artwork_cache_status_many, get_artwork_cache_missing_count, get_artwork_cache_status_count, get_setting, set_setting, update_portal_item, update_artwork_cache_stats, cleanup_history, is_seerr_media_failure_suppressed, record_seerr_media_failure, clear_seerr_media_failure, get_request_download_evidence, ) from ..models import Snapshot, TriageResult, RequestType from ..services.snapshot import ( _summarize_qbit, _torrent_progress, build_snapshot, jellyfin_item_matches_request, ) router = APIRouter(prefix="/requests", tags=["requests"], dependencies=[Depends(get_current_user)]) CACHE_TTL_SECONDS = 600 _detail_cache: Dict[str, Tuple[float, Dict[str, Any]]] = {} FAILED_DETAIL_CACHE_TTL_SECONDS = 3600 _failed_detail_cache: Dict[str, float] = {} REQUEST_CACHE_TTL_SECONDS = 600 logger = logging.getLogger(__name__) _sync_state: Dict[str, Any] = { "status": "idle", "stored": 0, "total": None, "skip": 0, "message": None, "started_at": None, "finished_at": None, } _sync_task: Optional[asyncio.Task] = None _sync_last_key = "requests_sync_last_at" RECENT_CACHE_MAX_DAYS = 180 RECENT_CACHE_TTL_SECONDS = 300 _recent_cache: Dict[str, Any] = {"items": [], "updated_at": None} _artwork_prefetch_state: Dict[str, Any] = { "status": "idle", "processed": 0, "total": 0, "message": "", "only_missing": False, "started_at": None, "finished_at": None, } _artwork_prefetch_task: Optional[asyncio.Task] = None STATUS_LABELS = { 1: "Waiting for approval", 2: "Approved", 3: "Declined", 4: "Ready to watch", 5: "Working on it", 6: "Partially ready", } REQUEST_STAGE_CODES = { "all": None, "pending": [1], "approved": [2], "declined": [3], "ready": [4], "working": [5], "partial": [6], "in_progress": [2, 5, 6], } def _cache_get(key: str) -> Optional[Dict[str, Any]]: cached = _detail_cache.get(key) if not cached: return None expires_at, payload = cached if expires_at < time.time(): _detail_cache.pop(key, None) return None return payload def _cache_set(key: str, payload: Dict[str, Any]) -> None: _detail_cache[key] = (time.time() + CACHE_TTL_SECONDS, payload) _failed_detail_cache.pop(key, None) def _status_label_with_jellyfin(current_status: Any, jellyfin_available: bool) -> str: if not jellyfin_available: return _status_label(current_status) try: status_code = int(current_status) except (TypeError, ValueError): status_code = None if status_code == 6: return STATUS_LABELS[6] return STATUS_LABELS[4] async def _request_is_available_in_jellyfin( jellyfin: JellyfinClient, title: Optional[str], year: Optional[int], media_type: Optional[str], request_payload: Optional[Dict[str, Any]], availability_cache: Dict[str, bool], ) -> bool: if not jellyfin.configured() or not title: return False cache_key = f"{media_type or ''}:{title.lower()}:{year or ''}:{request_payload.get('id') if isinstance(request_payload, dict) else ''}" cached_value = availability_cache.get(cache_key) if cached_value is not None: return cached_value types = ["Movie"] if media_type == "movie" else ["Series"] try: search = await jellyfin.search_items(title, types, limit=50) except Exception: availability_cache[cache_key] = False return False if isinstance(search, dict): items = search.get("Items") or search.get("items") or [] request_type = RequestType.movie if media_type == "movie" else RequestType.tv for item in items: if not isinstance(item, dict): continue if jellyfin_item_matches_request( item, title=title, year=year, request_type=request_type, request_payload=request_payload, ): availability_cache[cache_key] = True return True availability_cache[cache_key] = False return False def _failure_cache_has(key: str) -> bool: expires_at = _failed_detail_cache.get(key) if not expires_at: return False if expires_at < time.time(): _failed_detail_cache.pop(key, None) return False return True def _failure_cache_set(key: str, ttl_seconds: int = FAILED_DETAIL_CACHE_TTL_SECONDS) -> None: _failed_detail_cache[key] = time.time() + ttl_seconds def _extract_http_error_message(exc: httpx.HTTPStatusError) -> Optional[str]: response = exc.response if response is None: return None try: payload = response.json() except ValueError: payload = response.text if isinstance(payload, dict): message = payload.get("message") or payload.get("error") return str(message).strip() if message else json.dumps(payload, ensure_ascii=True) if isinstance(payload, str): trimmed = payload.strip() return trimmed or None return str(payload) def _should_persist_seerr_media_failure(exc: httpx.HTTPStatusError) -> bool: response = exc.response if response is None: return False return response.status_code == 404 or response.status_code >= 500 def _status_label(value: Any) -> str: if isinstance(value, int): return STATUS_LABELS.get(value, f"Status {value}") return "Unknown" def normalize_request_stage_filter(value: Optional[str]) -> str: if not isinstance(value, str): return "all" normalized = value.strip().lower().replace("-", "_").replace(" ", "_") if not normalized: return "all" if normalized in {"processing", "inprogress"}: normalized = "in_progress" return normalized if normalized in REQUEST_STAGE_CODES else "all" def request_stage_filter_codes(value: Optional[str]) -> Optional[list[int]]: normalized = normalize_request_stage_filter(value) codes = REQUEST_STAGE_CODES.get(normalized) return list(codes) if codes else None def _normalize_username(value: Any) -> Optional[str]: if not isinstance(value, str): return None normalized = value.strip().lower() if not normalized: return None if "@" in normalized: normalized = normalized.split("@", 1)[0] return normalized if normalized else None def _user_can_use_search_auto(user: Dict[str, Any]) -> bool: if user.get("role") == "admin": return True return bool(user.get("auto_search_enabled", True)) def _filter_snapshot_for_user(snapshot: Snapshot, user: Dict[str, Any]) -> Snapshot: if not _user_can_use_search_auto(user): snapshot.actions = [action for action in snapshot.actions if action.id != "search_auto"] if user.get("role") != "admin": # The standard request view is intentionally collaborative, but service payloads can # contain requester identities, internal URLs, download hashes and diagnostic errors. snapshot.timeline = [] snapshot.raw = {} return snapshot def _require_advanced_request_access(user: Dict[str, Any]) -> None: if user.get("role") != "admin": raise HTTPException( status_code=403, detail="Advanced request details are available to administrators only", ) def _quality_profile_id(value: Any) -> Optional[int]: if isinstance(value, int): return value if isinstance(value, str) and value.strip().isdigit(): return int(value.strip()) return None def _request_matches_user(request_data: Any, username: str) -> bool: requested_by = None if isinstance(request_data, dict): requested_by = request_data.get("requestedBy") or request_data.get("requestedByUser") if requested_by is None: requested_by = request_data.get("requestedByName") or request_data.get("requestedByUsername") if isinstance(requested_by, dict): candidates = [ requested_by.get("username"), requested_by.get("displayName"), requested_by.get("name"), requested_by.get("email"), ] else: candidates = [requested_by] username_norm = _normalize_username(username) if not username_norm: return False for candidate in candidates: candidate_norm = _normalize_username(candidate) if not candidate_norm: continue if "@" in candidate_norm: candidate_norm = candidate_norm.split("@", 1)[0] if candidate_norm == username_norm: return True return False def _normalize_requested_by(request_data: Any) -> Optional[str]: if not isinstance(request_data, dict): return None requested_by = request_data.get("requestedBy") if isinstance(requested_by, dict): for key in ("username", "displayName", "name", "email"): value = requested_by.get(key) normalized = _normalize_username(value) if normalized and "@" in normalized: normalized = normalized.split("@", 1)[0] if normalized: return normalized normalized = _normalize_username(requested_by) if normalized and "@" in normalized: normalized = normalized.split("@", 1)[0] return normalized def _extract_requested_by_id(request_data: Any) -> Optional[int]: if not isinstance(request_data, dict): return None requested_by = request_data.get("requestedBy") or request_data.get("requestedByUser") if isinstance(requested_by, dict): for key in ("id", "userId", "Id"): value = requested_by.get(key) if value is None: continue try: return int(value) except (TypeError, ValueError): continue return None def _format_upstream_error(service: str, exc: httpx.HTTPStatusError) -> str: response = exc.response status = response.status_code if response is not None else "unknown" message = "" if response is not None: try: payload = response.json() if isinstance(payload, dict): message = str(payload.get("message") or payload.get("error") or "").strip() elif isinstance(payload, list): validation_messages = [ str(item.get("errorMessage") or item.get("message") or "").strip() for item in payload if isinstance(item, dict) ] message = "; ".join(item for item in validation_messages if item) except ValueError: message = response.text.strip() if message: compact_message = " ".join(message.split())[:500] return f"{service} could not complete the request ({status}): {compact_message}" return f"{service} could not complete the request ({status})." def _request_display_name(request_data: Any) -> Optional[str]: if not isinstance(request_data, dict): return None requested_by = request_data.get("requestedBy") if isinstance(requested_by, dict): for key in ("displayName", "username", "name", "email"): value = requested_by.get(key) if isinstance(value, str) and value.strip(): return value.strip() if isinstance(requested_by, str) and requested_by.strip(): return requested_by.strip() return None def _parse_request_payload(item: Dict[str, Any]) -> Dict[str, Any]: media = item.get("media") or {} media_id = media.get("id") or item.get("mediaId") media_type = media.get("mediaType") or item.get("type") tmdb_id = media.get("tmdbId") or item.get("tmdbId") title = media.get("title") or media.get("name") or item.get("title") or item.get("name") year = media.get("year") or item.get("year") created_at = item.get("createdAt") or item.get("addedAt") or item.get("updatedAt") updated_at = item.get("updatedAt") or created_at requested_by = _request_display_name(item) requested_by_norm = _normalize_requested_by(item) requested_by_id = _extract_requested_by_id(item) return { "request_id": item.get("id"), "media_id": media_id, "media_type": media_type, "tmdb_id": tmdb_id, "status": item.get("status"), "title": title, "year": year, "requested_by": requested_by, "requested_by_norm": requested_by_norm, "requested_by_id": requested_by_id, "created_at": created_at, "updated_at": updated_at, } def _merge_request_media_details( request_payload: Dict[str, Any], details: Dict[str, Any] ) -> Dict[str, Any]: """Fill display metadata omitted by Seerr's request mutation/detail payloads.""" merged = dict(request_payload) media = request_payload.get("media") media = dict(media) if isinstance(media, dict) else {} media_type = _normalize_media_type( media.get("mediaType") or request_payload.get("mediaType") or request_payload.get("type") ) title = details.get("title") or details.get("name") if title and not (media.get("title") or media.get("name")): if media_type == "tv": media["name"] = title else: media["title"] = title date_value = details.get("releaseDate") or details.get("firstAirDate") if not media.get("year") and isinstance(date_value, str) and date_value[:4].isdigit(): media["year"] = int(date_value[:4]) for camel_key, snake_key in ( ("posterPath", "poster_path"), ("backdropPath", "backdrop_path"), ): if not (media.get(camel_key) or media.get(snake_key)): value = details.get(camel_key) or details.get(snake_key) if value: media[camel_key] = value if media_type and not media.get("mediaType"): media["mediaType"] = media_type merged["media"] = media return merged def _extract_artwork_paths(item: Dict[str, Any]) -> tuple[Optional[str], Optional[str]]: media = item.get("media") or {} poster_path = None backdrop_path = None if isinstance(media, dict): poster_path = media.get("posterPath") or media.get("poster_path") backdrop_path = media.get("backdropPath") or media.get("backdrop_path") if not poster_path: poster_path = item.get("posterPath") or item.get("poster_path") if not backdrop_path: backdrop_path = item.get("backdropPath") or item.get("backdrop_path") return poster_path, backdrop_path def _extract_tmdb_lookup(payload: Dict[str, Any]) -> tuple[Optional[int], Optional[str]]: media = payload.get("media") or {} if not isinstance(media, dict): media = {} tmdb_id = media.get("tmdbId") or payload.get("tmdbId") media_type = ( media.get("mediaType") or payload.get("mediaType") or payload.get("type") ) try: tmdb_id = int(tmdb_id) if tmdb_id is not None else None except (TypeError, ValueError): tmdb_id = None if isinstance(media_type, str): media_type = media_type.strip().lower() or None else: media_type = None return tmdb_id, media_type def _normalize_media_type(value: Any) -> Optional[str]: if not isinstance(value, str): return None normalized = value.strip().lower() if normalized in {"movie", "tv"}: return normalized return None def _normalize_seasons(value: Any) -> list[int]: if value is None: return [] if not isinstance(value, list): raise HTTPException(status_code=400, detail="seasons must be an array of positive integers") normalized: list[int] = [] for raw in value: try: season = int(raw) except (TypeError, ValueError) as exc: raise HTTPException( status_code=400, detail="seasons must contain only positive integers" ) from exc if season <= 0: raise HTTPException(status_code=400, detail="seasons must contain only positive integers") normalized.append(season) return sorted(set(normalized)) def _normalize_request_profiles(value: Any) -> list[Dict[str, Any]]: if not isinstance(value, list): return [] profiles: list[Dict[str, Any]] = [] for item in value: if not isinstance(item, dict): continue profile_id = _quality_profile_id(item.get("id")) name = str(item.get("name") or "").strip() if profile_id is None or not name: continue profiles.append({"id": profile_id, "name": name}) return profiles def _normalize_request_roots(value: Any) -> list[str]: if not isinstance(value, list): return [] roots: list[str] = [] for item in value: if not isinstance(item, dict): continue path = str(item.get("path") or "").strip() if path: roots.append(path) return roots def _normalize_seerr_servers(value: Any) -> list[Dict[str, Any]]: if isinstance(value, list): return [item for item in value if isinstance(item, dict)] if isinstance(value, dict): results = value.get("results") if isinstance(results, list): return [item for item in results if isinstance(item, dict)] return [] async def _resolve_request_destination( runtime: Any, seerr: JellyseerrClient, media_type: str, requested_profile_id: Optional[int] = None, ) -> Dict[str, Any]: if media_type == "tv": collector_name = "Sonarr" collector = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) configured_profile_id = _quality_profile_id(runtime.sonarr_quality_profile_id) configured_root = str(runtime.sonarr_root_folder or "").strip() else: collector_name = "Radarr" collector = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) configured_profile_id = _quality_profile_id(runtime.radarr_quality_profile_id) configured_root = str(runtime.radarr_root_folder or "").strip() if not collector.configured(): raise HTTPException(status_code=400, detail=f"{collector_name} is not configured") try: server_settings, profile_payload, root_payload = await asyncio.gather( seerr.get_service_settings(media_type), collector.get_quality_profiles(), collector.get_root_folders(), ) except httpx.HTTPStatusError as exc: service = "Seerr" if "/settings/" in str(exc.request.url) else collector_name raise HTTPException(status_code=502, detail=_format_upstream_error(service, exc)) from exc servers = [item for item in _normalize_seerr_servers(server_settings) if not item.get("is4k")] if not servers: raise HTTPException( status_code=409, detail=f"Seerr has no standard {collector_name} destination configured.", ) server = next((item for item in servers if item.get("isDefault")), servers[0]) profiles = _normalize_request_profiles(profile_payload) if not profiles: raise HTTPException(status_code=409, detail=f"{collector_name} has no quality profiles available.") profile_ids = {int(item["id"]) for item in profiles} default_profile_id = _quality_profile_id(server.get("activeProfileId")) if default_profile_id not in profile_ids: default_profile_id = configured_profile_id if configured_profile_id in profile_ids else profiles[0]["id"] selected_profile_id = requested_profile_id if requested_profile_id is not None else default_profile_id if selected_profile_id not in profile_ids: raise HTTPException( status_code=400, detail=f"The selected quality profile is not available in {collector_name}.", ) roots = _normalize_request_roots(root_payload) root_folder = str(server.get("activeDirectory") or "").strip() if root_folder not in roots: root_folder = configured_root if configured_root in roots else "" if not root_folder: raise HTTPException( status_code=409, detail=f"Seerr's {collector_name} library location does not match an active {collector_name} root folder.", ) server_id = _quality_profile_id(server.get("id")) if server_id is None: raise HTTPException(status_code=409, detail=f"Seerr's {collector_name} destination is invalid.") return { "collector": collector_name, "server_id": server_id, "server_name": str(server.get("name") or collector_name), "profile_id": int(selected_profile_id), "default_profile_id": int(default_profile_id), "profiles": profiles, "root_folder": root_folder, } def _artwork_missing_for_payload(payload: Dict[str, Any]) -> bool: poster_path, backdrop_path = _extract_artwork_paths(payload) tmdb_id, media_type = _extract_tmdb_lookup(payload) can_hydrate = bool(tmdb_id and media_type) if poster_path: if not is_tmdb_cached(poster_path, "w185") or not is_tmdb_cached(poster_path, "w342"): return True elif can_hydrate: return True if backdrop_path: if not is_tmdb_cached(backdrop_path, "w780"): return True elif can_hydrate: return True return False def _compute_cached_flags( poster_path: Optional[str], backdrop_path: Optional[str], cache_mode: str, poster_cached: Optional[bool] = None, backdrop_cached: Optional[bool] = None, ) -> tuple[bool, bool]: if cache_mode != "cache": return True, True poster = poster_cached backdrop = backdrop_cached if poster is None: poster = bool(poster_path) and is_tmdb_cached(poster_path, "w185") and is_tmdb_cached( poster_path, "w342" ) if backdrop is None: backdrop = bool(backdrop_path) and is_tmdb_cached(backdrop_path, "w780") return bool(poster), bool(backdrop) def _upsert_artwork_status( payload: Dict[str, Any], cache_mode: str, poster_cached: Optional[bool] = None, backdrop_cached: Optional[bool] = None, ) -> None: record = _build_artwork_status_record(payload, cache_mode, poster_cached, backdrop_cached) if not record: return upsert_artwork_cache_status(**record) def _build_request_cache_record(payload: Dict[str, Any], request_payload: Dict[str, Any]) -> Dict[str, Any]: return { "request_id": payload.get("request_id"), "media_id": payload.get("media_id"), "media_type": payload.get("media_type"), "status": payload.get("status"), "title": payload.get("title"), "year": payload.get("year"), "requested_by": payload.get("requested_by"), "requested_by_norm": payload.get("requested_by_norm"), "requested_by_id": payload.get("requested_by_id"), "created_at": payload.get("created_at"), "updated_at": payload.get("updated_at"), "payload_json": json.dumps(request_payload, ensure_ascii=True), } def _build_artwork_status_record( payload: Dict[str, Any], cache_mode: str, poster_cached: Optional[bool] = None, backdrop_cached: Optional[bool] = None, ) -> Optional[Dict[str, Any]]: parsed = _parse_request_payload(payload) request_id = parsed.get("request_id") if not isinstance(request_id, int): return None tmdb_id, media_type = _extract_tmdb_lookup(payload) poster_path, backdrop_path = _extract_artwork_paths(payload) has_tmdb = bool(tmdb_id and media_type) poster_cached_flag, backdrop_cached_flag = _compute_cached_flags( poster_path, backdrop_path, cache_mode, poster_cached, backdrop_cached ) return { "request_id": request_id, "tmdb_id": tmdb_id, "media_type": media_type, "poster_path": poster_path, "backdrop_path": backdrop_path, "has_tmdb": has_tmdb, "poster_cached": poster_cached_flag, "backdrop_cached": backdrop_cached_flag, } def _collect_artwork_cache_disk_stats() -> tuple[int, int]: cache_root = os.path.join(os.getcwd(), "data", "artwork") total_bytes = 0 total_files = 0 if not os.path.isdir(cache_root): return 0, 0 for root, _, files in os.walk(cache_root): for name in files: path = os.path.join(root, name) try: total_bytes += os.path.getsize(path) total_files += 1 except OSError: continue return total_bytes, total_files async def _get_request_details(client: JellyseerrClient, request_id: int) -> Optional[Dict[str, Any]]: cache_key = f"request:{request_id}" cached = _cache_get(cache_key) if isinstance(cached, dict): parsed_cached = _parse_request_payload(cached) if parsed_cached.get("title"): return cached details = await _get_media_details( client, parsed_cached.get("media_type"), parsed_cached.get("tmdb_id") ) if isinstance(details, dict): cached = _merge_request_media_details(cached, details) _cache_set(cache_key, cached) return cached if _failure_cache_has(cache_key): return None try: fetched = await client.get_request(str(request_id)) except httpx.HTTPStatusError: _failure_cache_set(cache_key) return None if isinstance(fetched, dict): parsed_fetched = _parse_request_payload(fetched) if not parsed_fetched.get("title"): details = await _get_media_details( client, parsed_fetched.get("media_type"), parsed_fetched.get("tmdb_id") ) if isinstance(details, dict): fetched = _merge_request_media_details(fetched, details) _cache_set(cache_key, fetched) return fetched return None async def _get_media_details( client: JellyseerrClient, media_type: Optional[str], tmdb_id: Optional[int] ) -> Optional[Dict[str, Any]]: if not tmdb_id or not media_type: return None normalized_media_type = str(media_type).strip().lower() if normalized_media_type not in {"movie", "tv"}: return None cache_key = f"media:{normalized_media_type}:{int(tmdb_id)}" cached = _cache_get(cache_key) if isinstance(cached, dict): return cached if is_seerr_media_failure_suppressed(normalized_media_type, int(tmdb_id)): logger.debug( "Seerr media hydration suppressed from db: media_type=%s tmdb_id=%s", normalized_media_type, tmdb_id, ) _failure_cache_set(cache_key, ttl_seconds=FAILED_DETAIL_CACHE_TTL_SECONDS) return None if _failure_cache_has(cache_key): return None try: if normalized_media_type == "movie": fetched = await client.get_movie(int(tmdb_id)) else: fetched = await client.get_tv(int(tmdb_id)) except httpx.HTTPStatusError as exc: _failure_cache_set(cache_key) if _should_persist_seerr_media_failure(exc): record_seerr_media_failure( normalized_media_type, int(tmdb_id), status_code=exc.response.status_code if exc.response is not None else None, error_message=_extract_http_error_message(exc), ) return None if isinstance(fetched, dict): clear_seerr_media_failure(normalized_media_type, int(tmdb_id)) _cache_set(cache_key, fetched) return fetched return None async def _hydrate_title_from_tmdb( client: JellyseerrClient, media_type: Optional[str], tmdb_id: Optional[int] ) -> tuple[Optional[str], Optional[int]]: details = await _get_media_details(client, media_type, tmdb_id) if not isinstance(details, dict): return None, None normalized_media_type = str(media_type).strip().lower() if media_type else None if normalized_media_type == "movie": title = details.get("title") release_date = details.get("releaseDate") year = int(release_date[:4]) if release_date else None return title, year if normalized_media_type == "tv": title = details.get("name") or details.get("title") first_air = details.get("firstAirDate") year = int(first_air[:4]) if first_air else None return title, year return None, None async def _hydrate_artwork_from_tmdb( client: JellyseerrClient, media_type: Optional[str], tmdb_id: Optional[int] ) -> tuple[Optional[str], Optional[str]]: details = await _get_media_details(client, media_type, tmdb_id) if not isinstance(details, dict): return None, None return ( details.get("posterPath") or details.get("poster_path"), details.get("backdropPath") or details.get("backdrop_path"), ) def _artwork_url(path: Optional[str], size: str, cache_mode: str) -> Optional[str]: if not path: return None if not path.startswith("/"): path = f"/{path}" if cache_mode == "cache": return f"/images/tmdb?path={quote(path)}&size={size}" return f"https://image.tmdb.org/t/p/{size}{path}" def _cache_is_stale(last_updated: Optional[str]) -> bool: if not last_updated: return True runtime = get_runtime_settings() ttl_seconds = max(60, int(runtime.requests_sync_ttl_minutes or 1440) * 60) try: parsed = datetime.fromisoformat(last_updated.replace("Z", "+00:00")) now = datetime.now(timezone.utc) return (now - parsed).total_seconds() > ttl_seconds except ValueError: return True def _parse_time(value: Optional[str], fallback_hour: int, fallback_minute: int) -> tuple[int, int]: if isinstance(value, str) and ":" in value: parts = value.strip().split(":") if len(parts) == 2: try: hour = int(parts[0]) minute = int(parts[1]) if 0 <= hour <= 23 and 0 <= minute <= 59: return hour, minute except ValueError: pass return fallback_hour, fallback_minute def _seconds_until(hour: int, minute: int) -> int: now = datetime.now(timezone.utc).astimezone() target = now.replace(hour=hour, minute=minute, second=0, microsecond=0) if target <= now: target = target + timedelta(days=1) return int((target - now).total_seconds()) async def _sync_all_requests(client: JellyseerrClient) -> int: take = 50 skip = 0 stored = 0 cache_mode = (get_runtime_settings().artwork_cache_mode or "remote").lower() logger.info("Seerr sync starting: take=%s", take) _sync_state.update( { "status": "running", "stored": 0, "total": None, "skip": 0, "message": "Starting sync", "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None, } ) while True: try: response = await client.get_recent_requests(take=take, skip=skip) except httpx.HTTPError as exc: logger.warning("Seerr sync failed at skip=%s: %s", skip, exc) _sync_state.update({"status": "failed", "message": f"Sync failed: {exc}"}) break if not isinstance(response, dict): logger.warning("Seerr sync stopped: non-dict response at skip=%s", skip) _sync_state.update({"status": "failed", "message": "Invalid response"}) break if _sync_state["total"] is None: page_info = response.get("pageInfo") or {} total = ( page_info.get("totalResults") or page_info.get("total") or response.get("totalResults") or response.get("total") ) if isinstance(total, int): _sync_state["total"] = total items = response.get("results") or [] if not isinstance(items, list) or not items: logger.info("Seerr sync completed: no more results at skip=%s", skip) break page_request_ids = [ payload.get("request_id") for item in items if isinstance(item, dict) for payload in [_parse_request_payload(item)] if isinstance(payload.get("request_id"), int) ] cached_by_request_id = get_request_cache_lookup(page_request_ids) page_cache_records: list[Dict[str, Any]] = [] page_artwork_records: list[Dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue payload = _parse_request_payload(item) request_id = payload.get("request_id") cached_title = None if isinstance(request_id, int): cached = cached_by_request_id.get(request_id) if not payload.get("title") and cached and cached.get("title"): cached_title = cached.get("title") needs_details = ( not payload.get("title") or not payload.get("media_id") or not payload.get("tmdb_id") or not payload.get("media_type") ) if needs_details: logger.debug("Seerr sync hydrate request_id=%s", request_id) details = await _get_request_details(client, request_id) if isinstance(details, dict): payload = _parse_request_payload(details) item = details poster_path, backdrop_path = _extract_artwork_paths(item) if cache_mode == "cache" and not (poster_path or backdrop_path): details = await _get_request_details(client, request_id) if isinstance(details, dict): item = details payload = _parse_request_payload(details) if not payload.get("title") and payload.get("tmdb_id") and payload.get("media_type"): hydrated_title, hydrated_year = await _hydrate_title_from_tmdb( client, payload.get("media_type"), payload.get("tmdb_id") ) if hydrated_title: payload["title"] = hydrated_title if hydrated_year: payload["year"] = hydrated_year if not payload.get("title") and cached_title: payload["title"] = cached_title if not isinstance(payload.get("request_id"), int): continue page_cache_records.append(_build_request_cache_record(payload, item)) if isinstance(item, dict): artwork_record = _build_artwork_status_record(item, cache_mode) if artwork_record: page_artwork_records.append(artwork_record) stored += 1 _sync_state["stored"] = stored if page_cache_records: upsert_request_cache_many(page_cache_records) if page_artwork_records: upsert_artwork_cache_status_many(page_artwork_records) if len(items) < take: logger.info("Seerr sync completed: stored=%s", stored) break skip += take _sync_state["skip"] = skip _sync_state["message"] = f"Synced {stored} requests" logger.debug("Seerr sync progress: stored=%s skip=%s", stored, skip) _sync_state.update( { "status": "completed", "stored": stored, "message": f"Sync complete: {stored} requests", "finished_at": datetime.now(timezone.utc).isoformat(), } ) set_setting(_sync_last_key, datetime.now(timezone.utc).isoformat()) _refresh_recent_cache_from_db() if cache_mode == "cache": update_artwork_cache_stats( missing_count=get_artwork_cache_missing_count(), total_requests=get_request_cache_count(), ) return stored async def _sync_delta_requests(client: JellyseerrClient) -> int: take = 50 skip = 0 stored = 0 unchanged_pages = 0 cache_mode = (get_runtime_settings().artwork_cache_mode or "remote").lower() logger.info("Seerr delta sync starting: take=%s", take) _sync_state.update( { "status": "running", "stored": 0, "total": None, "skip": 0, "message": "Starting delta sync", "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None, } ) while True: try: response = await client.get_recent_requests(take=take, skip=skip) except httpx.HTTPError as exc: logger.warning("Seerr delta sync failed at skip=%s: %s", skip, exc) _sync_state.update({"status": "failed", "message": f"Delta sync failed: {exc}"}) break if not isinstance(response, dict): logger.warning("Seerr delta sync stopped: non-dict response at skip=%s", skip) _sync_state.update({"status": "failed", "message": "Invalid response"}) break items = response.get("results") or [] if not isinstance(items, list) or not items: logger.info("Seerr delta sync completed: no more results at skip=%s", skip) break page_request_ids = [ payload.get("request_id") for item in items if isinstance(item, dict) for payload in [_parse_request_payload(item)] if isinstance(payload.get("request_id"), int) ] cached_by_request_id = get_request_cache_lookup(page_request_ids) page_cache_records: list[Dict[str, Any]] = [] page_artwork_records: list[Dict[str, Any]] = [] page_changed = False for item in items: if not isinstance(item, dict): continue payload = _parse_request_payload(item) request_id = payload.get("request_id") if isinstance(request_id, int): cached = cached_by_request_id.get(request_id) incoming_updated = payload.get("updated_at") cached_title = cached.get("title") if cached else None if cached and incoming_updated and cached.get("updated_at") == incoming_updated and cached.get("title"): continue needs_details = ( not payload.get("title") or not payload.get("media_id") or not payload.get("tmdb_id") or not payload.get("media_type") ) if needs_details: details = await _get_request_details(client, request_id) if isinstance(details, dict): payload = _parse_request_payload(details) item = details poster_path, backdrop_path = _extract_artwork_paths(item) if cache_mode == "cache" and not (poster_path or backdrop_path): details = await _get_request_details(client, request_id) if isinstance(details, dict): payload = _parse_request_payload(details) item = details if not payload.get("title") and payload.get("tmdb_id") and payload.get("media_type"): hydrated_title, hydrated_year = await _hydrate_title_from_tmdb( client, payload.get("media_type"), payload.get("tmdb_id") ) if hydrated_title: payload["title"] = hydrated_title if hydrated_year: payload["year"] = hydrated_year if not payload.get("title") and cached_title: payload["title"] = cached_title if not isinstance(payload.get("request_id"), int): continue page_cache_records.append(_build_request_cache_record(payload, item)) if isinstance(item, dict): artwork_record = _build_artwork_status_record(item, cache_mode) if artwork_record: page_artwork_records.append(artwork_record) stored += 1 page_changed = True _sync_state["stored"] = stored if page_cache_records: upsert_request_cache_many(page_cache_records) if page_artwork_records: upsert_artwork_cache_status_many(page_artwork_records) if not page_changed: unchanged_pages += 1 else: unchanged_pages = 0 if len(items) < take or unchanged_pages >= 2: logger.info("Seerr delta sync completed: stored=%s", stored) break skip += take _sync_state["skip"] = skip _sync_state["message"] = f"Delta synced {stored} requests" logger.debug("Seerr delta sync progress: stored=%s skip=%s", stored, skip) deduped = prune_duplicate_requests_cache() if deduped: logger.info("Seerr delta sync removed duplicate rows: %s", deduped) _sync_state.update( { "status": "completed", "stored": stored, "message": f"Delta sync complete: {stored} updated", "finished_at": datetime.now(timezone.utc).isoformat(), } ) set_setting(_sync_last_key, datetime.now(timezone.utc).isoformat()) _refresh_recent_cache_from_db() if cache_mode == "cache": update_artwork_cache_stats( missing_count=get_artwork_cache_missing_count(), total_requests=get_request_cache_count(), ) return stored async def _prefetch_artwork_cache( client: JellyseerrClient, only_missing: bool = False, total: Optional[int] = None, use_missing_query: bool = False, ) -> None: runtime = get_runtime_settings() cache_mode = (runtime.artwork_cache_mode or "remote").lower() if cache_mode != "cache": _artwork_prefetch_state.update( { "status": "failed", "message": "Artwork cache mode is not set to cache.", "finished_at": datetime.now(timezone.utc).isoformat(), } ) return total = total if total is not None else get_request_cache_count() _artwork_prefetch_state.update( { "status": "running", "processed": 0, "total": total, "message": "Starting missing artwork prefetch" if only_missing else "Starting artwork prefetch", "only_missing": only_missing, "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None, } ) if only_missing and total == 0: _artwork_prefetch_state.update( { "status": "completed", "processed": 0, "message": "No missing artwork to cache.", "finished_at": datetime.now(timezone.utc).isoformat(), } ) return offset = 0 limit = 200 processed = 0 while True: if use_missing_query: batch = get_request_cache_payloads_missing(limit=limit, offset=offset) else: batch = get_request_cache_payloads(limit=limit, offset=offset) if not batch: break page_cache_records: list[Dict[str, Any]] = [] page_artwork_records: list[Dict[str, Any]] = [] for row in batch: payload = row.get("payload") if not isinstance(payload, dict): if not only_missing: processed += 1 continue if only_missing and not use_missing_query and not _artwork_missing_for_payload(payload): continue poster_path, backdrop_path = _extract_artwork_paths(payload) tmdb_id, media_type = _extract_tmdb_lookup(payload) if (not poster_path or not backdrop_path) and client.configured() and tmdb_id and media_type: media = payload.get("media") or {} hydrated_poster, hydrated_backdrop = await _hydrate_artwork_from_tmdb( client, media_type, tmdb_id ) poster_path = poster_path or hydrated_poster backdrop_path = backdrop_path or hydrated_backdrop if hydrated_poster or hydrated_backdrop: media = dict(media) if isinstance(media, dict) else {} if hydrated_poster: media["posterPath"] = hydrated_poster if hydrated_backdrop: media["backdropPath"] = hydrated_backdrop payload["media"] = media parsed = _parse_request_payload(payload) request_id = parsed.get("request_id") if isinstance(request_id, int): page_cache_records.append(_build_request_cache_record(parsed, payload)) poster_cached_flag = False backdrop_cached_flag = False if poster_path: try: poster_cached_flag = bool( await cache_tmdb_image(poster_path, "w185") ) and bool(await cache_tmdb_image(poster_path, "w342")) except httpx.HTTPError: poster_cached_flag = False if backdrop_path: try: backdrop_cached_flag = bool(await cache_tmdb_image(backdrop_path, "w780")) except httpx.HTTPError: backdrop_cached_flag = False artwork_record = _build_artwork_status_record( payload, cache_mode, poster_cached=poster_cached_flag if poster_path else None, backdrop_cached=backdrop_cached_flag if backdrop_path else None, ) if artwork_record: page_artwork_records.append(artwork_record) processed += 1 if processed % 25 == 0: _artwork_prefetch_state.update( {"processed": processed, "message": f"Cached artwork for {processed} requests"} ) if page_cache_records: upsert_request_cache_many(page_cache_records) if page_artwork_records: upsert_artwork_cache_status_many(page_artwork_records) offset += limit total_requests = get_request_cache_count() missing_count = get_artwork_cache_missing_count() cache_bytes, cache_files = _collect_artwork_cache_disk_stats() update_artwork_cache_stats( cache_bytes=cache_bytes, cache_files=cache_files, missing_count=missing_count, total_requests=total_requests, ) _artwork_prefetch_state.update( { "status": "completed", "processed": processed, "message": f"Artwork cached for {processed} requests", "finished_at": datetime.now(timezone.utc).isoformat(), } ) async def start_artwork_prefetch( base_url: Optional[str], api_key: Optional[str], only_missing: bool = False ) -> Dict[str, Any]: global _artwork_prefetch_task if _artwork_prefetch_task and not _artwork_prefetch_task.done(): return dict(_artwork_prefetch_state) client = JellyseerrClient(base_url, api_key) status_count = get_artwork_cache_status_count() total_requests = get_request_cache_count() use_missing_query = only_missing and status_count >= total_requests and total_requests > 0 if only_missing and use_missing_query: total = get_artwork_cache_missing_count() else: total = total_requests _artwork_prefetch_state.update( { "status": "running", "processed": 0, "total": total, "message": "Seeding artwork cache status" if only_missing and not use_missing_query else ("Starting missing artwork prefetch" if only_missing else "Starting artwork prefetch"), "only_missing": only_missing, "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None, } ) if only_missing and total == 0: _artwork_prefetch_state.update( { "status": "completed", "processed": 0, "message": "No missing artwork to cache.", "finished_at": datetime.now(timezone.utc).isoformat(), } ) return dict(_artwork_prefetch_state) async def _runner() -> None: try: await _prefetch_artwork_cache( client, only_missing=only_missing, total=total, use_missing_query=use_missing_query, ) except Exception: logger.exception("Artwork prefetch failed") _artwork_prefetch_state.update( { "status": "failed", "message": "Artwork prefetch failed.", "finished_at": datetime.now(timezone.utc).isoformat(), } ) _artwork_prefetch_task = asyncio.create_task(_runner()) return dict(_artwork_prefetch_state) def get_artwork_prefetch_state() -> Dict[str, Any]: return dict(_artwork_prefetch_state) async def _ensure_requests_cache(client: JellyseerrClient) -> None: last_sync = get_setting(_sync_last_key) last_updated = last_sync or get_request_cache_last_updated() if _cache_is_stale(last_updated): logger.info("Requests cache stale or empty, starting sync.") await _sync_all_requests(client) else: logger.debug("Requests cache fresh: last_sync=%s", last_updated) def _refresh_recent_cache_from_db() -> None: since_iso = (datetime.now(timezone.utc) - timedelta(days=RECENT_CACHE_MAX_DAYS)).isoformat() items = get_cached_requests_since(since_iso) _recent_cache["items"] = items _recent_cache["updated_at"] = datetime.now(timezone.utc).isoformat() def _recent_cache_stale() -> bool: updated_at = _recent_cache.get("updated_at") if not updated_at: return True try: parsed = datetime.fromisoformat(updated_at) except ValueError: return True return (datetime.now(timezone.utc) - parsed).total_seconds() > RECENT_CACHE_TTL_SECONDS def _parse_iso_datetime(value: Optional[str]) -> Optional[datetime]: if not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed def _get_recent_from_cache( requested_by_norm: Optional[str], requested_by_id: Optional[int], limit: int, offset: int, since_iso: Optional[str], status_codes: Optional[list[int]] = None, ) -> List[Dict[str, Any]]: items = _recent_cache.get("items") or [] results = [] since_dt = _parse_iso_datetime(since_iso) for item in items: if requested_by_id is not None: if item.get("requested_by_id") != requested_by_id: continue elif requested_by_norm and item.get("requested_by_norm") != requested_by_norm: continue if since_dt: candidate = item.get("created_at") or item.get("updated_at") item_dt = _parse_iso_datetime(candidate) if not item_dt or item_dt < since_dt: continue if status_codes and item.get("status") not in status_codes: continue results.append(item) return results[offset : offset + limit] async def startup_warmup_requests_cache() -> None: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): try: await _ensure_requests_cache(client) except httpx.HTTPError as exc: logger.warning("Requests warmup skipped: %s", exc) repaired = repair_request_cache_titles() if repaired: logger.info("Requests cache titles repaired: %s", repaired) _refresh_recent_cache_from_db() async def run_requests_poll_loop() -> None: while True: runtime = get_runtime_settings() interval = max(60, int(runtime.requests_poll_interval_seconds or 300)) client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): try: await _ensure_requests_cache(client) except httpx.HTTPError as exc: logger.debug("Requests poll skipped: %s", exc) await asyncio.sleep(interval) async def run_requests_delta_loop() -> None: while True: runtime = get_runtime_settings() interval = max(60, int(runtime.requests_delta_sync_interval_minutes or 5) * 60) client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): if _sync_task and not _sync_task.done(): logger.debug("Delta sync skipped: another sync is running.") else: try: await _sync_delta_requests(client) except httpx.HTTPError as exc: logger.debug("Delta sync skipped: %s", exc) await asyncio.sleep(interval) async def run_daily_requests_full_sync() -> None: while True: runtime = get_runtime_settings() hour, minute = _parse_time(runtime.requests_full_sync_time, 0, 0) await asyncio.sleep(_seconds_until(hour, minute)) runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if not client.configured(): logger.info("Daily full sync skipped: Seerr not configured.") continue if _sync_task and not _sync_task.done(): logger.info("Daily full sync skipped: another sync is running.") continue try: await _sync_all_requests(client) except httpx.HTTPError as exc: logger.warning("Daily full sync failed: %s", exc) async def run_daily_db_cleanup() -> None: while True: runtime = get_runtime_settings() hour, minute = _parse_time(runtime.requests_cleanup_time, 2, 0) await asyncio.sleep(_seconds_until(hour, minute)) runtime = get_runtime_settings() result = cleanup_history(int(runtime.requests_cleanup_days or 90)) logger.info("Daily cleanup complete: %s", result) async def start_requests_sync(base_url: Optional[str], api_key: Optional[str]) -> Dict[str, Any]: global _sync_task if _sync_task and not _sync_task.done(): return dict(_sync_state) if not base_url: _sync_state.update({"status": "failed", "message": "Seerr not configured"}) return dict(_sync_state) client = JellyseerrClient(base_url, api_key) _sync_state.update( { "status": "running", "stored": 0, "total": None, "skip": 0, "message": "Starting sync", "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None, } ) async def _runner() -> None: try: await _sync_all_requests(client) except Exception as exc: logger.exception("Seerr sync failed") _sync_state.update( { "status": "failed", "message": f"Sync failed: {exc}", "finished_at": datetime.now(timezone.utc).isoformat(), } ) _sync_task = asyncio.create_task(_runner()) return dict(_sync_state) async def start_requests_delta_sync(base_url: Optional[str], api_key: Optional[str]) -> Dict[str, Any]: global _sync_task if _sync_task and not _sync_task.done(): return dict(_sync_state) if not base_url: _sync_state.update({"status": "failed", "message": "Seerr not configured"}) return dict(_sync_state) client = JellyseerrClient(base_url, api_key) _sync_state.update( { "status": "running", "stored": 0, "total": None, "skip": 0, "message": "Starting delta sync", "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None, } ) async def _runner() -> None: try: await _sync_delta_requests(client) except Exception as exc: logger.exception("Seerr delta sync failed") _sync_state.update( { "status": "failed", "message": f"Delta sync failed: {exc}", "finished_at": datetime.now(timezone.utc).isoformat(), } ) _sync_task = asyncio.create_task(_runner()) return dict(_sync_state) def get_requests_sync_state() -> Dict[str, Any]: return dict(_sync_state) async def _ensure_request_access( client: JellyseerrClient, request_id: int, user: Dict[str, str] ) -> None: if user.get("role") == "admin" or user.get("username"): return raise HTTPException(status_code=403, detail="Request not accessible for this user") def _build_recent_map(response: Dict[str, Any]) -> Dict[int, Dict[str, Any]]: mapping: Dict[int, Dict[str, Any]] = {} for item in response.get("results", []): media = item.get("media") or {} media_id = media.get("id") or item.get("mediaId") request_id = item.get("id") status = item.get("status") if isinstance(media_id, int) and isinstance(request_id, int): mapping[media_id] = { "requestId": request_id, "status": status, "statusLabel": _status_label(status), } return mapping def _queue_records(queue: Any) -> List[Dict[str, Any]]: if isinstance(queue, dict): records = queue.get("records") if isinstance(records, list): return records if isinstance(queue, list): return queue return [] def _download_ids(records: List[Dict[str, Any]]) -> List[str]: ids = [] for record in records: download_id = record.get("downloadId") or record.get("download_id") if isinstance(download_id, str) and download_id: ids.append(download_id) return ids def _log_arr_http_error(service_label: str, action: str, exc: httpx.HTTPStatusError) -> None: if exc.response is None: logger.warning("%s %s failed: %s", service_label, action, exc) return status = exc.response.status_code body = exc.response.text if isinstance(body, str): body = body.strip() if len(body) > 800: body = f"{body[:800]}...(truncated)" logger.warning("%s %s failed: status=%s body=%s", service_label, action, status, body) def _format_rejections(rejections: Any) -> Optional[str]: if isinstance(rejections, str): return rejections.strip() or None if isinstance(rejections, list): reasons = [] for item in rejections: reason = None if isinstance(item, dict): reason = ( item.get("reason") or item.get("message") or item.get("errorMessage") ) if not reason and item is not None: reason = str(item) if isinstance(reason, str) and reason.strip(): reasons.append(reason.strip()) if reasons: return "; ".join(reasons) return None def _release_push_accepted(response: Any) -> tuple[bool, Optional[str]]: if isinstance(response, list): if not response: return False, "the collector returned no download decision" reasons: List[str] = [] for item in response: accepted, reason = _release_push_accepted(item) if accepted: return True, None if reason: reasons.append(reason) return False, "; ".join(dict.fromkeys(reasons)) or "rejected" if not isinstance(response, dict): return True, None rejections = response.get("rejections") or response.get("rejectionReasons") reason = _format_rejections(rejections) if reason: return False, reason if response.get("rejected") is True: return False, "rejected" if response.get("downloadAllowed") is False: return False, "download not allowed" if response.get("approved") is False: return False, "not approved" return True, None def _filter_arr_release_results(results: Any) -> List[Dict[str, Any]]: if not isinstance(results, list): return [] keep: List[Dict[str, Any]] = [] seen: set[tuple[Any, Any]] = set() for item in results: if not isinstance(item, dict): continue key = (item.get("indexerId"), item.get("guid")) if not key[0] or not key[1] or key in seen: continue seen.add(key) keep.append( { "title": item.get("title"), "indexer": item.get("indexer"), "indexerId": item.get("indexerId"), "guid": item.get("guid"), "size": item.get("size"), "seeders": item.get("seeders"), "leechers": item.get("leechers"), "publishDate": item.get("publishDate"), "infoUrl": item.get("infoUrl"), "downloadUrl": item.get("downloadUrl"), "magnetUrl": item.get("magnetUrl"), "protocol": item.get("protocol"), "approved": item.get("approved"), "rejected": item.get("rejected"), "temporarilyRejected": item.get("temporarilyRejected"), "rejections": item.get("rejections"), "downloadAllowed": item.get("downloadAllowed"), "fullSeason": item.get("fullSeason"), "seasonNumber": item.get("seasonNumber"), } ) return keep[:30] def _build_release_push_payload(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: title = payload.get("title") download_url = payload.get("downloadUrl") magnet_url = payload.get("magnetUrl") protocol = str(payload.get("protocol") or "").strip().lower() if protocol not in {"torrent", "usenet"}: protocol = "torrent" if magnet_url or str(download_url or "").startswith("magnet:") else "usenet" if not isinstance(title, str) or not title.strip() or not download_url and not magnet_url: return None publish_date = payload.get("publishDate") if not isinstance(publish_date, str) or not publish_date.strip(): publish_date = datetime.now(timezone.utc).isoformat() result: Dict[str, Any] = { "title": title.strip(), "protocol": protocol, "publishDate": publish_date, "indexer": payload.get("indexer") or "Magent manual selection", } if isinstance(download_url, str) and download_url.strip(): if download_url.startswith("magnet:"): result["magnetUrl"] = download_url else: result["downloadUrl"] = download_url if isinstance(magnet_url, str) and magnet_url.strip(): result["magnetUrl"] = magnet_url for key in ("guid", "infoUrl", "size", "seeders", "leechers"): if payload.get(key) is not None: result[key] = payload[key] return result def _missing_episode_ids_by_season(episodes: Any) -> Dict[int, List[int]]: if not isinstance(episodes, list): return {} grouped: Dict[int, List[int]] = {} for episode in episodes: if not isinstance(episode, dict): continue if not episode.get("monitored", True): continue if episode.get("hasFile"): continue season_number = episode.get("seasonNumber") episode_id = episode.get("id") if isinstance(season_number, int) and isinstance(episode_id, int): grouped.setdefault(season_number, []).append(episode_id) return grouped def _replacement_file_name(file_data: Dict[str, Any]) -> str: raw = file_data.get("relativePath") or file_data.get("path") or file_data.get("sceneName") if isinstance(raw, str) and raw.strip(): return raw.strip().replace("\\", "/").rsplit("/", 1)[-1] return "Managed media file" def _replacement_quality_name(file_data: Dict[str, Any]) -> Optional[str]: quality = file_data.get("quality") if not isinstance(quality, dict): return None nested = quality.get("quality") if isinstance(nested, dict): value = nested.get("name") return str(value).strip() if value is not None and str(value).strip() else None value = quality.get("name") return str(value).strip() if value is not None and str(value).strip() else None def _replacement_file_payload( file_data: Dict[str, Any], *, episode_numbers: Optional[List[str]] = None, ) -> Optional[Dict[str, Any]]: file_id = file_data.get("id") if not isinstance(file_id, int) or file_id <= 0: return None size = file_data.get("size") return { "id": file_id, "name": _replacement_file_name(file_data), "quality": _replacement_quality_name(file_data), "size": int(size) if isinstance(size, (int, float)) and size >= 0 else None, "season_number": file_data.get("seasonNumber") if isinstance(file_data.get("seasonNumber"), int) else None, "episodes": episode_numbers or [], } def _linked_issue_for_replacement( issue_id: Any, *, request_id: str, user: Dict[str, str], ) -> Optional[Dict[str, Any]]: if issue_id is None: return None if not isinstance(issue_id, int) or issue_id <= 0: raise HTTPException(status_code=400, detail="A valid linked issue is required") issue = get_portal_item(issue_id) if not issue or str(issue.get("kind") or "").lower() != "issue": raise HTTPException(status_code=404, detail="Linked issue not found") if str(issue.get("external_ref") or "") != f"/requests/{request_id}": raise HTTPException(status_code=409, detail="The issue is not linked to this request") is_admin = str(user.get("role") or "").lower() == "admin" is_owner = str(issue.get("created_by_username") or "").lower() == str(user.get("username") or "").lower() if not (is_admin or is_owner): raise HTTPException(status_code=403, detail="You cannot update this linked issue") return issue def _record_replacement_activity( issue: Optional[Dict[str, Any]], *, user: Dict[str, str], event_type: str, message: str, ) -> None: if not issue: return current_status = str(issue.get("status") or "new").strip().lower() next_status: Optional[str] = None if event_type.endswith("_started"): next_status = "in_progress" elif event_type.endswith("_failed"): next_status = "blocked" if next_status and current_status not in {"done", "closed"}: update_portal_item( int(issue["id"]), status=next_status, issue_resolved_at=None, ) add_portal_item_activity( int(issue["id"]), event_type=event_type, actor_username=str(user.get("username") or "unknown"), actor_role=str(user.get("role") or "user"), message=message, ) def _released_episode(episode: Dict[str, Any]) -> bool: if episode.get("hasFile") is True: return True raw_date = episode.get("airDateUtc") or episode.get("airDate") if not isinstance(raw_date, str) or not raw_date.strip(): return False try: parsed = datetime.fromisoformat(raw_date.strip().replace("Z", "+00:00")) except ValueError: return False if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) <= datetime.now(timezone.utc) def _issue_episode_payloads( episodes: Any, ) -> List[Dict[str, Any]]: results: List[Dict[str, Any]] = [] if not isinstance(episodes, list): return results for episode in episodes: if not isinstance(episode, dict): continue episode_id = episode.get("id") season_number = episode.get("seasonNumber") episode_number = episode.get("episodeNumber") if not all(isinstance(value, int) for value in (episode_id, season_number, episode_number)): continue released = _released_episode(episode) has_file = episode.get("hasFile") is True or ( isinstance(episode.get("episodeFileId"), int) and episode.get("episodeFileId") > 0 ) monitored = episode.get("monitored") is not False results.append( { "id": episode_id, "season_number": season_number, "episode_number": episode_number, "code": f"S{season_number:02d}E{episode_number:02d}", "title": str(episode.get("title") or f"Episode {episode_number}").strip(), "released": released, "monitored": monitored, "has_file": has_file, "missing": released and monitored and not has_file, "best_fit": released and monitored and not has_file, "file_id": episode.get("episodeFileId") if has_file else None, } ) results.sort(key=lambda item: (item["season_number"], item["episode_number"])) return results def _issue_season_payloads(episodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]: grouped: Dict[int, List[Dict[str, Any]]] = {} for episode in episodes: grouped.setdefault(int(episode["season_number"]), []).append(episode) return [ { "season_number": season_number, "label": "Specials" if season_number == 0 else f"Season {season_number}", "episode_count": len(items), "available_count": sum(1 for item in items if item["has_file"]), "missing_count": sum(1 for item in items if item["missing"]), "best_fit": any(item["best_fit"] for item in items), } for season_number, items in sorted(grouped.items()) if any(item["released"] for item in items) ] async def _resolve_root_folder_path(client: Any, root_folder: str, service_name: str) -> str: if root_folder.isdigit(): folders = await client.get_root_folders() if isinstance(folders, list): for folder in folders: if folder.get("id") == int(root_folder): path = folder.get("path") if isinstance(path, str) and path: return path raise HTTPException(status_code=400, detail=f"{service_name} root folder id {root_folder} not found") return root_folder @router.get("/{request_id}/issue-options") async def issue_target_options( request_id: str, user: Dict[str, str] = Depends(get_current_user), ) -> Dict[str, Any]: if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") runtime = get_runtime_settings() seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if seerr.configured(): await _ensure_request_access(seerr, int(request_id), user) snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") if not isinstance(arr_item, dict): return { "request_id": request_id, "request_type": snapshot.request_type.value, "title": snapshot.title, "collector_id": None, "movie": None, "seasons": [], "episodes": [], "can_act": False, "message": "This title is not currently linked to Sonarr or Radarr.", } collector_id = arr_item.get("id") if not isinstance(collector_id, int): raise HTTPException(status_code=502, detail="Sonarr/Radarr returned an invalid media record") if snapshot.request_type == RequestType.movie: movie_file = arr_item.get("movieFile") if isinstance(arr_item.get("movieFile"), dict) else None return { "request_id": request_id, "request_type": "movie", "title": snapshot.title, "collector_id": collector_id, "movie": { "selected_label": snapshot.title, "has_file": bool(movie_file), "missing": not bool(movie_file), "best_fit": not bool(movie_file), "file_id": movie_file.get("id") if movie_file else None, }, "seasons": [], "episodes": [], "can_act": _user_can_use_search_auto(user), "message": "Choose the movie to continue.", } sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not sonarr.configured(): raise HTTPException(status_code=400, detail="Sonarr is not configured") try: episodes = await sonarr.get_episodes(collector_id) except Exception as exc: logger.warning("Sonarr issue options failed request_id=%s error=%s", request_id, exc) raise HTTPException( status_code=502, detail="Magent could not read the seasons and episodes from Sonarr.", ) from exc episode_options = _issue_episode_payloads(episodes) return { "request_id": request_id, "request_type": "tv", "title": snapshot.title, "collector_id": collector_id, "movie": None, "seasons": _issue_season_payloads(episode_options), "episodes": episode_options, "can_act": _user_can_use_search_auto(user), "message": "Choose a season, then select every affected episode.", } @router.get("/{request_id}/replacement-options") async def replacement_options( request_id: str, user: Dict[str, str] = Depends(get_current_user), ) -> Dict[str, Any]: if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") runtime = get_runtime_settings() seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if seerr.configured(): await _ensure_request_access(seerr, int(request_id), user) snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") if not isinstance(arr_item, dict): return { "request_id": request_id, "request_type": snapshot.request_type.value, "title": snapshot.title, "files": [], "message": "This title is not currently linked to a Sonarr/Radarr library item.", } files: List[Dict[str, Any]] = [] if snapshot.request_type == RequestType.movie: movie_file = arr_item.get("movieFile") if isinstance(movie_file, dict): option = _replacement_file_payload(movie_file) if option: files.append(option) elif snapshot.request_type == RequestType.tv: series_id = arr_item.get("id") if not isinstance(series_id, int): raise HTTPException(status_code=502, detail="Sonarr returned an invalid series record") sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not sonarr.configured(): raise HTTPException(status_code=400, detail="Sonarr is not configured") try: episode_files, episodes = await asyncio.gather( sonarr.get_episode_files(series_id), sonarr.get_episodes(series_id), ) except Exception as exc: logger.warning("Sonarr replacement options failed request_id=%s error=%s", request_id, exc) raise HTTPException( status_code=502, detail="Magent could not read the managed episode files from Sonarr.", ) from exc episode_labels: Dict[int, List[str]] = {} if isinstance(episodes, list): for episode in episodes: if not isinstance(episode, dict): continue file_id = episode.get("episodeFileId") season_number = episode.get("seasonNumber") episode_number = episode.get("episodeNumber") if not all(isinstance(value, int) for value in (file_id, season_number, episode_number)): continue label = f"S{season_number:02d}E{episode_number:02d}" episode_labels.setdefault(file_id, []).append(label) if isinstance(episode_files, list): for file_data in episode_files: if not isinstance(file_data, dict): continue option = _replacement_file_payload( file_data, episode_numbers=episode_labels.get(file_data.get("id"), []), ) if option: files.append(option) files.sort( key=lambda item: ( item.get("season_number") if isinstance(item.get("season_number"), int) else 9999, ",".join(item.get("episodes") or []), str(item.get("name") or ""), ) ) return { "request_id": request_id, "request_type": snapshot.request_type.value, "title": snapshot.title, "files": files, "can_replace": _user_can_use_search_auto(user), "message": ( "Choose the exact managed file to remove and replace." if files else "Sonarr/Radarr does not currently report a managed file for this title." ), } @router.post("/{request_id}/actions/replace") async def action_replace_media( request_id: str, payload: Dict[str, Any], user: Dict[str, str] = Depends(get_current_user), ) -> Dict[str, Any]: if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") if not _user_can_use_search_auto(user): raise HTTPException(status_code=403, detail="Media replacement is disabled for this user") if payload.get("confirmed") is not True: raise HTTPException(status_code=400, detail="Replacement confirmation is required") raw_file_ids = payload.get("file_ids") if raw_file_ids is None: raw_file_ids = [payload.get("file_id")] if not isinstance(raw_file_ids, list): raise HTTPException(status_code=400, detail="Managed files must be supplied as a list") file_ids = list(dict.fromkeys( value for value in raw_file_ids if isinstance(value, int) and not isinstance(value, bool) and value > 0 )) if not file_ids or len(file_ids) != len(raw_file_ids) or len(file_ids) > 100: raise HTTPException(status_code=400, detail="Choose between 1 and 100 valid managed files") linked_issue = _linked_issue_for_replacement( payload.get("issue_id"), request_id=request_id, user=user, ) runtime = get_runtime_settings() seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if seerr.configured(): await _ensure_request_access(seerr, int(request_id), user) snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") if not isinstance(arr_item, dict): raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr") collector = "Sonarr" if snapshot.request_type == RequestType.tv else "Radarr" target_names: List[str] = [] try: if snapshot.request_type == RequestType.movie: movie_id = arr_item.get("id") movie_file = arr_item.get("movieFile") if not isinstance(movie_id, int) or not isinstance(movie_file, dict): 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") target_names = [_replacement_file_name(movie_file)] radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) if not radarr.configured(): raise HTTPException(status_code=400, detail="Radarr is not configured") await radarr.delete_movie_file(file_ids[0]) await radarr.search(movie_id) elif snapshot.request_type == RequestType.tv: series_id = arr_item.get("id") if not isinstance(series_id, int): raise HTTPException(status_code=502, detail="Sonarr returned an invalid series record") sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not sonarr.configured(): raise HTTPException(status_code=400, detail="Sonarr is not configured") episode_files, episodes = await asyncio.gather( sonarr.get_episode_files(series_id), sonarr.get_episodes(series_id), ) selected_files = [ item for item in episode_files if isinstance(item, dict) and item.get("id") in file_ids ] 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") for episode in episodes if isinstance(episode, dict) and episode.get("episodeFileId") in file_ids and isinstance(episode.get("id"), int) ] if isinstance(episodes, list) else [] if not episode_ids: raise HTTPException(status_code=409, detail="Sonarr could not match episodes to this file") target_names = [_replacement_file_name(file_data) for file_data in selected_files] for selected_file_id in file_ids: await sonarr.delete_episode_file(selected_file_id) await sonarr.search_episodes(episode_ids) else: raise HTTPException(status_code=400, detail="Unknown request type") except HTTPException as exc: _record_replacement_activity( linked_issue, user=user, event_type="replacement_failed", message=f"The media replacement could not be started: {exc.detail}", ) raise except Exception as exc: logger.exception("%s media replacement failed request_id=%s file_ids=%s", collector, request_id, file_ids) detail = ( f"{collector} could not complete the replacement. Check the request action history " "before trying again." ) await asyncio.to_thread( save_action, request_id, "replace_media", "Replace media file", "failed", detail, ) _record_replacement_activity( linked_issue, user=user, event_type="replacement_failed", message=detail, ) raise HTTPException(status_code=502, detail=detail) from exc file_label = "the selected managed file" if len(target_names) == 1 else f"{len(target_names)} selected managed files" message = f"{collector} removed {file_label} and started a replacement search." await asyncio.to_thread( save_action, request_id, "replace_media", "Replace media file", "ok", message, ) _record_replacement_activity( linked_issue, user=user, event_type="replacement_started", message=message, ) return { "status": "ok", "message": message, "collector": collector, "request_id": request_id, "file_ids": file_ids, } def _positive_id_list( value: Any, *, field: str, maximum: int = 200, minimum: int = 1, ) -> List[int]: if value is None: return [] if not isinstance(value, list): raise HTTPException(status_code=400, detail=f"{field} must be a list") normalized = list( dict.fromkeys( item for item in value if isinstance(item, int) and not isinstance(item, bool) and item >= minimum ) ) if len(normalized) != len(value) or len(normalized) > maximum: raise HTTPException(status_code=400, detail=f"Choose up to {maximum} valid {field.replace('_', ' ')}") return normalized @router.post("/{request_id}/actions/search-missing") async def action_search_missing_media( request_id: str, payload: Dict[str, Any], user: Dict[str, str] = Depends(get_current_user), ) -> Dict[str, Any]: if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") if not _user_can_use_search_auto(user): raise HTTPException(status_code=403, detail="Collection searches are disabled for this user") linked_issue = _linked_issue_for_replacement( payload.get("issue_id"), request_id=request_id, user=user ) episode_ids = _positive_id_list(payload.get("episode_ids"), field="episode_ids") season_numbers = _positive_id_list( payload.get("season_numbers"), field="season_numbers", maximum=100, minimum=0 ) runtime = get_runtime_settings() snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") 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"]) try: if snapshot.request_type == RequestType.movie: radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) if not radarr.configured(): raise HTTPException(status_code=400, detail="Radarr is not configured") await radarr.search(collector_id) message = "Radarr started searching for the missing movie." searched_ids: List[int] = [] else: sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not sonarr.configured(): raise HTTPException(status_code=400, detail="Sonarr is not configured") episodes = await sonarr.get_episodes(collector_id) if not isinstance(episodes, list): raise HTTPException(status_code=502, detail="Sonarr did not return an episode list") episode_map = { int(item["id"]): item for item in episodes if isinstance(item, dict) and isinstance(item.get("id"), int) } if episode_ids: if any(item_id not in episode_map for item_id in episode_ids): raise HTTPException(status_code=409, detail="One or more selected episodes are no longer in Sonarr") searched_ids = episode_ids else: searched_ids = [ item_id for item_id, episode in episode_map.items() if _released_episode(episode) and episode.get("monitored") is not False and not ( episode.get("hasFile") is True or (isinstance(episode.get("episodeFileId"), int) and episode.get("episodeFileId") > 0) ) and (not season_numbers or episode.get("seasonNumber") in season_numbers) ] if searched_ids: await sonarr.search_episodes(searched_ids) message = f"Sonarr started searching for {len(searched_ids)} selected missing episode(s)." else: await sonarr.search(collector_id) message = "Sonarr refreshed the series and started a full missing-episode search." except HTTPException as exc: _record_replacement_activity( linked_issue, user=user, event_type="missing_search_failed", message=f"The missing-content search could not start: {exc.detail}", ) raise except Exception as exc: logger.exception("missing content search failed request_id=%s", request_id) detail = "Sonarr/Radarr could not start the missing-content search." _record_replacement_activity( linked_issue, user=user, event_type="missing_search_failed", message=detail ) raise HTTPException(status_code=502, detail=detail) from exc await asyncio.to_thread( 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 ) return {"status": "ok", "message": message, "episode_ids": searched_ids} @router.post("/{request_id}/actions/repair-subtitles") async def action_repair_subtitles( request_id: str, payload: Dict[str, Any], user: Dict[str, str] = Depends(get_current_user), ) -> Dict[str, Any]: if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") if not _user_can_use_search_auto(user): raise HTTPException(status_code=403, detail="Subtitle repairs are disabled for this user") linked_issue = _linked_issue_for_replacement( payload.get("issue_id"), request_id=request_id, user=user ) episode_ids = _positive_id_list(payload.get("episode_ids"), field="episode_ids", maximum=100) forced = payload.get("forced") is True runtime = get_runtime_settings() bazarr = BazarrClient(runtime.bazarr_base_url, runtime.bazarr_api_key) if not bazarr.configured() or not runtime.bazarr_api_key: raise HTTPException(status_code=400, detail="Bazarr is not configured") language = str(runtime.bazarr_default_language or "en").strip().lower() or "en" snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") 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"]) try: if snapshot.request_type == RequestType.movie: await bazarr.search_movie_subtitles( collector_id, language=language, forced=forced ) repaired_count = 1 message = f"Bazarr started a fresh {language.upper()} subtitle search for the movie." else: if not episode_ids: raise HTTPException(status_code=400, detail="Choose at least one episode") sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) episodes = await sonarr.get_episodes(collector_id) valid_ids = { int(item["id"]) for item in episodes if isinstance(item, dict) and isinstance(item.get("id"), int) } if isinstance(episodes, list) else set() if any(episode_id not in valid_ids for episode_id in episode_ids): raise HTTPException(status_code=409, detail="One or more selected episodes are no longer in Sonarr") for episode_id in episode_ids: await bazarr.search_episode_subtitles( collector_id, episode_id, language=language, forced=forced, ) repaired_count = len(episode_ids) message = f"Bazarr started fresh {language.upper()} subtitle searches for {repaired_count} episode(s)." except HTTPException as exc: _record_replacement_activity( linked_issue, user=user, event_type="subtitle_repair_failed", message=f"The subtitle repair could not start: {exc.detail}", ) raise except Exception as exc: logger.exception("Bazarr subtitle repair failed request_id=%s", request_id) detail = "Bazarr could not start the subtitle repair." _record_replacement_activity( linked_issue, user=user, event_type="subtitle_repair_failed", message=detail ) raise HTTPException(status_code=502, detail=detail) from exc await asyncio.to_thread( save_action, request_id, "repair_subtitles", "Repair subtitles", "ok", message ) _record_replacement_activity( linked_issue, user=user, event_type="subtitle_repair_started", message=message ) return {"status": "ok", "message": message, "count": repaired_count} @router.get("/{request_id}/snapshot", response_model=Snapshot) async def get_snapshot(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> Snapshot: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = await build_snapshot(request_id) return _filter_snapshot_for_user(snapshot, user) @router.post("/{request_id}/actions/recheck") async def action_recheck(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict: if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") runtime = get_runtime_settings() seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if not seerr.configured(): raise HTTPException(status_code=400, detail="Seerr is not configured") await _ensure_request_access(seerr, int(request_id), user) try: fresh_request = await seerr.get_request(request_id) except httpx.HTTPStatusError as exc: detail = _format_upstream_error("Seerr", exc) await asyncio.to_thread( save_action, request_id, "recheck_pipeline", "Recheck request status", "failed", detail, ) raise HTTPException(status_code=502, detail=detail) from exc except Exception as exc: logger.warning("Request recheck failed while reading Seerr: request_id=%s error=%s", request_id, exc) detail = "Magent could not reach Seerr to recheck this request." await asyncio.to_thread( save_action, request_id, "recheck_pipeline", "Recheck request status", "failed", detail, ) raise HTTPException(status_code=502, detail=detail) from exc if not isinstance(fresh_request, dict): raise HTTPException(status_code=404, detail="Request not found in Seerr") parsed = _parse_request_payload(fresh_request) if not parsed.get("title"): details = await _get_media_details( seerr, parsed.get("media_type"), parsed.get("tmdb_id") ) if isinstance(details, dict): fresh_request = _merge_request_media_details(fresh_request, details) parsed = _parse_request_payload(fresh_request) if parsed.get("request_id") != int(request_id): raise HTTPException(status_code=502, detail="Seerr returned an unexpected request record") cache_record = _build_request_cache_record(parsed, fresh_request) await asyncio.to_thread(upsert_request_cache, **cache_record) _cache_set(f"request:{request_id}", fresh_request) _refresh_recent_cache_from_db() snapshot = _filter_snapshot_for_user(await build_snapshot(request_id), user) status_label = str((snapshot.presentation.get("status") or {}).get("label") or "Status updated") message = f"Recheck complete. {status_label}." await asyncio.to_thread( save_action, request_id, "recheck_pipeline", "Recheck request status", "ok", message, ) return {"status": "ok", "message": message, "snapshot": snapshot} @router.get("/{request_id}/download-progress") async def get_download_progress( request_id: str, user: Dict[str, str] = Depends(get_current_user) ) -> Dict[str, Any]: """Return a lightweight qBittorrent update for an open request page.""" if not request_id.isdigit(): raise HTTPException(status_code=400, detail="Invalid request id") runtime = get_runtime_settings() seerr = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if seerr.configured(): await _ensure_request_access(seerr, int(request_id), user) 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] = [] if isinstance(historical_torrents, list): hashes = list( dict.fromkeys( str(torrent.get("hash") or "").strip() for torrent in historical_torrents if isinstance(torrent, dict) and torrent.get("hash") ) ) qbittorrent = QBittorrentClient( runtime.qbittorrent_base_url, runtime.qbittorrent_username, runtime.qbittorrent_password, ) if not qbittorrent.configured(): raise HTTPException(status_code=503, detail="qBittorrent is not configured") try: if hashes: result = await qbittorrent.get_torrents_by_hashes("|".join(hashes)) else: result = await qbittorrent.get_torrents_by_tag(f"magent-{request_id}") except Exception as exc: 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 [] for torrent in torrents: if isinstance(torrent, dict): torrent["progressPercent"] = _torrent_progress(torrent) if torrents: summary = _summarize_qbit(torrents) state = str(summary.get("state") or "idle") message = str(summary.get("message") or "Download found in qBittorrent.") elif evidence.get("observed"): state = "missing" message = "The previous download is no longer visible in qBittorrent." else: state = "not_started" message = "No download attempt has been observed." return { "request_id": request_id, "state": state, "summary": message, "torrents": torrents, "updated_at": datetime.now(timezone.utc).isoformat(), } @router.get("/recent") async def recent_requests( take: int = 6, skip: int = 0, days: int = 90, stage: str = "all", user: Dict[str, str] = Depends(get_current_user), ) -> dict: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) mode = (runtime.requests_data_source or "prefer_cache").lower() allow_remote = mode == "always_js" if allow_remote: if not client.configured(): raise HTTPException(status_code=400, detail="Seerr not configured") try: await _ensure_requests_cache(client) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc username_norm = _normalize_username(user.get("username", "")) requested_by_id = user.get("jellyseerr_user_id") requested_by = None if user.get("role") == "admin" else username_norm requested_by_id = None if user.get("role") == "admin" else requested_by_id since_iso = None if days > 0: since_iso = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() status_codes = request_stage_filter_codes(stage) if _recent_cache_stale(): _refresh_recent_cache_from_db() rows = _get_recent_from_cache( requested_by, requested_by_id, take, skip, since_iso, status_codes=status_codes, ) cache_mode = (runtime.artwork_cache_mode or "remote").lower() allow_title_hydrate = False allow_artwork_hydrate = client.configured() jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) jellyfin_cache: Dict[str, bool] = {} results = [] for row in rows: status = row.get("status") title = row.get("title") title_is_placeholder = ( isinstance(title, str) and row.get("request_id") is not None and title.strip().lower() == f"request {row.get('request_id')}" ) year = row.get("year") details = None if row.get("request_id") and mode != "always_js": cached_payload = get_request_cache_payload(int(row["request_id"])) if isinstance(cached_payload, dict): details = cached_payload if (not title or title_is_placeholder) and row.get("request_id"): if details is None and (allow_remote or allow_title_hydrate): details = await _get_request_details(client, int(row["request_id"])) if isinstance(details, dict): payload = _parse_request_payload(details) title = payload.get("title") or title year = payload.get("year") or year if not title and payload.get("tmdb_id") and (allow_remote or allow_title_hydrate): hydrated_title, hydrated_year = await _hydrate_title_from_tmdb( client, payload.get("media_type"), payload.get("tmdb_id") ) if hydrated_title: title = hydrated_title if hydrated_year: year = hydrated_year if allow_remote and isinstance(payload.get("request_id"), int): upsert_request_cache( request_id=payload.get("request_id"), media_id=payload.get("media_id"), media_type=payload.get("media_type"), status=payload.get("status"), title=title or payload.get("title"), year=year or payload.get("year"), requested_by=payload.get("requested_by"), requested_by_norm=payload.get("requested_by_norm"), requested_by_id=payload.get("requested_by_id"), created_at=payload.get("created_at"), updated_at=payload.get("updated_at"), payload_json=json.dumps(details, ensure_ascii=True), ) row["title"] = title row["year"] = year row["media_type"] = payload.get("media_type") or row.get("media_type") row["status"] = payload.get("status") or row.get("status") if details is None and row.get("request_id") and allow_remote: details = await _get_request_details(client, int(row["request_id"])) poster_path = None backdrop_path = None if isinstance(details, dict): media = details.get("media") or {} if isinstance(media, dict): poster_path = media.get("posterPath") or media.get("poster_path") backdrop_path = media.get("backdropPath") or media.get("backdrop_path") tmdb_id = media.get("tmdbId") or details.get("tmdbId") else: tmdb_id = details.get("tmdbId") media_type = media.get("mediaType") if isinstance(media, dict) else None media_type = media_type or details.get("type") or row.get("media_type") if not poster_path and tmdb_id and allow_artwork_hydrate: hydrated_poster, hydrated_backdrop = await _hydrate_artwork_from_tmdb( client, media_type, tmdb_id ) poster_path = poster_path or hydrated_poster backdrop_path = backdrop_path or hydrated_backdrop if (hydrated_poster or hydrated_backdrop) and isinstance(details, dict): media = dict(media) if isinstance(media, dict) else {} if hydrated_poster: media["posterPath"] = hydrated_poster if hydrated_backdrop: media["backdropPath"] = hydrated_backdrop details["media"] = media payload = _parse_request_payload(details) if isinstance(payload.get("request_id"), int): upsert_request_cache( request_id=payload.get("request_id"), media_id=payload.get("media_id"), media_type=payload.get("media_type"), status=payload.get("status"), title=payload.get("title"), year=payload.get("year"), requested_by=payload.get("requested_by"), requested_by_norm=payload.get("requested_by_norm"), requested_by_id=payload.get("requested_by_id"), created_at=payload.get("created_at"), updated_at=payload.get("updated_at"), 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"}: is_available = await _request_is_available_in_jellyfin( jellyfin, title, year, row.get("media_type"), details if isinstance(details, dict) else None, jellyfin_cache, ) status_label = _status_label_with_jellyfin(status, is_available) results.append( { "id": row.get("request_id"), "title": title, "year": year, "type": row.get("media_type"), "status": status, "statusLabel": status_label, "mediaId": row.get("media_id"), "createdAt": row.get("created_at") or row.get("updated_at"), "artwork": { "poster_url": _artwork_url(poster_path, "w185", cache_mode), "backdrop_url": _artwork_url(backdrop_path, "w780", cache_mode), }, } ) return {"results": results} @router.get("/search") async def search_requests( query: str, page: int = 1, media_type: Optional[str] = None, user: Dict[str, str] = Depends(get_current_user), ) -> dict: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Seerr not configured") try: response = await client.search(query=query, page=page) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc if not isinstance(response, dict): return {"results": []} try: await _ensure_requests_cache(client) except httpx.HTTPStatusError: pass requested_media_type = _normalize_media_type(media_type) if media_type is not None else None if media_type is not None and requested_media_type is None: raise HTTPException(status_code=400, detail="media_type must be 'movie' or 'tv'") results = [] jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) jellyfin_cache: Dict[str, bool] = {} for item in response.get("results", []): media_type = item.get("mediaType") if requested_media_type is not None and media_type != requested_media_type: continue title = item.get("title") or item.get("name") year = None if item.get("releaseDate"): year = int(item["releaseDate"][:4]) if item.get("firstAirDate"): year = int(item["firstAirDate"][:4]) request_id = None status = None status_label = None requested_by = None accessible = False media_info = item.get("mediaInfo") or {} media_info_id = media_info.get("id") requests = media_info.get("requests") if isinstance(requests, list) and requests: request_id = requests[0].get("id") status = requests[0].get("status") status_label = _status_label(status) elif isinstance(media_info_id, int): cached = get_cached_request_by_media_id( media_info_id, ) if cached: request_id = cached.get("request_id") status = cached.get("status") status_label = _status_label(status) if isinstance(request_id, int): details = get_request_cache_payload(request_id) if not isinstance(details, dict): details = await _get_request_details(client, request_id) if user.get("role") == "admin": requested_by = _request_display_name(details) accessible = True if status is not None: is_available = await _request_is_available_in_jellyfin( jellyfin, title, year, media_type, details if isinstance(details, dict) else None, jellyfin_cache, ) status_label = _status_label_with_jellyfin(status, is_available) results.append( { "title": title, "year": year, "type": media_type, "tmdbId": item.get("id"), "requestId": request_id, "status": status, "statusLabel": status_label, "requestedBy": requested_by, "accessible": accessible, "overview": item.get("overview"), "posterPath": item.get("posterPath") or item.get("poster_path"), "backdropPath": item.get("backdropPath") or item.get("backdrop_path"), } ) return {"results": results} @router.get("/request-options") async def request_options( media_type: str, tmdb_id: int, user: Dict[str, str] = Depends(get_current_user), ) -> Dict[str, Any]: del user normalized_media_type = _normalize_media_type(media_type) if normalized_media_type is None: raise HTTPException(status_code=400, detail="media_type must be 'movie' or 'tv'") if tmdb_id <= 0: raise HTTPException(status_code=400, detail="tmdb_id must be a positive integer") runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Seerr not configured") try: details, destination = await asyncio.gather( client.get_movie(tmdb_id) if normalized_media_type == "movie" else client.get_tv(tmdb_id), _resolve_request_destination(runtime, client, normalized_media_type), ) except HTTPException: raise except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=_format_upstream_error("Seerr", exc)) from exc if not isinstance(details, dict): raise HTTPException(status_code=502, detail="Seerr returned invalid media details") title = str(details.get("title") or details.get("name") or "Untitled") date_value = details.get("releaseDate") or details.get("firstAirDate") year = int(date_value[:4]) if isinstance(date_value, str) and len(date_value) >= 4 and date_value[:4].isdigit() else None seasons: list[Dict[str, Any]] = [] if normalized_media_type == "tv": for season in details.get("seasons", []): if not isinstance(season, dict): continue season_number = _quality_profile_id(season.get("seasonNumber")) if season_number is None or season_number <= 0: continue seasons.append( { "seasonNumber": season_number, "name": str(season.get("name") or f"Season {season_number}"), "episodeCount": _quality_profile_id(season.get("episodeCount")) or 0, "airDate": season.get("airDate"), } ) media_info = details.get("mediaInfo") if isinstance(details.get("mediaInfo"), dict) else {} requests_list = media_info.get("requests") existing_request_id = None if isinstance(requests_list, list) and requests_list and isinstance(requests_list[0], dict): existing_request_id = _quality_profile_id(requests_list[0].get("id")) return { "media": { "title": title, "year": year, "type": normalized_media_type, "tmdbId": tmdb_id, "overview": details.get("overview"), "posterPath": details.get("posterPath") or details.get("poster_path"), "backdropPath": details.get("backdropPath") or details.get("backdrop_path"), "seasons": seasons, "existingRequestId": existing_request_id, }, "destination": { "collector": destination["collector"], "serverName": destination["server_name"], "defaultProfileId": destination["default_profile_id"], "profiles": destination["profiles"], }, } @router.post("/create") async def create_request( payload: Dict[str, Any], user: Dict[str, Any] = Depends(get_current_user) ) -> Dict[str, Any]: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Seerr not configured") media_type = _normalize_media_type( payload.get("mediaType") or payload.get("type") or payload.get("media_type") ) if media_type is None: raise HTTPException(status_code=400, detail="mediaType must be 'movie' or 'tv'") raw_tmdb_id = payload.get("tmdbId") if raw_tmdb_id is None: raw_tmdb_id = payload.get("mediaId") if raw_tmdb_id is None: raw_tmdb_id = payload.get("id") try: tmdb_id = int(raw_tmdb_id) except (TypeError, ValueError) as exc: raise HTTPException(status_code=400, detail="tmdbId must be a valid integer") from exc if tmdb_id <= 0: raise HTTPException(status_code=400, detail="tmdbId must be a positive integer") seasons = _normalize_seasons(payload.get("seasons")) if media_type == "tv" else [] raw_profile_id = payload.get("profileId") profile_id = _quality_profile_id(raw_profile_id) if raw_profile_id is not None else None if raw_profile_id is not None and (profile_id is None or profile_id <= 0): raise HTTPException(status_code=400, detail="profileId must be a positive integer") raw_is_4k = payload.get("is4k") if raw_is_4k is not None and not isinstance(raw_is_4k, bool): raise HTTPException(status_code=400, detail="is4k must be true or false") is_4k = raw_is_4k if isinstance(raw_is_4k, bool) else None try: details = await (client.get_movie(tmdb_id) if media_type == "movie" else client.get_tv(tmdb_id)) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=_format_upstream_error("Seerr", exc)) from exc if not isinstance(details, dict): raise HTTPException(status_code=502, detail="Invalid response from Seerr media lookup") media_info = details.get("mediaInfo") if isinstance(details.get("mediaInfo"), dict) else {} requests_list = media_info.get("requests") existing_request: Optional[Dict[str, Any]] = None if isinstance(requests_list, list) and requests_list: first_request = requests_list[0] if isinstance(first_request, dict): existing_request = first_request title = details.get("title") or details.get("name") year: Optional[int] = None date_value = details.get("releaseDate") or details.get("firstAirDate") if isinstance(date_value, str) and len(date_value) >= 4 and date_value[:4].isdigit(): year = int(date_value[:4]) if isinstance(existing_request, dict): existing_request_id = _quality_profile_id(existing_request.get("id")) existing_status = existing_request.get("status") if existing_request_id is not None: request_payload = await _get_request_details(client, existing_request_id) if isinstance(request_payload, dict): parsed_payload = _parse_request_payload(request_payload) upsert_request_cache(**_build_request_cache_record(parsed_payload, request_payload)) _cache_set(f"request:{existing_request_id}", request_payload) title = parsed_payload.get("title") or title year = parsed_payload.get("year") or year return { "status": "exists", "requestId": existing_request_id, "type": media_type, "tmdbId": tmdb_id, "title": title, "year": year, "statusCode": existing_status, "statusLabel": _status_label(existing_status), } if media_type == "tv" and seasons: valid_seasons = { _quality_profile_id(item.get("seasonNumber")) for item in details.get("seasons", []) if isinstance(item, dict) } invalid_seasons = [season for season in seasons if season not in valid_seasons] if invalid_seasons: raise HTTPException( status_code=400, detail=f"Season selection is not available for this series: {invalid_seasons}", ) destination = await _resolve_request_destination(runtime, client, media_type, profile_id) try: created = await client.create_request( media_type=media_type, media_id=tmdb_id, seasons=seasons if media_type == "tv" else None, is_4k=is_4k, server_id=destination["server_id"], profile_id=destination["profile_id"], root_folder=destination["root_folder"], ) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=_format_upstream_error("Seerr", exc)) from exc if not isinstance(created, dict): raise HTTPException(status_code=502, detail="Invalid response from Seerr request create") created = _merge_request_media_details(created, details) parsed = _parse_request_payload(created) request_id = _quality_profile_id(parsed.get("request_id")) status_code = parsed.get("status") title = parsed.get("title") or title year = parsed.get("year") or year if request_id is not None: upsert_request_cache(**_build_request_cache_record(parsed, created)) _cache_set(f"request:{request_id}", created) _recent_cache["updated_at"] = None await asyncio.to_thread( save_action, str(request_id), "request_created", "Create request", "ok", f"{media_type} request created from discovery by {user.get('username')}.", ) return { "status": "created", "requestId": request_id, "type": media_type, "tmdbId": tmdb_id, "title": title, "year": year, "statusCode": status_code, "statusLabel": _status_label(status_code), } @router.post("/{request_id}/ai/triage", response_model=TriageResult) async def ai_triage(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> TriageResult: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = _filter_snapshot_for_user(await build_snapshot(request_id), user) return triage_snapshot(snapshot) @router.post("/{request_id}/actions/search") async def action_search(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") 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") results: List[Dict[str, Any]] = [] collector = "Sonarr" if snapshot.request_type == RequestType.tv else "Radarr" try: if snapshot.request_type == RequestType.tv: sonarr = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not sonarr.configured(): raise HTTPException(status_code=400, detail="Sonarr not configured") episodes = await sonarr.get_episodes(int(arr_item["id"])) missing_by_season = _missing_episode_ids_by_season(episodes) season_numbers = sorted(missing_by_season) if not season_numbers: message = "Sonarr has no missing monitored episodes to search for." await asyncio.to_thread( save_action, request_id, "search_releases", "Search and choose a download", "ok", message, ) return {"status": "ok", "message": message, "collector": collector, "releases": []} searches = await asyncio.gather( *(sonarr.search_releases(int(arr_item["id"]), season) for season in season_numbers) ) season_results = [item for item in searches if isinstance(item, list)] longest_result = max((len(item) for item in season_results), default=0) for position in range(longest_result): for search_results in season_results: if position < len(search_results): results.append(search_results[position]) elif snapshot.request_type == RequestType.movie: radarr = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) if not radarr.configured(): raise HTTPException(status_code=400, detail="Radarr not configured") movie_results = await radarr.search_releases(int(arr_item["id"])) if isinstance(movie_results, list): results = movie_results else: raise HTTPException(status_code=400, detail="Unknown request type") except HTTPException: raise except httpx.HTTPStatusError as exc: _log_arr_http_error(collector, "interactive release search", exc) detail = _format_upstream_error(collector, exc) await asyncio.to_thread( save_action, request_id, "search_releases", "Search and choose a download", "failed", detail, ) raise HTTPException(status_code=502, detail=detail) from exc except Exception as exc: logger.exception("%s interactive release search failed request_id=%s", collector, request_id) detail = f"{collector} could not complete the release search: {exc}" await asyncio.to_thread( save_action, request_id, "search_releases", "Search and choose a download", "failed", detail, ) raise HTTPException(status_code=502, detail=detail) from exc releases = _filter_arr_release_results(results) await asyncio.to_thread( save_action, request_id, "search_releases", "Search and choose a download", "ok", f"{collector} found {len(releases)} releases.", ) return {"status": "ok", "collector": collector, "releases": releases} @router.post("/{request_id}/actions/search_auto") async def action_search_auto(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict: if not _user_can_use_search_auto(user): raise HTTPException(status_code=403, detail="Auto search and download is disabled for this user") runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = await build_snapshot(request_id) arr_item = snapshot.raw.get("arr", {}).get("item") if not isinstance(arr_item, dict): raise HTTPException(status_code=404, detail="Item not found in Sonarr/Radarr") if snapshot.request_type.value == "tv": client = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Sonarr not configured") target_profile_id = _quality_profile_id(runtime.sonarr_quality_profile_id) current_profile_id = _quality_profile_id(arr_item.get("qualityProfileId")) profile_message = None series_id = _quality_profile_id(arr_item.get("id")) if target_profile_id and series_id and current_profile_id != target_profile_id: series = await client.get_series(series_id) if not isinstance(series, dict): raise HTTPException(status_code=502, detail="Could not load Sonarr series before search") series["qualityProfileId"] = target_profile_id await client.update_series(series) profile_message = f"Sonarr quality profile updated to {target_profile_id} before search." episodes = await client.get_episodes(int(arr_item["id"])) missing_by_season = _missing_episode_ids_by_season(episodes) if not missing_by_season: message = "No missing monitored episodes found." if profile_message: message = f"{profile_message} {message}" await asyncio.to_thread( save_action, request_id, "search_auto", "Search and auto-download", "ok", message ) return {"status": "ok", "message": message, "searched": []} responses = [] for season_number in sorted(missing_by_season.keys()): episode_ids = missing_by_season[season_number] if episode_ids: response = await client.search_episodes(episode_ids) responses.append( {"season": season_number, "episodeCount": len(episode_ids), "response": response} ) message = "Search sent to Sonarr." if profile_message: message = f"{profile_message} {message}" await asyncio.to_thread( save_action, request_id, "search_auto", "Search and auto-download", "ok", message ) return {"status": "ok", "message": message, "searched": responses} if snapshot.request_type.value == "movie": client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Radarr not configured") target_profile_id = _quality_profile_id(runtime.radarr_quality_profile_id) current_profile_id = _quality_profile_id(arr_item.get("qualityProfileId")) profile_message = None movie_id = _quality_profile_id(arr_item.get("id")) if target_profile_id and movie_id and current_profile_id != target_profile_id: movie = await client.get_movie(movie_id) if not isinstance(movie, dict): raise HTTPException(status_code=502, detail="Could not load Radarr movie before search") movie["qualityProfileId"] = target_profile_id await client.update_movie(movie) profile_message = f"Radarr quality profile updated to {target_profile_id} before search." response = await client.search(int(arr_item["id"])) message = "Search sent to Radarr." if profile_message: message = f"{profile_message} {message}" await asyncio.to_thread( save_action, request_id, "search_auto", "Search and auto-download", "ok", message ) return {"status": "ok", "message": message, "response": response} raise HTTPException(status_code=400, detail="Unknown request type") @router.post("/{request_id}/actions/qbit/resume") async def action_resume(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = await build_snapshot(request_id) queue = snapshot.raw.get("arr", {}).get("queue") download_ids = _download_ids(_queue_records(queue)) if not download_ids: message = "Nothing to force resume." await asyncio.to_thread( save_action, request_id, "resume_torrent", "Resume torrent", "ok", message ) return {"status": "ok", "message": message} runtime = get_runtime_settings() client = QBittorrentClient( runtime.qbittorrent_base_url, runtime.qbittorrent_username, runtime.qbittorrent_password, ) if not client.configured(): raise HTTPException(status_code=400, detail="qBittorrent not configured") try: torrents = await client.get_torrents_by_hashes("|".join(download_ids)) torrent_list = torrents if isinstance(torrents, list) else [] downloading_states = {"downloading", "stalleddl", "queueddl", "checkingdl", "forceddl"} if torrent_list and all( str(t.get("state", "")).lower() in downloading_states for t in torrent_list ): message = "No need to force resume. Already downloading." await asyncio.to_thread( save_action, request_id, "resume_torrent", "Resume torrent", "ok", message ) return {"status": "ok", "message": message} await client.resume_torrents("|".join(download_ids)) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc message = "Resume sent to qBittorrent." await asyncio.to_thread( save_action, request_id, "resume_torrent", "Resume torrent", "ok", message ) return {"status": "ok", "resumed": download_ids, "message": message} @router.post("/{request_id}/actions/readd") async def action_readd(request_id: str, user: Dict[str, str] = Depends(get_current_user)) -> dict: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = await build_snapshot(request_id) jelly = snapshot.raw.get("jellyseerr") or {} media = jelly.get("media") or {} if snapshot.request_type.value == "tv": tvdb_id = media.get("tvdbId") if not tvdb_id: raise HTTPException(status_code=400, detail="Missing tvdbId for series") title = snapshot.title if title in {None, "", "Unknown"}: title = ( media.get("name") or media.get("title") or jelly.get("title") or jelly.get("name") ) if not runtime.sonarr_quality_profile_id or not runtime.sonarr_root_folder: raise HTTPException(status_code=400, detail="Sonarr profile/root not configured") client = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Sonarr not configured") try: existing = await client.get_series_by_tvdb_id(int(tvdb_id)) except httpx.HTTPStatusError as exc: detail = _format_upstream_error("Sonarr", exc) await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail ) raise HTTPException(status_code=502, detail=detail) from exc if isinstance(existing, list) and existing: series_id = existing[0].get("id") message = f"Already in Sonarr (seriesId {series_id})." await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "ok", message ) return {"status": "ok", "message": message, "seriesId": series_id} root_folder = await _resolve_root_folder_path(client, runtime.sonarr_root_folder, "Sonarr") try: response = await client.add_series( int(tvdb_id), runtime.sonarr_quality_profile_id, root_folder, title=title ) except httpx.HTTPStatusError as exc: detail = _format_upstream_error("Sonarr", exc) await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail ) raise HTTPException(status_code=502, detail=detail) from exc except ValueError as exc: detail = f"Sonarr could not add this series: {exc}" await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail ) raise HTTPException(status_code=502, detail=detail) from exc await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "ok", f"Re-added in Sonarr to {root_folder}.", ) return {"status": "ok", "response": response, "rootFolder": root_folder} if snapshot.request_type.value == "movie": tmdb_id = media.get("tmdbId") if not tmdb_id: raise HTTPException(status_code=400, detail="Missing tmdbId for movie") if not runtime.radarr_quality_profile_id or not runtime.radarr_root_folder: raise HTTPException(status_code=400, detail="Radarr profile/root not configured") client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) if not client.configured(): raise HTTPException(status_code=400, detail="Radarr not configured") try: existing = await client.get_movie_by_tmdb_id(int(tmdb_id)) except httpx.HTTPStatusError as exc: detail = _format_upstream_error("Radarr", exc) await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail ) raise HTTPException(status_code=502, detail=detail) from exc if isinstance(existing, list) and existing: movie_id = existing[0].get("id") message = f"Already in Radarr (movieId {movie_id})." await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "ok", message ) return {"status": "ok", "message": message, "movieId": movie_id} root_folder = await _resolve_root_folder_path(client, runtime.radarr_root_folder, "Radarr") title = snapshot.title if title in {None, "", "Unknown"}: title = ( media.get("title") or media.get("name") or jelly.get("title") or jelly.get("name") ) try: response = await client.add_movie( int(tmdb_id), runtime.radarr_quality_profile_id, root_folder, title=title ) except httpx.HTTPStatusError as exc: detail = _format_upstream_error("Radarr", exc) await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail ) raise HTTPException(status_code=502, detail=detail) from exc except ValueError as exc: detail = f"Radarr could not add this movie: {exc}" await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "failed", detail ) raise HTTPException(status_code=502, detail=detail) from exc await asyncio.to_thread( save_action, request_id, "readd_to_arr", "Re-add to Sonarr/Radarr", "ok", f"Re-added in Radarr to {root_folder}.", ) return {"status": "ok", "response": response, "rootFolder": root_folder} raise HTTPException(status_code=400, detail="Unknown request type") @router.get("/{request_id}/history") async def request_history( request_id: str, limit: int = 10, user: Dict[str, str] = Depends(get_current_user) ) -> dict: _require_advanced_request_access(user) runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshots = await asyncio.to_thread(get_recent_snapshots, request_id, limit) return {"snapshots": snapshots} @router.get("/{request_id}/actions") async def request_actions( request_id: str, limit: int = 10, user: Dict[str, str] = Depends(get_current_user) ) -> dict: _require_advanced_request_access(user) runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) actions = await asyncio.to_thread(get_recent_actions, request_id, limit) return {"actions": actions} @router.post("/{request_id}/actions/grab") async def action_grab( request_id: str, payload: Dict[str, Any], user: Dict[str, str] = Depends(get_current_user) ) -> dict: runtime = get_runtime_settings() client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key) if client.configured(): await _ensure_request_access(client, int(request_id), user) snapshot = await build_snapshot(request_id) guid = payload.get("guid") indexer_id = payload.get("indexerId") release_title = payload.get("title") if not guid or not indexer_id: raise HTTPException(status_code=400, detail="Missing guid or indexerId") try: arr_indexer_id = int(indexer_id) except (TypeError, ValueError) as exc: raise HTTPException(status_code=400, detail="indexerId must be an integer") from exc logger.info( "Collector grab requested: request_id=%s guid=%s indexer_id=%s has_title=%s", request_id, guid, indexer_id, bool(release_title), ) if snapshot.request_type.value == "tv": arr_client: Any = SonarrClient(runtime.sonarr_base_url, runtime.sonarr_api_key) service_label = "Sonarr" elif snapshot.request_type.value == "movie": arr_client = RadarrClient(runtime.radarr_base_url, runtime.radarr_api_key) service_label = "Radarr" else: raise HTTPException(status_code=400, detail="Unknown request type") if not arr_client.configured(): raise HTTPException(status_code=400, detail=f"{service_label} not configured") arr_error: Optional[str] = None try: response = await arr_client.grab_release(str(guid), arr_indexer_id) action_message = ( f"{release_title or 'Selected release'} was sent through {service_label} for download and import." ) await asyncio.to_thread( save_action, request_id, "grab", "Download selected release", "ok", action_message ) return { "status": "ok", "message": action_message, "response": {"collector": service_label, "queued": True}, } except httpx.HTTPStatusError as exc: _log_arr_http_error(service_label, "release grab", exc) status_code = exc.response.status_code if exc.response is not None else None arr_error = _format_upstream_error(service_label, exc) push_payload = _build_release_push_payload(payload) if status_code == 404 else None if push_payload is not None: logger.info( "%s release cache miss; retrying through release push request_id=%s", service_label, request_id, ) try: response = await arr_client.push_release(push_payload) accepted, rejection = _release_push_accepted(response) if accepted: action_message = ( f"{release_title or 'Selected release'} was sent through {service_label} for download and import." ) await asyncio.to_thread( save_action, request_id, "grab", "Download selected release", "ok", action_message, ) return { "status": "ok", "message": action_message, "response": {"collector": service_label, "queued": True}, } arr_error = rejection or f"{service_label} rejected the selected release" except httpx.HTTPStatusError as push_exc: _log_arr_http_error(service_label, "release push", push_exc) arr_error = _format_upstream_error(service_label, push_exc) except Exception as push_exc: logger.exception("%s release push failed request_id=%s", service_label, request_id) arr_error = str(push_exc) except Exception as exc: logger.exception("%s release grab failed request_id=%s", service_label, request_id) arr_error = str(exc) failure_message = ( f"The selected release could not be started through {service_label}. " + (arr_error or f"{service_label} did not accept the release.") ) await asyncio.to_thread( save_action, request_id, "grab", "Download selected release", "failed", failure_message ) raise HTTPException(status_code=502, detail=failure_message)