from __future__ import annotations import logging import time from datetime import datetime, timezone from typing import Any, Dict, Optional, Tuple import httpx from fastapi import APIRouter, Depends, HTTPException, Query from ..auth import get_current_user from ..clients.jellyfin import JellyfinClient from ..db import ( add_portal_comment, count_portal_items, create_portal_item, get_portal_item, get_portal_overview, list_portal_comments, list_portal_items, update_portal_item, ) from ..services.notifications import send_portal_notification from ..runtime import get_runtime_settings router = APIRouter(prefix="/portal", tags=["portal"], dependencies=[Depends(get_current_user)]) logger = logging.getLogger(__name__) PORTAL_KINDS = {"request", "issue", "feature"} PORTAL_STATUSES = { # Existing generic statuses "new", "triaging", "planned", "in_progress", "blocked", "done", "declined", "closed", # Seerr-style request pipeline statuses "pending", "approved", "processing", "partially_available", "available", "failed", } PORTAL_PRIORITIES = {"low", "normal", "high", "urgent"} PORTAL_MEDIA_TYPES = {"movie", "tv"} PORTAL_REQUEST_STATUSES = {"pending", "approved", "declined"} PORTAL_MEDIA_STATUSES = { "unknown", "pending", "processing", "partially_available", "available", "failed", } PORTAL_ISSUE_TYPES = { "general", "playback", "transcode", "service_unavailable", "broken_media", "audio", "subtitle", "quality", "metadata", "missing_content", "other", } _MEDIA_STATUS_CACHE: Dict[str, Any] = {"expires_at": 0.0, "payload": None} _MEDIA_STATUS_CACHE_SECONDS = 15.0 REQUEST_STATUS_TRANSITIONS: Dict[str, set[str]] = { "pending": {"pending", "approved", "declined"}, "approved": {"approved", "declined"}, "declined": {"declined", "pending", "approved"}, } MEDIA_STATUS_TRANSITIONS: Dict[str, set[str]] = { "unknown": {"unknown", "pending", "processing", "failed"}, "pending": {"pending", "processing", "partially_available", "available", "failed"}, "processing": {"processing", "partially_available", "available", "failed"}, "partially_available": {"partially_available", "processing", "available", "failed"}, "available": {"available", "processing"}, "failed": {"failed", "processing", "available"}, } LEGACY_STATUS_TO_WORKFLOW: Dict[str, Tuple[str, str]] = { "new": ("pending", "pending"), "triaging": ("pending", "pending"), "planned": ("approved", "pending"), "in_progress": ("approved", "processing"), "blocked": ("approved", "failed"), "done": ("approved", "available"), "closed": ("approved", "available"), "pending": ("pending", "pending"), "approved": ("approved", "pending"), "declined": ("declined", "unknown"), "processing": ("approved", "processing"), "partially_available": ("approved", "partially_available"), "available": ("approved", "available"), "failed": ("approved", "failed"), } def _clean_text(value: Any) -> Optional[str]: if value is None: return None if isinstance(value, str): trimmed = value.strip() return trimmed if trimmed else None return str(value) def _require_text(value: Any, field: str, *, max_length: int = 5000) -> str: normalized = _clean_text(value) if not normalized: raise HTTPException(status_code=400, detail=f"{field} is required") if len(normalized) > max_length: raise HTTPException( status_code=400, detail=f"{field} is too long (max {max_length} characters)", ) return normalized def _normalize_choice( value: Any, *, field: str, allowed: set[str], default: Optional[str] = None, allow_empty: bool = False, ) -> Optional[str]: if value is None: return default normalized = _clean_text(value) if not normalized: return None if allow_empty else default candidate = normalized.lower() if candidate not in allowed: allowed_values = ", ".join(sorted(allowed)) raise HTTPException(status_code=400, detail=f"Invalid {field}. Allowed: {allowed_values}") return candidate def _normalize_year(value: Any, *, allow_empty: bool = True) -> Optional[int]: if value is None: return None if isinstance(value, str): stripped = value.strip() if not stripped: return None if allow_empty else 0 value = stripped try: year = int(value) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="year must be an integer") from None if year < 1800 or year > 2100: raise HTTPException(status_code=400, detail="year must be between 1800 and 2100") return year def _normalize_int(value: Any, field: str, *, allow_empty: bool = True) -> Optional[int]: if value is None: return None if isinstance(value, str): stripped = value.strip() if not stripped: return None if allow_empty else 0 value = stripped try: return int(value) except (TypeError, ValueError): raise HTTPException(status_code=400, detail=f"{field} must be an integer") from None def _normalize_bool(value: Any, *, default: bool = False) -> bool: if value is None: return default if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): candidate = value.strip().lower() if candidate in {"1", "true", "yes", "on"}: return True if candidate in {"0", "false", "no", "off"}: return False raise HTTPException(status_code=400, detail="Boolean value expected") def _workflow_to_item_status(request_status: str, media_status: str) -> str: if request_status == "declined": return "declined" if request_status == "pending": return "pending" if media_status == "available": return "available" if media_status == "partially_available": return "partially_available" if media_status == "failed": return "failed" if media_status == "processing": return "processing" return "approved" def _item_status_to_workflow(item: Dict[str, Any]) -> Tuple[str, str]: request_status = _normalize_choice( item.get("workflow_request_status"), field="request_status", allowed=PORTAL_REQUEST_STATUSES, allow_empty=True, ) media_status = _normalize_choice( item.get("workflow_media_status"), field="media_status", allowed=PORTAL_MEDIA_STATUSES, allow_empty=True, ) if request_status and media_status: return request_status, media_status status = _clean_text(item.get("status")) if status: mapped = LEGACY_STATUS_TO_WORKFLOW.get(status.lower()) if mapped: return mapped return "pending", "pending" def _stage_label_for_workflow(request_status: str, media_status: str) -> str: if request_status == "declined": return "Declined" if request_status == "pending": return "Waiting for approval" if media_status == "available": return "Ready to watch" if media_status == "partially_available": return "Partially available" if media_status == "processing": return "Working on it" if media_status == "failed": return "Needs attention" return "Approved" def _normalize_request_pipeline( request_status: Optional[str], media_status: Optional[str], *, fallback_request_status: str = "pending", fallback_media_status: str = "pending", ) -> Tuple[str, str]: normalized_request = _normalize_choice( request_status, field="request_status", allowed=PORTAL_REQUEST_STATUSES, default=fallback_request_status, ) normalized_media = _normalize_choice( media_status, field="media_status", allowed=PORTAL_MEDIA_STATUSES, default=fallback_media_status, ) request_value = normalized_request or fallback_request_status media_value = normalized_media or fallback_media_status if request_value == "declined": return request_value, "unknown" if request_value == "pending": if media_value not in {"pending", "unknown"}: media_value = "pending" return request_value, media_value if media_value == "unknown": media_value = "pending" return request_value, media_value def _validate_pipeline_transition( current_request: str, current_media: str, requested_request: str, requested_media: str, ) -> Tuple[str, str]: allowed_request = REQUEST_STATUS_TRANSITIONS.get(current_request, {current_request}) if requested_request not in allowed_request: allowed_text = ", ".join(sorted(allowed_request)) raise HTTPException( status_code=400, detail=( f"Invalid request_status transition: {current_request} -> {requested_request}. " f"Allowed: {allowed_text}" ), ) normalized_request, normalized_media = _normalize_request_pipeline( requested_request, requested_media, fallback_request_status=current_request, fallback_media_status=current_media, ) if normalized_request != "approved": return normalized_request, normalized_media if current_request != "approved": allowed_media = PORTAL_MEDIA_STATUSES - {"unknown"} else: allowed_media = MEDIA_STATUS_TRANSITIONS.get(current_media, {current_media}) if normalized_media not in allowed_media: allowed_text = ", ".join(sorted(allowed_media)) raise HTTPException( status_code=400, detail=( f"Invalid media_status transition: {current_media} -> {normalized_media}. " f"Allowed: {allowed_text}" ), ) return normalized_request, normalized_media def _ensure_item_exists(item_id: Optional[int], *, field: str = "related_item_id") -> None: if item_id is None: return target = get_portal_item(item_id) if not target: raise HTTPException(status_code=400, detail=f"{field} references an unknown portal item") def _sanitize_metadata_json(value: Any) -> Optional[str]: text = _clean_text(value) if text is None: return None if len(text) > 50000: raise HTTPException(status_code=400, detail="metadata_json is too long (max 50000 characters)") return text def _is_admin(user: Dict[str, Any]) -> bool: return str(user.get("role") or "").strip().lower() == "admin" def _is_owner(user: Dict[str, Any], item: Dict[str, Any]) -> bool: return str(user.get("username") or "") == str(item.get("created_by_username") or "") def _public_media_status_payload( *, status: str, headline: str, message: str, latency_ms: Optional[int] = None, version: Optional[str] = None, restart_pending: Optional[bool] = None, active_streams: Optional[int] = None, transcoding_streams: Optional[int] = None, session_check_available: bool = False, ) -> Dict[str, Any]: return { "checked_at": datetime.now(timezone.utc).isoformat(), "status": status, "headline": headline, "message": message, "latency_ms": latency_ms, "server": { "version": version, "restart_pending": restart_pending, }, "activity": { "active_streams": active_streams, "transcoding_streams": transcoding_streams, "available": session_check_available, }, } def _serialize_item(item: Dict[str, Any], user: Dict[str, Any]) -> Dict[str, Any]: is_admin = _is_admin(user) is_owner = _is_owner(user, item) serialized = dict(item) serialized["permissions"] = { "can_edit": is_admin or is_owner, "can_comment": True, "can_moderate": is_admin, "can_raise_issue": str(item.get("kind") or "") == "request", } kind = str(item.get("kind") or "").strip().lower() if kind == "request": request_status, media_status = _item_status_to_workflow(item) serialized["workflow"] = { "request_status": request_status, "media_status": media_status, "stage_label": _stage_label_for_workflow(request_status, media_status), "is_terminal": media_status in {"available", "failed"} or request_status == "declined", } elif kind == "issue": serialized["issue"] = { "issue_type": _clean_text(item.get("issue_type")) or "general", "related_item_id": _normalize_int(item.get("related_item_id"), "related_item_id"), "is_resolved": bool(_clean_text(item.get("issue_resolved_at"))), "resolved_at": _clean_text(item.get("issue_resolved_at")), } return serialized async def _notify( *, event_type: str, item: Dict[str, Any], user: Dict[str, Any], note: Optional[str] = None, ) -> None: try: result = await send_portal_notification( event_type=event_type, item=item, actor_username=str(user.get("username") or "unknown"), actor_role=str(user.get("role") or "user"), note=note, ) logger.info( "portal notification dispatched event=%s item_id=%s status=%s", event_type, item.get("id"), result.get("status"), ) except Exception: logger.exception( "portal notification failed event=%s item_id=%s", event_type, item.get("id"), ) @router.get("/overview") async def portal_overview(current_user: Dict[str, Any] = Depends(get_current_user)) -> Dict[str, Any]: mine = count_portal_items(mine_username=str(current_user.get("username") or "")) return { "overview": get_portal_overview(), "my_items": mine, } @router.get("/issues/media-status") async def portal_media_status() -> Dict[str, Any]: """Return a short, privacy-safe Jellyfin health check for guided issue reporting.""" now = time.monotonic() cached_payload = _MEDIA_STATUS_CACHE.get("payload") if isinstance(cached_payload, dict) and now < float(_MEDIA_STATUS_CACHE.get("expires_at") or 0): return cached_payload runtime = get_runtime_settings() jellyfin = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key) if not jellyfin.configured(): payload = _public_media_status_payload( status="not_configured", headline="Media server status is unavailable", message="Magent cannot run a playback check right now. Your report can still be submitted.", ) _MEDIA_STATUS_CACHE.update( expires_at=now + _MEDIA_STATUS_CACHE_SECONDS, payload=payload, ) return payload started_at = time.perf_counter() try: system_info = await jellyfin.get_system_info() except (httpx.HTTPError, RuntimeError, ValueError): latency_ms = round((time.perf_counter() - started_at) * 1000) payload = _public_media_status_payload( status="down", headline="The media server is not responding", message=( "This looks broader than one title. The report will include the failed server check " "so an administrator can investigate the service first." ), latency_ms=latency_ms, ) _MEDIA_STATUS_CACHE.update( expires_at=now + _MEDIA_STATUS_CACHE_SECONDS, payload=payload, ) return payload except Exception: logger.exception("guided issue Jellyfin system check failed") latency_ms = round((time.perf_counter() - started_at) * 1000) payload = _public_media_status_payload( status="down", headline="The media server check failed", message="Your report can still be submitted and will include this failed service check.", latency_ms=latency_ms, ) _MEDIA_STATUS_CACHE.update( expires_at=now + _MEDIA_STATUS_CACHE_SECONDS, payload=payload, ) return payload latency_ms = round((time.perf_counter() - started_at) * 1000) info = system_info if isinstance(system_info, dict) else {} version_value = info.get("Version") version = str(version_value).strip() if version_value is not None else None restart_pending = bool(info.get("HasPendingRestart")) session_check_available = False active_streams: Optional[int] = None transcoding_streams: Optional[int] = None try: sessions = await jellyfin.get_sessions() if isinstance(sessions, list): session_check_available = True active_streams = sum( 1 for session in sessions if isinstance(session, dict) and session.get("NowPlayingItem") ) transcoding_streams = sum( 1 for session in sessions if isinstance(session, dict) and session.get("NowPlayingItem") and session.get("TranscodingInfo") ) except Exception: logger.warning("guided issue Jellyfin session check unavailable", exc_info=True) if restart_pending: status = "degraded" headline = "Media server is online but needs attention" message = "Jellyfin is responding, but it reports that a restart is pending." elif session_check_available and active_streams: status = "up" headline = "Media server is online and actively streaming" message = ( "Other playback is currently working, so this is more likely specific to the title, " "audio track, subtitle, client, or transcode path." ) else: status = "up" headline = "Media server is online" message = "Jellyfin responded normally. Continue with the report if playback is still failing." payload = _public_media_status_payload( status=status, headline=headline, message=message, latency_ms=latency_ms, version=version, restart_pending=restart_pending, active_streams=active_streams, transcoding_streams=transcoding_streams, session_check_available=session_check_available, ) _MEDIA_STATUS_CACHE.update( expires_at=now + _MEDIA_STATUS_CACHE_SECONDS, payload=payload, ) return payload @router.get("/items") async def portal_list_items( kind: Optional[str] = None, status: Optional[str] = None, request_status: Optional[str] = None, media_status: Optional[str] = None, source_system: Optional[str] = None, source_request_id: Optional[int] = None, related_item_id: Optional[int] = None, mine: bool = False, search: Optional[str] = None, limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0), current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: kind_value = _normalize_choice( kind, field="kind", allowed=PORTAL_KINDS, allow_empty=True ) status_value = _normalize_choice( status, field="status", allowed=PORTAL_STATUSES, allow_empty=True ) request_status_value = _normalize_choice( request_status, field="request_status", allowed=PORTAL_REQUEST_STATUSES, allow_empty=True ) media_status_value = _normalize_choice( media_status, field="media_status", allowed=PORTAL_MEDIA_STATUSES, allow_empty=True ) source_system_value = _clean_text(source_system) if source_system_value: source_system_value = source_system_value.lower() mine_username = str(current_user.get("username") or "") if mine else None items = list_portal_items( kind=kind_value, status=status_value, workflow_request_status=request_status_value, workflow_media_status=media_status_value, source_system=source_system_value, source_request_id=source_request_id, related_item_id=related_item_id, mine_username=mine_username, search=_clean_text(search), limit=limit, offset=offset, ) total = count_portal_items( kind=kind_value, status=status_value, workflow_request_status=request_status_value, workflow_media_status=media_status_value, source_system=source_system_value, source_request_id=source_request_id, related_item_id=related_item_id, mine_username=mine_username, search=_clean_text(search), ) return { "items": [_serialize_item(item, current_user) for item in items], "total": total, "limit": limit, "offset": offset, "has_more": offset + len(items) < total, "filters": { "kind": kind_value, "status": status_value, "request_status": request_status_value, "media_status": media_status_value, "source_system": source_system_value, "source_request_id": source_request_id, "related_item_id": related_item_id, "mine": mine, "search": _clean_text(search), }, } @router.get("/requests") async def portal_list_requests( request_status: Optional[str] = None, media_status: Optional[str] = None, mine: bool = False, search: Optional[str] = None, limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0), current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: mine_username = str(current_user.get("username") or "") if mine else None request_status_value = _normalize_choice( request_status, field="request_status", allowed=PORTAL_REQUEST_STATUSES, allow_empty=True ) media_status_value = _normalize_choice( media_status, field="media_status", allowed=PORTAL_MEDIA_STATUSES, allow_empty=True ) items = list_portal_items( kind="request", workflow_request_status=request_status_value, workflow_media_status=media_status_value, mine_username=mine_username, search=_clean_text(search), limit=limit, offset=offset, ) total = count_portal_items( kind="request", workflow_request_status=request_status_value, workflow_media_status=media_status_value, mine_username=mine_username, search=_clean_text(search), ) return { "items": [_serialize_item(item, current_user) for item in items], "total": total, "limit": limit, "offset": offset, "has_more": offset + len(items) < total, "filters": { "request_status": request_status_value, "media_status": media_status_value, "mine": mine, "search": _clean_text(search), }, } @router.post("/items") async def portal_create_item( payload: Dict[str, Any], current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: is_admin = _is_admin(current_user) kind = _normalize_choice( payload.get("kind"), field="kind", allowed=PORTAL_KINDS, default="request", ) title = _require_text(payload.get("title"), "title", max_length=220) description = _require_text(payload.get("description"), "description", max_length=10000) media_type = _normalize_choice( payload.get("media_type"), field="media_type", allowed=PORTAL_MEDIA_TYPES, allow_empty=True, ) year = _normalize_year(payload.get("year")) external_ref = _clean_text(payload.get("external_ref")) source_system = _clean_text(payload.get("source_system")) if is_admin else None if source_system: source_system = source_system.lower() source_request_id = ( _normalize_int(payload.get("source_request_id"), "source_request_id") if is_admin else None ) related_item_id = _normalize_int(payload.get("related_item_id"), "related_item_id") _ensure_item_exists(related_item_id) workflow_request_status: Optional[str] = None workflow_media_status: Optional[str] = None issue_type: Optional[str] = None issue_resolved_at: Optional[str] = None status: Optional[str] = None if kind == "request": workflow_request_status, workflow_media_status = _normalize_request_pipeline( payload.get("request_status"), payload.get("media_status"), fallback_request_status="pending", fallback_media_status="pending", ) status = _workflow_to_item_status(workflow_request_status, workflow_media_status) else: status = _normalize_choice( payload.get("status") if is_admin else None, field="status", allowed=PORTAL_STATUSES, default="new", ) if kind == "issue": issue_type = _normalize_choice( payload.get("issue_type"), field="issue_type", allowed=PORTAL_ISSUE_TYPES, default="general", ) if related_item_id is not None and not source_system: source_system = "portal_request" source_request_id = related_item_id priority = _normalize_choice( payload.get("priority"), field="priority", allowed=PORTAL_PRIORITIES, default="normal", ) assignee_username = _clean_text(payload.get("assignee_username")) if is_admin else None metadata_json = _sanitize_metadata_json(payload.get("metadata_json")) if is_admin else None created = create_portal_item( kind=kind or "request", title=title, description=description, created_by_username=str(current_user.get("username") or "unknown"), created_by_id=_normalize_int(current_user.get("jellyseerr_user_id"), "jellyseerr_user_id"), media_type=media_type, year=year, external_ref=external_ref, source_system=source_system, source_request_id=source_request_id, related_item_id=related_item_id, status=status or "new", workflow_request_status=workflow_request_status, workflow_media_status=workflow_media_status, issue_type=issue_type, issue_resolved_at=issue_resolved_at, metadata_json=metadata_json, priority=priority or "normal", assignee_username=assignee_username, ) initial_comment = _clean_text(payload.get("comment")) if initial_comment: add_portal_comment( int(created["id"]), author_username=str(current_user.get("username") or "unknown"), author_role=str(current_user.get("role") or "user"), message=initial_comment, is_internal=False, ) comments = list_portal_comments(int(created["id"]), include_internal=is_admin) await _notify( event_type="portal_item_created", item=created, user=current_user, note=f"kind={created.get('kind')} priority={created.get('priority')}", ) return { "item": _serialize_item(created, current_user), "comments": comments, } @router.post("/requests/{item_id}/issues") async def portal_create_issue_for_request( item_id: int, payload: Dict[str, Any], current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: request_item = get_portal_item(item_id) if not request_item: raise HTTPException(status_code=404, detail="Portal request not found") if str(request_item.get("kind") or "").lower() != "request": raise HTTPException(status_code=400, detail="Only request items can have linked issues") title = _require_text(payload.get("title"), "title", max_length=220) description = _require_text(payload.get("description"), "description", max_length=10000) issue_type = _normalize_choice( payload.get("issue_type"), field="issue_type", allowed=PORTAL_ISSUE_TYPES, default="general", ) status = _normalize_choice( payload.get("status"), field="status", allowed=PORTAL_STATUSES, default="new", ) priority = _normalize_choice( payload.get("priority"), field="priority", allowed=PORTAL_PRIORITIES, default="normal", ) created = create_portal_item( kind="issue", title=title, description=description, created_by_username=str(current_user.get("username") or "unknown"), created_by_id=_normalize_int(current_user.get("jellyseerr_user_id"), "jellyseerr_user_id"), media_type=request_item.get("media_type"), year=request_item.get("year"), external_ref=_clean_text(payload.get("external_ref")), source_system="portal_request", source_request_id=item_id, related_item_id=item_id, status=status or "new", issue_type=issue_type, priority=priority or "normal", assignee_username=_clean_text(payload.get("assignee_username")) if _is_admin(current_user) else None, ) initial_comment = _clean_text(payload.get("comment")) if initial_comment: add_portal_comment( int(created["id"]), author_username=str(current_user.get("username") or "unknown"), author_role=str(current_user.get("role") or "user"), message=initial_comment, is_internal=False, ) comments = list_portal_comments(int(created["id"]), include_internal=_is_admin(current_user)) await _notify( event_type="portal_issue_created", item=created, user=current_user, note=f"linked_request_id={item_id}", ) return { "item": _serialize_item(created, current_user), "comments": comments, "linked_request_id": item_id, } @router.get("/requests/{item_id}/issues") async def portal_list_request_issues( item_id: int, limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0), current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: request_item = get_portal_item(item_id) if not request_item: raise HTTPException(status_code=404, detail="Portal request not found") if str(request_item.get("kind") or "").lower() != "request": raise HTTPException(status_code=400, detail="Only request items can have linked issues") items = list_portal_items( kind="issue", related_item_id=item_id, limit=limit, offset=offset, ) total = count_portal_items(kind="issue", related_item_id=item_id) return { "items": [_serialize_item(item, current_user) for item in items], "total": total, "limit": limit, "offset": offset, "has_more": offset + len(items) < total, "linked_request_id": item_id, } @router.patch("/requests/{item_id}/pipeline") async def portal_update_request_pipeline( item_id: int, payload: Dict[str, Any], current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: if not _is_admin(current_user): raise HTTPException(status_code=403, detail="Admin access required") item = get_portal_item(item_id) if not item: raise HTTPException(status_code=404, detail="Portal request not found") if str(item.get("kind") or "").lower() != "request": raise HTTPException(status_code=400, detail="Only request items support pipeline updates") current_request_status, current_media_status = _item_status_to_workflow(item) requested_request = _normalize_choice( payload.get("request_status"), field="request_status", allowed=PORTAL_REQUEST_STATUSES, default=current_request_status, ) or current_request_status requested_media = _normalize_choice( payload.get("media_status"), field="media_status", allowed=PORTAL_MEDIA_STATUSES, default=current_media_status, ) or current_media_status next_request_status, next_media_status = _validate_pipeline_transition( current_request_status, current_media_status, requested_request, requested_media, ) next_status = _workflow_to_item_status(next_request_status, next_media_status) updated = update_portal_item( item_id, status=next_status, workflow_request_status=next_request_status, workflow_media_status=next_media_status, ) if not updated: raise HTTPException(status_code=404, detail="Portal request not found") comment_text = _clean_text(payload.get("comment")) if comment_text: add_portal_comment( item_id, author_username=str(current_user.get("username") or "unknown"), author_role=str(current_user.get("role") or "admin"), message=comment_text, is_internal=_normalize_bool(payload.get("is_internal"), default=False), ) await _notify( event_type="portal_request_pipeline_updated", item=updated, user=current_user, note=f"{current_request_status}/{current_media_status} -> {next_request_status}/{next_media_status}", ) comments = list_portal_comments(item_id, include_internal=True) return { "item": _serialize_item(updated, current_user), "comments": comments, } @router.get("/items/{item_id}") async def portal_get_item( item_id: int, current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: item = get_portal_item(item_id) if not item: raise HTTPException(status_code=404, detail="Portal item not found") comments = list_portal_comments(item_id, include_internal=_is_admin(current_user)) return { "item": _serialize_item(item, current_user), "comments": comments, } @router.patch("/items/{item_id}") async def portal_update_item( item_id: int, payload: Dict[str, Any], current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: item = get_portal_item(item_id) if not item: raise HTTPException(status_code=404, detail="Portal item not found") is_admin = _is_admin(current_user) is_owner = _is_owner(current_user, item) if not (is_admin or is_owner): raise HTTPException(status_code=403, detail="Only the owner or admin can edit this item") editable_owner_fields = {"title", "description", "media_type", "year", "external_ref"} editable_admin_fields = { "status", "priority", "assignee_username", "source_system", "source_request_id", "related_item_id", "request_status", "media_status", "issue_type", "issue_resolved_at", "metadata_json", } provided_fields = set(payload.keys()) unknown_fields = provided_fields - (editable_owner_fields | editable_admin_fields) if unknown_fields: unknown = ", ".join(sorted(unknown_fields)) raise HTTPException(status_code=400, detail=f"Unsupported fields: {unknown}") if not is_admin: forbidden = provided_fields - editable_owner_fields if forbidden: forbidden_text = ", ".join(sorted(forbidden)) raise HTTPException( status_code=403, detail=f"Admin access required to update: {forbidden_text}" ) updates: Dict[str, Any] = {} if "title" in payload: updates["title"] = _require_text(payload.get("title"), "title", max_length=220) if "description" in payload: updates["description"] = _require_text( payload.get("description"), "description", max_length=10000 ) if "media_type" in payload: updates["media_type"] = _normalize_choice( payload.get("media_type"), field="media_type", allowed=PORTAL_MEDIA_TYPES, allow_empty=True, ) if "year" in payload: updates["year"] = _normalize_year(payload.get("year")) if "external_ref" in payload: updates["external_ref"] = _clean_text(payload.get("external_ref")) if is_admin: kind = str(item.get("kind") or "").lower() if "priority" in payload: updates["priority"] = _normalize_choice( payload.get("priority"), field="priority", allowed=PORTAL_PRIORITIES, default=item.get("priority") or "normal", ) if "assignee_username" in payload: updates["assignee_username"] = _clean_text(payload.get("assignee_username")) if "source_system" in payload: source_system = _clean_text(payload.get("source_system")) updates["source_system"] = source_system.lower() if source_system else None if "source_request_id" in payload: updates["source_request_id"] = _normalize_int( payload.get("source_request_id"), "source_request_id" ) if "related_item_id" in payload: related_item_id = _normalize_int(payload.get("related_item_id"), "related_item_id") _ensure_item_exists(related_item_id) updates["related_item_id"] = related_item_id if "metadata_json" in payload: updates["metadata_json"] = _sanitize_metadata_json(payload.get("metadata_json")) if kind == "request": current_request_status, current_media_status = _item_status_to_workflow(item) request_status_input = payload.get("request_status") media_status_input = payload.get("media_status") explicit_status = payload.get("status") if explicit_status is not None and request_status_input is None and media_status_input is None: explicit_status_normalized = _normalize_choice( explicit_status, field="status", allowed=PORTAL_STATUSES, default=item.get("status") or "pending", ) request_status_input, media_status_input = LEGACY_STATUS_TO_WORKFLOW.get( explicit_status_normalized or "pending", (current_request_status, current_media_status), ) if request_status_input is not None or media_status_input is not None: requested_request = _normalize_choice( request_status_input, field="request_status", allowed=PORTAL_REQUEST_STATUSES, default=current_request_status, ) or current_request_status requested_media = _normalize_choice( media_status_input, field="media_status", allowed=PORTAL_MEDIA_STATUSES, default=current_media_status, ) or current_media_status next_request_status, next_media_status = _validate_pipeline_transition( current_request_status, current_media_status, requested_request, requested_media, ) updates["workflow_request_status"] = next_request_status updates["workflow_media_status"] = next_media_status updates["status"] = _workflow_to_item_status(next_request_status, next_media_status) elif "status" in payload: updates["status"] = _normalize_choice( payload.get("status"), field="status", allowed=PORTAL_STATUSES, default=item.get("status") or "pending", ) else: if "status" in payload: updates["status"] = _normalize_choice( payload.get("status"), field="status", allowed=PORTAL_STATUSES, default=item.get("status") or "new", ) if kind == "issue": if "issue_type" in payload: updates["issue_type"] = _normalize_choice( payload.get("issue_type"), field="issue_type", allowed=PORTAL_ISSUE_TYPES, default=item.get("issue_type") or "general", ) if "issue_resolved_at" in payload: updates["issue_resolved_at"] = _clean_text(payload.get("issue_resolved_at")) if "status" in payload: next_status = str(updates.get("status") or item.get("status") or "").lower() if next_status in {"done", "closed"}: updates.setdefault("issue_resolved_at", datetime.now(timezone.utc).isoformat()) elif next_status in {"new", "triaging", "planned", "in_progress", "blocked"}: updates.setdefault("issue_resolved_at", None) if not updates: comments = list_portal_comments(item_id, include_internal=is_admin) return { "item": _serialize_item(item, current_user), "comments": comments, } updated = update_portal_item(item_id, **updates) if not updated: raise HTTPException(status_code=404, detail="Portal item not found") changed_fields = [key for key in updates.keys() if item.get(key) != updated.get(key)] if changed_fields: await _notify( event_type="portal_item_updated", item=updated, user=current_user, note=f"changed={','.join(sorted(changed_fields))}", ) comments = list_portal_comments(item_id, include_internal=is_admin) return { "item": _serialize_item(updated, current_user), "comments": comments, } @router.get("/items/{item_id}/comments") async def portal_get_comments( item_id: int, limit: int = Query(default=200, ge=1, le=500), current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: item = get_portal_item(item_id) if not item: raise HTTPException(status_code=404, detail="Portal item not found") comments = list_portal_comments( item_id, include_internal=_is_admin(current_user), limit=limit, ) return {"comments": comments} @router.post("/items/{item_id}/comments") async def portal_create_comment( item_id: int, payload: Dict[str, Any], current_user: Dict[str, Any] = Depends(get_current_user), ) -> Dict[str, Any]: item = get_portal_item(item_id) if not item: raise HTTPException(status_code=404, detail="Portal item not found") is_admin = _is_admin(current_user) message = _require_text(payload.get("message"), "message", max_length=10000) is_internal = _normalize_bool(payload.get("is_internal"), default=False) if is_internal and not is_admin: raise HTTPException(status_code=403, detail="Only admins can add internal comments") comment = add_portal_comment( item_id, author_username=str(current_user.get("username") or "unknown"), author_role=str(current_user.get("role") or "user"), message=message, is_internal=is_internal, ) updated_item = get_portal_item(item_id) if updated_item: await _notify( event_type="portal_comment_added", item=updated_item, user=current_user, note=f"internal={is_internal}", ) return {"comment": comment}