332 lines
18 KiB
Python
332 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, 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 sqlalchemy import case, func, or_, select
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from ..admin_security import verify_admin
|
|
from ..community_review import ExternalReviewError, map_observation, publish_observation, reject_observation, suggest_aliases
|
|
from ..config import settings
|
|
from ..dependencies import Db
|
|
from ..importer import ImportAlreadyRunning, ImportSourceError, import_records
|
|
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
|
|
from ..storage import delete_screenshot, signed_screenshot_url
|
|
from ..time_utils import aware
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _admin(request: Request, db: Db, authorization: Annotated[str | None, Header()] = None) -> str:
|
|
return verify_admin(request, db, authorization, settings)
|
|
|
|
|
|
@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(
|
|
select(CatchReport.moderation_status, func.count()).group_by(CatchReport.moderation_status)
|
|
)}
|
|
observation_counts = {status: count for status, count in db.execute(
|
|
select(ExternalObservation.status, func.count()).group_by(ExternalObservation.status)
|
|
)}
|
|
payload = {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"build": {"version": settings.app_version, "revision": settings.app_revision, "environment": settings.deployment_environment},
|
|
"counts": {
|
|
"catch_reports": report_counts,
|
|
"external_observations": observation_counts,
|
|
"data_sources": db.scalar(select(func.count()).select_from(DataSource)) or 0,
|
|
"enabled_data_sources": db.scalar(select(func.count()).select_from(DataSource).where(DataSource.enabled.is_(True))) or 0,
|
|
"official_import_runs": db.scalar(select(func.count()).select_from(OfficialRecordImport)) or 0,
|
|
"community_import_runs": db.scalar(select(func.count()).select_from(CommunityImportRun)) or 0,
|
|
},
|
|
}
|
|
return JSONResponse(payload, headers={"Content-Disposition": "attachment; filename=rf4spotter-diagnostics.json"})
|
|
|
|
|
|
@router.get("/api/v1/admin/moderation-history", response_model=list[AdminModerationHistoryOut])
|
|
def admin_moderation_history(
|
|
db: Db,
|
|
_: Annotated[str, Depends(_admin)],
|
|
limit: int = Query(50, ge=1, le=200),
|
|
offset: int = Query(0, ge=0),
|
|
) -> list[AdminModerationHistoryOut]:
|
|
report_events = list(db.scalars(
|
|
select(ModerationEvent).order_by(ModerationEvent.created_at.desc()).limit(limit + offset)
|
|
))
|
|
external_events = list(db.scalars(
|
|
select(ExternalObservation).where(ExternalObservation.reviewed_at.is_not(None))
|
|
.order_by(ExternalObservation.reviewed_at.desc()).limit(limit + offset)
|
|
))
|
|
history = [AdminModerationHistoryOut(
|
|
entity_type="catch_report", entity_id=event.catch_report_id,
|
|
decided_at=event.created_at, action=event.new_status.value,
|
|
moderator=event.moderator, reason=event.reason,
|
|
) for event in report_events]
|
|
history.extend(AdminModerationHistoryOut(
|
|
entity_type="external_observation", entity_id=observation.id,
|
|
decided_at=observation.reviewed_at, action=observation.status,
|
|
moderator=None, reason=observation.review_note,
|
|
) for observation in external_events if observation.reviewed_at is not None)
|
|
history.sort(key=lambda event: aware(event.decided_at), reverse=True)
|
|
return history[offset:offset + limit]
|
|
|
|
|
|
@router.get("/api/v1/admin/moderation-history-export")
|
|
def admin_moderation_history_export(
|
|
db: Db,
|
|
_: Annotated[str, Depends(_admin)],
|
|
limit: int = Query(1000, ge=1, le=5000),
|
|
) -> JSONResponse:
|
|
"""Return an anonymized, analysis-safe decision export."""
|
|
events = admin_moderation_history(db, _, limit=limit, offset=0)
|
|
payload = {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"count": len(events),
|
|
"events": [{
|
|
"entity_type": event.entity_type,
|
|
"decided_at": event.decided_at.isoformat(),
|
|
"action": event.action,
|
|
"requires_confirmation": event.requires_confirmation,
|
|
} for event in events],
|
|
}
|
|
return JSONResponse(payload, headers={
|
|
"Content-Disposition": "attachment; filename=rf4spotter-moderation-history.json",
|
|
})
|
|
|
|
|
|
@router.get("/api/v1/admin/imports", response_model=list[ImportRunOut])
|
|
def admin_imports(
|
|
db: Db,
|
|
_: Annotated[str, Depends(_admin)],
|
|
limit: int = Query(20, ge=1, le=100),
|
|
offset: int = Query(0, ge=0),
|
|
) -> list[OfficialRecordImport]:
|
|
query = select(OfficialRecordImport).order_by(
|
|
OfficialRecordImport.started_at.desc(), OfficialRecordImport.id.desc()
|
|
).offset(offset).limit(limit)
|
|
return list(db.scalars(query))
|
|
|
|
|
|
@router.post("/api/v1/admin/imports/official-records", response_model=ImportRunOut, status_code=201)
|
|
def admin_start_official_import(db: Db, _: Annotated[str, Depends(_admin)]) -> OfficialRecordImport:
|
|
try:
|
|
return import_records(
|
|
db,
|
|
url=settings.official_records_url,
|
|
region=settings.official_records_region,
|
|
category=settings.official_records_category,
|
|
)
|
|
except ImportAlreadyRunning as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except (ImportSourceError, httpx.HTTPError) as exc:
|
|
raise HTTPException(status_code=502, detail=f"official records import failed: {exc}") from exc
|
|
|
|
|
|
def _external_out(item: ExternalObservation) -> ExternalObservationOut:
|
|
allowed_payload = {
|
|
key: value for key, value in (item.payload or {}).items()
|
|
if key in {
|
|
"bait", "fishing_method", "rig_type", "retrieve_method", "retrieve_speed",
|
|
"player_name", "published_at", "region", "category",
|
|
} and (value is None or isinstance(value, (str, int, float, bool)))
|
|
}
|
|
missing_fields = []
|
|
if item.x is None or item.y is None:
|
|
missing_fields.append("coordinates")
|
|
if item.weight_g is None:
|
|
missing_fields.append("weight_g")
|
|
return ExternalObservationOut(
|
|
id=item.id, source_system=item.source_system, source_external_id=item.source_external_id,
|
|
source_url=item.source_url, fish_name=item.fish_name, fish_external_id=item.fish_external_id,
|
|
waterbody_name=item.waterbody_name, waterbody_external_id=item.waterbody_external_id,
|
|
x=item.x, y=item.y, weight_g=item.weight_g, published_at=item.published_at,
|
|
first_seen_at=item.first_seen_at, last_seen_at=item.last_seen_at, reviewed_at=item.reviewed_at,
|
|
status=item.status,
|
|
fish_slug=item.fish.slug if item.fish else None,
|
|
waterbody_slug=item.waterbody.slug if item.waterbody else None,
|
|
catch_report_id=item.catch_report_id, review_note=item.review_note,
|
|
missing_fields=missing_fields, source_payload=allowed_payload,
|
|
moderation_version=item.moderation_version,
|
|
source_check_status=item.source_check_status, source_checked_at=item.source_checked_at,
|
|
)
|
|
|
|
|
|
@router.get("/api/v1/admin/external-observations", response_model=list[ExternalObservationOut])
|
|
def admin_external_observations(
|
|
db: Db, _: Annotated[str, Depends(_admin)],
|
|
status: Literal["staged", "mapped", "ready", "published", "rejected", "withdrawn", "review"] | None = None,
|
|
source_system: str | None = None,
|
|
completeness: Literal["all", "complete", "incomplete"] = "all",
|
|
order: Literal["newest", "oldest", "risk"] = "newest",
|
|
q: str | None = Query(None, max_length=100),
|
|
limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0),
|
|
) -> list[ExternalObservationOut]:
|
|
query = select(ExternalObservation).options(
|
|
joinedload(ExternalObservation.fish), joinedload(ExternalObservation.waterbody),
|
|
)
|
|
if status == "review":
|
|
query = query.where(ExternalObservation.status.in_(["staged", "mapped", "ready"]))
|
|
elif status:
|
|
query = query.where(ExternalObservation.status == status)
|
|
if source_system:
|
|
query = query.where(ExternalObservation.source_system == source_system)
|
|
if completeness == "complete":
|
|
query = query.where(
|
|
ExternalObservation.x.is_not(None), ExternalObservation.y.is_not(None),
|
|
ExternalObservation.weight_g.is_not(None),
|
|
)
|
|
elif completeness == "incomplete":
|
|
query = query.where(or_(
|
|
ExternalObservation.x.is_(None), ExternalObservation.y.is_(None),
|
|
ExternalObservation.weight_g.is_(None),
|
|
))
|
|
if q and q.strip():
|
|
term = q.strip()
|
|
query = query.where(or_(
|
|
ExternalObservation.fish_name.icontains(term, autoescape=True),
|
|
ExternalObservation.waterbody_name.icontains(term, autoescape=True),
|
|
))
|
|
if order == "risk":
|
|
incomplete = case(
|
|
(or_(ExternalObservation.x.is_(None), ExternalObservation.y.is_(None), ExternalObservation.weight_g.is_(None)), 0),
|
|
else_=1,
|
|
)
|
|
workflow = case(
|
|
(ExternalObservation.status == "staged", 0),
|
|
(ExternalObservation.status == "mapped", 1),
|
|
else_=2,
|
|
)
|
|
ordering = (incomplete, workflow, ExternalObservation.last_seen_at.asc(), ExternalObservation.id.desc())
|
|
else:
|
|
direction = ExternalObservation.last_seen_at.asc() if order == "oldest" else ExternalObservation.last_seen_at.desc()
|
|
ordering = (direction, ExternalObservation.id.desc())
|
|
items = db.scalars(query.order_by(*ordering).offset(offset).limit(limit))
|
|
return [_external_out(item) for item in items]
|
|
|
|
|
|
@router.get("/api/v1/admin/external-observations/{observation_id}/alias-suggestions", response_model=ExternalAliasSuggestionOut)
|
|
def admin_external_alias_suggestions(
|
|
observation_id: UUID, db: Db, _: Annotated[str, Depends(_admin)],
|
|
) -> ExternalAliasSuggestionOut:
|
|
observation = db.get(ExternalObservation, observation_id)
|
|
if observation is None:
|
|
raise HTTPException(status_code=404, detail="external observation not found")
|
|
fish, waterbody = suggest_aliases(db, observation)
|
|
return ExternalAliasSuggestionOut(
|
|
fish_slug=fish.slug if fish else None,
|
|
waterbody_slug=waterbody.slug if waterbody else None,
|
|
)
|
|
|
|
|
|
@router.patch("/api/v1/admin/external-observations/{observation_id}/mapping", response_model=ExternalObservationOut)
|
|
def admin_map_external_observation(
|
|
observation_id: UUID, payload: ExternalObservationMapping, db: Db,
|
|
_: Annotated[str, Depends(_admin)],
|
|
) -> ExternalObservationOut:
|
|
observation = db.scalar(select(ExternalObservation).where(ExternalObservation.id == observation_id).with_for_update())
|
|
fish = db.scalar(select(Fish).where(Fish.slug == payload.fish_slug))
|
|
waterbody = db.scalar(select(Waterbody).where(Waterbody.slug == payload.waterbody_slug))
|
|
if observation is None:
|
|
raise HTTPException(status_code=404, detail="external observation not found")
|
|
if fish is None or waterbody is None:
|
|
raise HTTPException(status_code=422, detail="unknown fish or waterbody")
|
|
if observation.moderation_version != payload.expected_version:
|
|
raise HTTPException(status_code=409, detail="observation changed; reload the queue")
|
|
observation.moderation_version += 1
|
|
try:
|
|
return _external_out(map_observation(db, observation, fish, waterbody, note=payload.note))
|
|
except ExternalReviewError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
|
|
|
|
@router.post("/api/v1/admin/external-observations/{observation_id}/publish", response_model=ExternalObservationPublished)
|
|
def admin_publish_external_observation(
|
|
observation_id: UUID, payload: ExternalObservationAction, db: Db, _: Annotated[str, Depends(_admin)],
|
|
) -> ExternalObservationPublished:
|
|
observation = db.scalar(select(ExternalObservation).where(ExternalObservation.id == observation_id).with_for_update())
|
|
if observation is None:
|
|
raise HTTPException(status_code=404, detail="external observation not found")
|
|
if observation.moderation_version != payload.expected_version:
|
|
raise HTTPException(status_code=409, detail="observation changed; reload the queue")
|
|
observation.moderation_version += 1
|
|
try:
|
|
report = publish_observation(db, observation)
|
|
except ExternalReviewError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
public_cache.invalidate()
|
|
return ExternalObservationPublished(observation_id=observation.id, catch_report_id=report.id, status=observation.status)
|
|
|
|
|
|
@router.patch("/api/v1/admin/external-observations/{observation_id}/reject", response_model=ExternalObservationOut)
|
|
def admin_reject_external_observation(
|
|
observation_id: UUID, payload: ExternalObservationDecision, db: Db,
|
|
_: Annotated[str, Depends(_admin)],
|
|
) -> ExternalObservationOut:
|
|
observation = db.scalar(select(ExternalObservation).where(ExternalObservation.id == observation_id).with_for_update())
|
|
if observation is None:
|
|
raise HTTPException(status_code=404, detail="external observation not found")
|
|
if observation.moderation_version != payload.expected_version:
|
|
raise HTTPException(status_code=409, detail="observation changed; reload the queue")
|
|
observation.moderation_version += 1
|
|
try:
|
|
return _external_out(reject_observation(db, observation, reason=payload.reason))
|
|
except ExternalReviewError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
|
|
|
|
@router.get("/api/v1/admin/catch-reports", response_model=list[AdminCatchReportOut])
|
|
def admin_reports(db: Db, _: Annotated[str, Depends(_admin)], status: ModerationStatus = ModerationStatus.pending, limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0)) -> list[AdminCatchReportOut]:
|
|
reports = list(db.scalars(select(CatchReport).options(joinedload(CatchReport.fish), joinedload(CatchReport.waterbody), joinedload(CatchReport.spot), joinedload(CatchReport.bait)).where(CatchReport.source_type == SourceType.user, CatchReport.moderation_status == status, CatchReport.deleted_at.is_(None)).order_by(CatchReport.reported_at, CatchReport.id).offset(offset).limit(limit)))
|
|
return [AdminCatchReportOut(id=r.id, fish=r.fish.name_ru, waterbody=r.waterbody.name_ru, coordinates=f"{r.spot.x}:{r.spot.y}" if r.spot else "—", weight_g=r.weight_g, bait=r.bait.name if r.bait else None, player_name=r.player_name, reported_at=r.reported_at, moderation_status=r.moderation_status.value, comment=(r.raw_payload or {}).get("comment"), screenshot_url=signed_screenshot_url(r.screenshot_key) if r.screenshot_key else None, moderation_version=r.moderation_version) for r in reports]
|
|
|
|
|
|
@router.patch("/api/v1/admin/catch-reports/{report_id}", response_model=CatchReportCreated)
|
|
def moderate_report(report_id: UUID, payload: ModerationUpdate, db: Db, moderator: Annotated[str, Depends(_admin)]) -> CatchReportCreated:
|
|
report = db.scalar(select(CatchReport).where(CatchReport.id == report_id).with_for_update())
|
|
if report is None or report.source_type != SourceType.user or report.deleted_at is not None:
|
|
raise HTTPException(status_code=404, detail="catch report not found")
|
|
if report.moderation_version != payload.expected_version:
|
|
raise HTTPException(status_code=409, detail="report changed; reload the queue")
|
|
previous = report.moderation_status
|
|
report.moderation_status = ModerationStatus(payload.status)
|
|
report.moderation_version += 1
|
|
db.add(ModerationEvent(catch_report=report, created_at=datetime.now(timezone.utc), previous_status=previous, new_status=report.moderation_status, moderator=moderator, reason=payload.reason))
|
|
db.commit()
|
|
public_cache.invalidate()
|
|
return CatchReportCreated(id=report.id, moderation_status=report.moderation_status.value)
|
|
|
|
|
|
@router.delete("/api/v1/admin/catch-reports/{report_id}", status_code=204, response_class=Response)
|
|
def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_admin)], expected_version: int = Query(ge=0)) -> Response:
|
|
report = db.scalar(select(CatchReport).where(CatchReport.id == report_id).with_for_update())
|
|
if report is None or report.source_type != SourceType.user or report.deleted_at is not None:
|
|
raise HTTPException(status_code=404, detail="catch report not found")
|
|
if report.moderation_version != expected_version:
|
|
raise HTTPException(status_code=409, detail="report changed; reload the queue")
|
|
previous = report.moderation_status
|
|
if report.screenshot_key:
|
|
try:
|
|
delete_screenshot(report.screenshot_key)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=502, detail="screenshot deletion failed") from exc
|
|
report.moderation_status = ModerationStatus.rejected
|
|
report.moderation_version += 1
|
|
report.deleted_at = datetime.now(timezone.utc)
|
|
report.player_name = None
|
|
report.source_url = None
|
|
report.screenshot_key = None
|
|
report.raw_payload = None
|
|
db.add(ModerationEvent(catch_report=report, created_at=report.deleted_at, previous_status=previous, new_status=ModerationStatus.rejected, moderator=moderator, reason="user report deleted and anonymized"))
|
|
db.commit()
|
|
public_cache.invalidate()
|
|
return Response(status_code=204)
|