from __future__ import annotations import asyncio from datetime import datetime, timedelta, timezone from html import escape import json import logging from typing import Any, Dict, Optional from ..config import settings as env_settings from ..db import ( add_portal_item_activity, get_portal_item, get_user_by_username, list_portal_item_activity, list_portal_items, update_portal_item, ) from ..runtime import get_runtime_settings from .invite_email import resolve_user_delivery_email, send_generic_email from .snapshot import build_snapshot from .media_repair import evaluate_media_repair logger = logging.getLogger(__name__) _SYSTEM_USER = "Magent" _MEDIA_REPAIR_STARTED_EVENTS = {"replacement_started", "missing_search_started"} def _now() -> datetime: return datetime.now(timezone.utc) def _parse_datetime(value: Any) -> Optional[datetime]: if not isinstance(value, str) or not value.strip(): return None try: parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) except ValueError: return None return parsed.replace(tzinfo=parsed.tzinfo or timezone.utc).astimezone(timezone.utc) def _metadata(item: Dict[str, Any]) -> Dict[str, Any]: raw = item.get("metadata_json") if not isinstance(raw, str) or not raw.strip(): return {} try: parsed = json.loads(raw) except (TypeError, ValueError): return {} return parsed if isinstance(parsed, dict) else {} def issue_resolution_state(item: Dict[str, Any]) -> Dict[str, Any]: state = _metadata(item).get("resolutionConfirmation") return dict(state) if isinstance(state, dict) else {} def _metadata_with_resolution(item: Dict[str, Any], state: Dict[str, Any]) -> str: metadata = _metadata(item) metadata["resolutionConfirmation"] = state return json.dumps(metadata, separators=(",", ":"), sort_keys=True) def _interval_delta(value: int, unit: str) -> timedelta: safe_value = max(1, min(int(value), 365)) normalized_unit = str(unit or "days").strip().lower() if normalized_unit == "weeks": return timedelta(weeks=safe_value) if normalized_unit == "months": return timedelta(days=30 * safe_value) return timedelta(days=safe_value) def _workflow_settings() -> tuple[int, int, str]: runtime = get_runtime_settings() attempts = max(0, min(int(runtime.issue_confirmation_contact_attempts or 0), 10)) interval_value = max(1, min(int(runtime.issue_confirmation_interval_value or 1), 365)) interval_unit = str(runtime.issue_confirmation_interval_unit or "days").strip().lower() if interval_unit not in {"days", "weeks", "months"}: interval_unit = "days" return attempts, interval_value, interval_unit def _app_url() -> str: runtime = get_runtime_settings() for value in (runtime.magent_application_url, runtime.magent_proxy_base_url, env_settings.cors_allow_origin): candidate = str(value or "").strip() if candidate: return candidate.rstrip("/") return f"http://localhost:{int(runtime.magent_application_port or 3000)}" def _issue_url(item_id: int) -> str: return f"{_app_url()}/portal/issues?item={item_id}" def _activity( item_id: int, event_type: str, message: str, *, actor_username: str = _SYSTEM_USER, actor_role: str = "system", metadata: Optional[Dict[str, Any]] = None, ) -> None: add_portal_item_activity( item_id, event_type=event_type, actor_username=actor_username, actor_role=actor_role, message=message, metadata_json=json.dumps(metadata, separators=(",", ":"), sort_keys=True) if metadata else None, ) def _activity_metadata(entry: Dict[str, Any]) -> Dict[str, Any]: raw = entry.get("metadata_json") if not isinstance(raw, str) or not raw.strip(): return {} try: parsed = json.loads(raw) except (TypeError, ValueError): return {} return parsed if isinstance(parsed, dict) else {} def _repair_tracking(item_id: int) -> tuple[Dict[str, Any], list[Dict[str, Any]]]: activity = list_portal_item_activity(item_id, limit=500) for entry in reversed(activity): # A rejected repair must not be proposed again simply because the same # replacement file is still present. Wait for a NEW repair attempt. if str(entry.get("event_type") or "") == "resolution_rejected": return {}, activity if str(entry.get("event_type") or "") not in _MEDIA_REPAIR_STARTED_EVENTS: continue tracking = _activity_metadata(entry).get("repairTracking") if isinstance(tracking, dict): return dict(tracking), activity return {}, activity async def _media_repair_evidence(tracking: Dict[str, Any]) -> Dict[str, Any]: snapshot = await build_snapshot(str(tracking.get("requestId") or "")) raw = snapshot.raw if isinstance(snapshot.raw, dict) else {} jellyfin = dict(raw.get("jellyfin") or {}) jellyfin["found"] = jellyfin.get("catalogFound", jellyfin.get("found")) return await evaluate_media_repair( tracking, (raw.get("arr") or {}).get("item"), jellyfin, episodes=(raw.get("arr") or {}).get("episodes"), ) def _close_issue( item: Dict[str, Any], *, reason: str, confirmed: bool, actor_username: str = _SYSTEM_USER, actor_role: str = "system", ) -> Dict[str, Any]: now = _now().isoformat() state = issue_resolution_state(item) state.update( { "status": "confirmed" if confirmed else "auto_closed", "confirmedAt": now if confirmed else state.get("confirmedAt"), "closedAt": now, "nextContactAt": None, "closedReason": reason, } ) updated = update_portal_item( int(item["id"]), status="closed", issue_resolved_at=now, metadata_json=_metadata_with_resolution(item, state), ) if not updated: raise RuntimeError("Issue could not be closed") _activity( int(item["id"]), "resolution_confirmed" if confirmed else "issue_auto_closed", reason, actor_username=actor_username, actor_role=actor_role, ) return updated async def _contact_reporter(item: Dict[str, Any]) -> Dict[str, Any]: maximum, interval_value, interval_unit = _workflow_settings() state = issue_resolution_state(item) attempts = max(0, int(state.get("attemptsSent") or 0)) if maximum <= 0: return _close_issue( item, reason="Issue closed automatically because reporter confirmation emails are disabled.", confirmed=False, ) if attempts >= maximum: return _close_issue( item, reason=f"Issue closed automatically after {attempts} confirmation email attempt(s) without a response.", confirmed=False, ) attempt_number = attempts + 1 reporter = get_user_by_username(str(item.get("created_by_username") or "")) recipient = resolve_user_delivery_email(reporter) issue_url = f"{_app_url()}/issues/confirm/{int(item['id'])}" sent = False delivery_error: Optional[str] = None if recipient: subject = f"Ready to try again? Grizzlyflix issue #{item['id']}" body_text = ( "Your repair looks ready to test.\n\n" f"{item.get('title') or 'Your reported issue'}\n\n" "Please try the affected content in Grizzlyflix. Is it fixed?\n\n" f"YES — it works: {issue_url}#yes\n" f"NO — still broken: {issue_url}#no\n\n" "Confirm your answer in Magent. You may need to sign in first.\n" "Yes closes the report. No keeps it open for another look.\n\n" f"Reminder {attempt_number} of {maximum}. If we do not hear back after the reminder period, this report will close automatically." ) body_html = ( '
' '
' '

GRIZZLYFLIX · MAGENT

' '

Ready to try again?

' '

Your repair looks ready to test. Give the affected content a try, then let us know:

' f'

{escape(str(item.get("title") or "Your reported issue"))}

' '

Is it fixed?

' f'YES — it works' f'NO — still broken' '

Confirm your answer in Magent. You may need to sign in first.
Yes closes the report. No keeps it open for another look.

' f'

Reminder {attempt_number} of {maximum} · Issue #{int(item["id"])}
If we do not hear back after the reminder period, this report will close automatically.

' '
' ) try: await send_generic_email( recipient_email=recipient, subject=subject, body_text=body_text, body_html=body_html, ) sent = True except Exception as exc: delivery_error = str(exc) logger.exception("issue confirmation email failed item_id=%s attempt=%s", item.get("id"), attempt_number) else: delivery_error = "No email address is stored for the reporter." now = _now() state.update( { "status": "awaiting_confirmation", "attemptsSent": attempt_number, "maximumAttempts": maximum, "lastContactAt": now.isoformat(), "nextContactAt": (now + _interval_delta(interval_value, interval_unit)).isoformat(), "intervalValue": interval_value, "intervalUnit": interval_unit, "lastDeliverySucceeded": sent, "lastDeliveryError": delivery_error, } ) updated = update_portal_item( int(item["id"]), metadata_json=_metadata_with_resolution(item, state), ) if not updated: raise RuntimeError("Issue confirmation schedule could not be saved") if sent: message = f"Confirmation email {attempt_number} of {maximum} was sent to the reporter." else: message = f"Confirmation email {attempt_number} of {maximum} could not be delivered." _activity( int(item["id"]), "confirmation_email_sent" if sent else "confirmation_email_failed", message, metadata={ "attempt": attempt_number, "maximum": maximum, "nextContactAt": state["nextContactAt"], "deliveryError": delivery_error, }, ) return updated async def begin_issue_confirmation( item_id: int, *, actor_username: str, actor_role: str, ) -> Dict[str, Any]: item = get_portal_item(item_id) if not item or str(item.get("kind") or "").lower() != "issue": raise ValueError("Issue not found") now = _now().isoformat() maximum, interval_value, interval_unit = _workflow_settings() state = { "status": "awaiting_confirmation", "startedAt": now, "attemptsSent": 0, "maximumAttempts": maximum, "lastContactAt": None, "nextContactAt": now, "intervalValue": interval_value, "intervalUnit": interval_unit, "confirmedAt": None, "closedAt": None, } updated = update_portal_item( item_id, status="awaiting_confirmation", issue_resolved_at=None, metadata_json=_metadata_with_resolution(item, state), ) if not updated: raise RuntimeError("Issue confirmation workflow could not be started") _activity( item_id, "resolution_proposed", "The issue was marked fixed and sent to the reporter for confirmation.", actor_username=actor_username, actor_role=actor_role, metadata={"maximumAttempts": maximum, "intervalValue": interval_value, "intervalUnit": interval_unit}, ) return await _contact_reporter(updated) def respond_to_issue_confirmation( item_id: int, *, resolved: bool, actor_username: str, actor_role: str, ) -> Dict[str, Any]: item = get_portal_item(item_id) if not item or str(item.get("kind") or "").lower() != "issue": raise ValueError("Issue not found") if str(item.get("status") or "").lower() != "awaiting_confirmation": raise ValueError("This issue is not waiting for resolution confirmation") if resolved: return _close_issue( item, reason="The reporter confirmed that the issue is fixed.", confirmed=True, actor_username=actor_username, actor_role=actor_role, ) now = _now().isoformat() state = issue_resolution_state(item) state.update( { "status": "reported_still_broken", "reporterResponseAt": now, "nextContactAt": None, "closedAt": None, } ) updated = update_portal_item( item_id, status="in_progress", issue_resolved_at=None, metadata_json=_metadata_with_resolution(item, state), ) if not updated: raise RuntimeError("Issue could not be reopened") _activity( item_id, "resolution_rejected", "The reporter said the issue is still happening. The issue was returned to In progress.", actor_username=actor_username, actor_role=actor_role, ) return updated async def process_active_media_repairs() -> Dict[str, int]: items = list_portal_items(kind="issue", status="in_progress", limit=500) result = {"checked": 0, "waiting": 0, "completed": 0, "failed": 0} for item in items: tracking, activity = _repair_tracking(int(item["id"])) if not tracking: continue result["checked"] += 1 try: evidence = await _media_repair_evidence(tracking) if evidence.get("complete"): _activity( int(item["id"]), "repair_verified", str(evidence.get("message") or "Magent verified the repaired media in Jellyfin."), metadata={ "requestId": tracking.get("requestId"), "actionId": tracking.get("actionId"), }, ) await begin_issue_confirmation( int(item["id"]), actor_username=_SYSTEM_USER, actor_role="system", ) result["completed"] += 1 continue result["waiting"] += 1 if evidence.get("phase") == "indexing" and not any( str(entry.get("event_type") or "") == "repair_imported" for entry in activity ): _activity( int(item["id"]), "repair_imported", str(evidence.get("message") or "The repaired file was imported and is waiting for Jellyfin."), metadata={ "requestId": tracking.get("requestId"), "actionId": tracking.get("actionId"), }, ) except Exception: result["failed"] += 1 logger.exception("automatic media repair check failed item_id=%s", item.get("id")) return result async def process_due_issue_confirmations(now: Optional[datetime] = None) -> Dict[str, int]: current = (now or _now()).astimezone(timezone.utc) items = list_portal_items(kind="issue", status="awaiting_confirmation", limit=500) result = {"checked": len(items), "contacted": 0, "closed": 0, "failed": 0} maximum, _, _ = _workflow_settings() for item in items: state = issue_resolution_state(item) due_at = _parse_datetime(state.get("nextContactAt")) if due_at and due_at > current: continue try: attempts = max(0, int(state.get("attemptsSent") or 0)) if maximum <= 0 or attempts >= maximum: _close_issue( item, reason=( "Issue closed automatically because reporter confirmation emails are disabled." if maximum <= 0 else f"Issue closed automatically after {attempts} confirmation email attempt(s) without a response." ), confirmed=False, ) result["closed"] += 1 else: await _contact_reporter(item) result["contacted"] += 1 except Exception: result["failed"] += 1 logger.exception("issue confirmation processing failed item_id=%s", item.get("id")) return result async def run_issue_confirmation_loop() -> None: while True: try: repair_result = await process_active_media_repairs() if repair_result["completed"] or repair_result["failed"]: logger.info("automatic media repair sweep complete result=%s", repair_result) result = await process_due_issue_confirmations() if result["contacted"] or result["closed"] or result["failed"]: logger.info("issue confirmation sweep complete result=%s", result) except asyncio.CancelledError: raise except Exception: logger.exception("issue confirmation sweep failed") await asyncio.sleep(60)