from __future__ import annotations from collections import Counter from datetime import datetime, timedelta, timezone import hashlib import hmac import logging import time as time_module from typing import Annotated, Literal from uuid import UUID import uuid import httpx from fastapi import Depends, FastAPI, File, Header, HTTPException, Query, Request, Response, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from sqlalchemy import delete, func, select, text from sqlalchemy.orm import Session, joinedload from .activity import activity_rows from .database import get_session from .config import settings from .community_review import ExternalReviewError, map_observation, publish_observation, reject_observation from .importer import ImportSourceError, import_records, normalize from .logging_config import configure_logging from .models import Bait, BaitKind, CatchReport, ExternalObservation, Fish, ModerationEvent, ModerationStatus, OfficialRecordImport, SourceType, Spot, SubmissionAttempt, Waterbody from .readiness import readiness_report from .schemas import ActivityOut, AdminCatchReportOut, BaitOut, CatchOut, CatchReportCreate, CatchReportCreated, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, FishOut, ImportRunOut, ModerationUpdate, OfficialRecordOut, SpotOut, WaterbodyOut from .storage import ScreenshotError, client as storage_client, delete_screenshot, signed_screenshot_url, upload_screenshot configure_logging(settings.log_level) logger = logging.getLogger("rf4.api") app = FastAPI(title="RF4 Spotter API", version="0.1.0") app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:4321", "http://127.0.0.1:4321"], allow_methods=["GET", "POST", "PATCH", "DELETE"], allow_headers=["Authorization", "Content-Type"], ) Db = Annotated[Session, Depends(get_session)] @app.middleware("http") async def structured_request_log(request: Request, call_next): request_id = uuid.uuid4().hex started = time_module.perf_counter() status_code = 500 try: response = await call_next(request) status_code = response.status_code response.headers["X-Request-ID"] = request_id return response except Exception as exc: logger.error("request failed", extra={"request_id": request_id, "error_type": type(exc).__name__}) raise finally: logger.log( logging.DEBUG if request.url.path in {"/health", "/ready"} else logging.INFO, "request completed", extra={ "request_id": request_id, "method": request.method, "path": request.url.path, "status_code": status_code, "duration_ms": round((time_module.perf_counter() - started) * 1000, 2), }, ) @app.get("/health") def health() -> dict[str, str]: return {"status": "ok"} @app.get("/ready") def ready(db: Db) -> JSONResponse: is_ready, components = readiness_report( db, storage_client(), import_required=settings.official_import_required, import_interval_seconds=settings.import_interval_seconds, ) return JSONResponse( status_code=200 if is_ready else 503, content={"status": "ready" if is_ready else "not_ready", "components": components}, ) @app.get("/api/v1/fishes", response_model=list[FishOut]) def fishes(db: Db) -> list[Fish]: return list(db.scalars(select(Fish).order_by(Fish.name_ru))) @app.get("/api/v1/waterbodies", response_model=list[WaterbodyOut]) def waterbodies(db: Db) -> list[Waterbody]: return list(db.scalars(select(Waterbody).order_by(Waterbody.name_ru))) @app.get("/api/v1/baits", response_model=list[BaitOut]) def baits(db: Db) -> list[Bait]: return list(db.scalars(select(Bait).order_by(Bait.name))) @app.get("/api/v1/activity", response_model=list[ActivityOut]) def activity( db: Db, hours: int = Query(24), waterbody: str | None = None, fish: str | None = None, method: str | None = None, sort: Literal["activity", "confidence", "freshness"] = "activity", limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ) -> list[ActivityOut]: if hours not in {6, 12, 24, 72}: raise HTTPException(status_code=422, detail="hours must be one of: 6, 12, 24, 72") rows = activity_rows(db, hours=hours, waterbody=waterbody, fish=fish, method=method) keys = {"activity": lambda r: r.activity_score, "confidence": lambda r: r.confidence_score, "freshness": lambda r: r.last_confirmed_at} rows.sort(key=keys[sort], reverse=True) return rows[offset:offset + limit] def _spot_or_404(db: Session, spot_id: UUID) -> Spot: spot = db.scalar(select(Spot).options(joinedload(Spot.waterbody)).where(Spot.id == spot_id)) if spot is None: raise HTTPException(status_code=404, detail="spot not found") return spot @app.get("/api/v1/spots/{spot_id}", response_model=SpotOut) def spot_detail(spot_id: UUID, db: Db) -> SpotOut: spot = _spot_or_404(db, spot_id) reports = list(db.scalars(select(CatchReport).options(joinedload(CatchReport.bait)).where(CatchReport.spot_id == spot.id, CatchReport.moderation_status == ModerationStatus.approved, CatchReport.deleted_at.is_(None)))) now = datetime.now(timezone.utc) def count_since(delta: timedelta) -> int: return sum(_aware(r.reported_at) >= now - delta for r in reports) bait_counts = Counter(r.bait.name for r in reports if r.bait) return SpotOut(id=spot.id, waterbody_slug=spot.waterbody.slug, waterbody=spot.waterbody.name_ru, x=spot.x, y=spot.y, description=spot.description, catches_24h=count_since(timedelta(hours=24)), catches_3d=count_since(timedelta(days=3)), catches_7d=count_since(timedelta(days=7)), top_baits=[name for name, _ in bait_counts.most_common(5)]) @app.get("/api/v1/spots/{spot_id}/catches", response_model=list[CatchOut]) def spot_catches(spot_id: UUID, db: Db, limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0)) -> list[CatchOut]: _spot_or_404(db, spot_id) reports = list(db.scalars(select(CatchReport).options(joinedload(CatchReport.fish), joinedload(CatchReport.bait)).where(CatchReport.spot_id == spot_id, CatchReport.moderation_status == ModerationStatus.approved, CatchReport.deleted_at.is_(None)).order_by(CatchReport.reported_at.desc()).offset(offset).limit(limit))) return [CatchOut(id=r.id, fish=r.fish.name_ru, weight_g=r.weight_g, bait=r.bait.name if r.bait else None, player_name=r.player_name, caught_at=r.caught_at, reported_at=r.reported_at, retrieve_method=r.retrieve_method, retrieve_speed=r.retrieve_speed) for r in reports] @app.get("/api/v1/records", response_model=list[OfficialRecordOut]) def records( db: Db, fish: str | None = None, waterbody: str | None = None, category: str | None = None, limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0), ) -> list[OfficialRecordOut]: query = select(CatchReport).options(joinedload(CatchReport.fish), joinedload(CatchReport.waterbody), joinedload(CatchReport.bait)).where(CatchReport.source_type == SourceType.official_record) if fish: query = query.join(CatchReport.fish).where(Fish.slug == fish) if waterbody: query = query.join(CatchReport.waterbody).where(Waterbody.slug == waterbody) items = list(db.scalars(query.order_by(CatchReport.caught_at.desc(), CatchReport.weight_g.desc()).offset(offset).limit(limit))) if category: items = [item for item in items if (item.raw_payload or {}).get("category") == category] return [OfficialRecordOut(id=r.id, fish=r.fish.name_ru, weight_g=r.weight_g, waterbody=r.waterbody.name_ru, bait=r.bait.name if r.bait else None, player_name=r.player_name, record_date=r.caught_at, category=(r.raw_payload or {}).get("category"), region=(r.raw_payload or {}).get("region"), source_url=r.source_url) for r in items] def _admin(authorization: Annotated[str | None, Header()] = None) -> str: if not authorization or authorization != f"Bearer {settings.admin_token}": raise HTTPException(status_code=401, detail="invalid admin token", headers={"WWW-Authenticate": "Bearer"}) return "admin" @app.get("/api/v1/imports", response_model=list[ImportRunOut]) def imports(db: Db, limit: int = Query(20, ge=1, le=100)) -> list[OfficialRecordImport]: return list(db.scalars(select(OfficialRecordImport).order_by(OfficialRecordImport.started_at.desc()).limit(limit))) @app.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)) @app.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 (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: 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, last_seen_at=item.last_seen_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, ) @app.get("/api/v1/admin/external-observations", response_model=list[ExternalObservationOut]) def admin_external_observations( db: Db, _: Annotated[str, Depends(_admin)], status: str | None = None, source_system: str | None = None, 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: query = query.where(ExternalObservation.status == status) if source_system: query = query.where(ExternalObservation.source_system == source_system) items = db.scalars(query.order_by(ExternalObservation.last_seen_at.desc()).offset(offset).limit(limit)) return [_external_out(item) for item in items] @app.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.get(ExternalObservation, observation_id) 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") 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 @app.post("/api/v1/admin/external-observations/{observation_id}/publish", response_model=ExternalObservationPublished) def admin_publish_external_observation( observation_id: UUID, db: Db, _: Annotated[str, Depends(_admin)], ) -> ExternalObservationPublished: observation = db.get(ExternalObservation, observation_id) if observation is None: raise HTTPException(status_code=404, detail="external observation not found") try: report = publish_observation(db, observation) except ExternalReviewError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc return ExternalObservationPublished(observation_id=observation.id, catch_report_id=report.id, status=observation.status) @app.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.get(ExternalObservation, observation_id) if observation is None: raise HTTPException(status_code=404, detail="external observation not found") 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 @app.post("/api/v1/catch-reports", response_model=CatchReportCreated, status_code=201) def create_catch_report(payload: CatchReportCreate, request: Request, db: Db) -> CatchReportCreated: if payload.website: raise HTTPException(status_code=400, detail="invalid submission") _check_rate_limit(request.client.host if request.client else "unknown", db) fish = db.scalar(select(Fish).where(Fish.slug == payload.fish_slug)) waterbody = db.scalar(select(Waterbody).where(Waterbody.slug == payload.waterbody_slug)) if fish is None or waterbody is None: raise HTTPException(status_code=422, detail="unknown fish or waterbody") spot = db.scalar(select(Spot).where(Spot.waterbody_id == waterbody.id, Spot.x == payload.x, Spot.y == payload.y)) if spot is None: spot = Spot(waterbody=waterbody, x=payload.x, y=payload.y) db.add(spot) bait = None if payload.bait_name and payload.bait_name.strip(): key = normalize(payload.bait_name) bait = db.scalar(select(Bait).where(Bait.normalized_name == key)) if bait is None: bait = Bait(name=payload.bait_name.strip(), normalized_name=key, kind=BaitKind.unknown) db.add(bait) report = CatchReport(fish=fish, spot=spot, waterbody=waterbody, bait=bait, weight_g=payload.weight_g, fishing_method=payload.fishing_method, rig_type=payload.rig_type, retrieve_method=payload.retrieve_method, retrieve_speed=payload.retrieve_speed, caught_at=payload.caught_at, reported_at=datetime.now(timezone.utc), player_name=payload.player_name, source_type=SourceType.user, source_url=payload.source_url, source_confidence=60, moderation_status=ModerationStatus.pending, raw_payload={"comment": payload.comment} if payload.comment else None) db.add(report) db.commit() return CatchReportCreated(id=report.id, moderation_status=report.moderation_status.value) @app.post("/api/v1/catch-reports/{report_id}/screenshot", status_code=204, response_class=Response) def add_screenshot(report_id: UUID, db: Db, screenshot: UploadFile = File()) -> Response: report = db.get(CatchReport, report_id) if report is None or report.source_type != SourceType.user or report.moderation_status != ModerationStatus.pending: raise HTTPException(status_code=404, detail="pending catch report not found") if report.screenshot_key: raise HTTPException(status_code=409, detail="screenshot already uploaded") raw = screenshot.file.read(settings.screenshot_max_bytes + 1) try: report.screenshot_key = upload_screenshot(raw, filename=screenshot.filename, content_type=screenshot.content_type) except ScreenshotError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc db.commit() return Response(status_code=204) @app.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)) -> 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).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) for r in reports] @app.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.get(CatchReport, report_id) 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") previous = report.moderation_status report.moderation_status = ModerationStatus(payload.status) 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() return CatchReportCreated(id=report.id, moderation_status=report.moderation_status.value) @app.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)]) -> Response: report = db.get(CatchReport, report_id) 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") 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.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() return Response(status_code=204) def _check_rate_limit(client: str, db: Session) -> None: now = datetime.now(timezone.utc) cutoff = now - timedelta(minutes=10) client_hash = hmac.new(settings.rate_limit_secret.encode(), client.encode(), hashlib.sha256).hexdigest() if db.get_bind().dialect.name == "postgresql": lock_key = int(client_hash[:16], 16) & 0x7FFF_FFFF_FFFF_FFFF db.execute(text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": lock_key}) db.execute(delete(SubmissionAttempt).where(SubmissionAttempt.created_at < now - timedelta(days=1))) recent = db.scalar(select(func.count()).select_from(SubmissionAttempt).where(SubmissionAttempt.client_hash == client_hash, SubmissionAttempt.created_at >= cutoff)) or 0 if recent >= 5: db.commit() raise HTTPException(status_code=429, detail="too many submissions") db.add(SubmissionAttempt(client_hash=client_hash, created_at=now)) db.commit() def _aware(value: datetime) -> datetime: return value if value.tzinfo else value.replace(tzinfo=timezone.utc)