feat: extend admin operations and media review

apps/api/app/routers/admin.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Annotated, Literal
 from uuid import UUID
 
 import httpx
 from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response
-from fastapi.responses import JSONResponse
+from fastapi.responses import FileResponse, JSONResponse
 from sqlalchemy import case, func, or_, select
 from sqlalchemy.orm import joinedload
 
@@ -15,6 +15,7 @@ from ..community_review import ExternalReviewError, map_observation, publish_obs
 from ..config import settings
 from ..dependencies import Db
 from ..importer import ImportAlreadyRunning, ImportSourceError, import_records
+from ..media_catalog import review_assets, review_file
 from ..models import CatchReport, CommunityImportRun, DataSource, ExternalObservation, Fish, ModerationEvent, ModerationStatus, OfficialRecordImport, SourceType, Waterbody
 from ..public_cache import public_cache
 from ..schemas import AdminCatchReportOut, AdminModerationHistoryOut, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationAction, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, ImportRunOut, ModerationUpdate
@@ -29,6 +30,26 @@ def _admin(request: Request, db: Db, authorization: Annotated[str | None, Header
     return verify_admin(request, db, authorization, settings)
 
 
+@router.get("/api/v1/admin/media/catalog")
+def admin_media_catalog(
+    _: Annotated[str, Depends(_admin)],
+    entity_type: str | None = Query(None, pattern="^(fish|waterbody|tackle|reference)$"),
+    status: str | None = Query(None, pattern="^(approved|upgrade_queued|upgrade_stored)$"),
+    limit: int = Query(50, ge=1, le=100),
+    offset: int = Query(0, ge=0),
+) -> list[dict]:
+    return review_assets(entity_type, status)[offset:offset + limit]
+
+
+@router.get("/api/v1/admin/media/assets/{digest}", response_class=FileResponse)
+def admin_media_asset(digest: str, _: Annotated[str, Depends(_admin)]) -> FileResponse:
+    item = review_file(digest)
+    if not item:
+        raise HTTPException(status_code=404, detail="Media review asset not found")
+    path, media_type = item
+    return FileResponse(path, media_type=media_type, headers={"Cache-Control": "private, no-store"})
+
+
 @router.get("/api/v1/admin/diagnostics")
 def admin_diagnostics(db: Db, _: Annotated[str, Depends(_admin)]) -> JSONResponse:
     report_counts = {status.value: count for status, count in db.execute(
@@ -131,6 +152,52 @@ def admin_start_official_import(db: Db, _: Annotated[str, Depends(_admin)]) -> O
         raise HTTPException(status_code=502, detail=f"official records import failed: {exc}") from exc
 
 
+@router.get("/api/v1/admin/source-status")
+def admin_source_status(db: Db, _: Annotated[str, Depends(_admin)]) -> list[dict[str, object]]:
+    """Return safe operational details needed by the owner dashboard."""
+    now = datetime.now(timezone.utc)
+    result: list[dict[str, object]] = []
+    for source in db.scalars(select(DataSource).order_by(DataSource.name)):
+        runs = list(db.scalars(
+            select(CommunityImportRun)
+            .where(CommunityImportRun.source_system == source.key)
+            .order_by(CommunityImportRun.started_at.desc()).limit(20)
+        ))
+        latest = runs[0] if runs else None
+        success = next((run for run in runs if run.status == "success"), None)
+        recent_failures = sum(
+            1 for run in runs
+            if run.status == "failed" and aware(run.started_at) >= now - timedelta(hours=24)
+        )
+        next_allowed = (
+            aware(latest.started_at) + timedelta(seconds=settings.community_import_interval_seconds)
+            if latest else None
+        )
+        cooldown_seconds = max(0, int((next_allowed - now).total_seconds())) if next_allowed else 0
+        if not source.enabled:
+            state = "disabled"
+        elif latest is None:
+            state = "waiting"
+        elif latest.status == "failed":
+            state = "source_changed" if "CommunityParseError" in (latest.error_summary or "") else "temporarily_limited"
+        elif aware(latest.started_at) < now - timedelta(seconds=settings.community_import_interval_seconds * 2):
+            state = "stale"
+        else:
+            state = "healthy"
+        result.append({
+            "source_system": source.key,
+            "name": source.name,
+            "status": state,
+            "last_started_at": latest.started_at if latest else None,
+            "last_success_at": success.started_at if success else None,
+            "next_allowed_at": next_allowed,
+            "cooldown_seconds": cooldown_seconds,
+            "recent_failures_24h": recent_failures,
+            "backoff_recommended": recent_failures >= 5,
+        })
+    return result
+
+
 def _external_out(item: ExternalObservation) -> ExternalObservationOut:
     allowed_payload = {
         key: value for key, value in (item.payload or {}).items()