637 lines
36 KiB
Python
637 lines
36 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from datetime import datetime, timedelta, timezone
|
|
from ipaddress import IPv4Address, IPv6Address, IPv4Network, IPv6Network
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import secrets
|
|
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.exc import IntegrityError
|
|
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, suggest_aliases
|
|
from .importer import ImportAlreadyRunning, ImportSourceError, import_records, normalize
|
|
from .logging_config import configure_logging
|
|
from .models import Bait, BaitKind, CatchReport, CommunityImportRun, DataSource, ExternalObservation, Fish, ModerationEvent, ModerationStatus, OfficialRecordImport, SourceType, Spot, SubmissionAttempt, Waterbody
|
|
from .readiness import readiness_report
|
|
from .public_cache import public_cache
|
|
from .schemas import ActivityOut, AdminCatchReportOut, BaitOut, CatchOut, CatchReportAccepted, CatchReportCreate, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, FishOut, ImportRunOut, ImportRunPublicOut, ModerationUpdate, OfficialRecordOut, PaginatedActivityOut, PaginatedOfficialRecordOut, PublicObservationOut, SourceStatusOut, 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=settings.cors_origins,
|
|
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
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
|
response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
|
|
if request.url.path.startswith("/api/v1/admin/") or request.url.path == "/api/v1/catch-reports":
|
|
response.headers["Cache-Control"] = "no-store"
|
|
if settings.deployment_environment == "production":
|
|
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
|
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,
|
|
community_import_interval_seconds=settings.community_import_interval_seconds,
|
|
)
|
|
return JSONResponse(
|
|
status_code=200 if is_ready else 503,
|
|
content={"status": "ready" if is_ready else "not_ready", "version": settings.app_version, "revision": settings.app_revision, "components": components},
|
|
)
|
|
|
|
|
|
@app.get("/api/v1/fishes", response_model=list[FishOut])
|
|
def fishes(db: Db, limit: int = Query(200, ge=1, le=500), offset: int = Query(0, ge=0)) -> list[Fish]:
|
|
return list(db.scalars(select(Fish).order_by(Fish.name_ru, Fish.id).offset(offset).limit(limit)))
|
|
|
|
|
|
@app.get("/api/v1/waterbodies", response_model=list[WaterbodyOut])
|
|
def waterbodies(db: Db, limit: int = Query(200, ge=1, le=500), offset: int = Query(0, ge=0)) -> list[Waterbody]:
|
|
return list(db.scalars(select(Waterbody).order_by(Waterbody.name_ru, Waterbody.id).offset(offset).limit(limit)))
|
|
|
|
|
|
@app.get("/api/v1/baits", response_model=list[BaitOut])
|
|
def baits(db: Db, limit: int = Query(200, ge=1, le=500), offset: int = Query(0, ge=0)) -> list[Bait]:
|
|
return list(db.scalars(select(Bait).order_by(Bait.name, Bait.id).offset(offset).limit(limit)))
|
|
|
|
|
|
@app.get("/api/v1/public-spot-pages")
|
|
def public_spot_pages(db: Db, limit: int = Query(500, ge=1, le=500), offset: int = Query(0, ge=0)) -> list[str]:
|
|
rows = db.execute(select(Waterbody.slug, Spot.x, Spot.y, Fish.slug)
|
|
.select_from(CatchReport).join(Spot, CatchReport.spot_id == Spot.id)
|
|
.join(Waterbody, Spot.waterbody_id == Waterbody.id).join(Fish, CatchReport.fish_id == Fish.id)
|
|
.where(CatchReport.moderation_status == ModerationStatus.approved, CatchReport.deleted_at.is_(None))
|
|
.distinct().order_by(Waterbody.slug, Spot.x, Spot.y, Fish.slug).offset(offset).limit(limit))
|
|
return [path for water, x, y, fish in rows for path in
|
|
(f"/spots/{water}-{x}x{y}", f"/waterbodies/{water}/{fish}")]
|
|
|
|
|
|
@app.get("/api/v1/activity", response_model=PaginatedActivityOut)
|
|
def activity(
|
|
db: Db, response: Response, 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),
|
|
) -> PaginatedActivityOut:
|
|
if hours not in {6, 12, 24, 72}:
|
|
raise HTTPException(status_code=422, detail="hours must be one of: 6, 12, 24, 72")
|
|
response.headers["Cache-Control"] = "no-store"
|
|
generation = public_cache.generation()
|
|
cache_key = ("activity", hours, waterbody, fish, method, sort, limit, offset)
|
|
cached = public_cache.get(cache_key, settings.public_cache_seconds)
|
|
if cached is not None:
|
|
response.headers["X-Cache"] = "HIT"
|
|
return cached
|
|
rows = activity_rows(db, hours=hours, waterbody=waterbody, fish=fish, method=method)
|
|
total = len(rows)
|
|
keys = {
|
|
"activity": lambda r: (r.activity_score, r.confidence_score, r.last_confirmed_at, str(r.spot_id)),
|
|
"confidence": lambda r: (r.confidence_score, r.activity_score, r.last_confirmed_at, str(r.spot_id)),
|
|
"freshness": lambda r: (r.last_confirmed_at, r.activity_score, r.confidence_score, str(r.spot_id)),
|
|
}
|
|
rows.sort(key=keys[sort], reverse=True)
|
|
page = rows[offset:offset + limit]
|
|
response.headers["X-Cache"] = "MISS"
|
|
return public_cache.set(cache_key, PaginatedActivityOut(items=page, total=total, limit=limit, offset=offset), generation=generation)
|
|
|
|
|
|
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/resolve", response_model=SpotOut)
|
|
def resolve_spot(
|
|
db: Db, waterbody: str, x: int = Query(ge=-10_000, le=10_000),
|
|
y: int = Query(ge=-10_000, le=10_000),
|
|
) -> SpotOut:
|
|
spot = db.scalar(select(Spot).options(joinedload(Spot.waterbody)).join(Spot.waterbody).where(
|
|
Waterbody.slug == waterbody, Spot.x == x, Spot.y == y,
|
|
))
|
|
if spot is None:
|
|
raise HTTPException(status_code=404, detail="spot not found")
|
|
return spot_detail(spot.id, db)
|
|
|
|
|
|
@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(), CatchReport.id.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, source_system=_report_source(r), source_url=r.source_url) for r in reports]
|
|
|
|
|
|
@app.get("/api/v1/spots/{spot_id}/timeline")
|
|
def spot_timeline(spot_id: UUID, db: Db) -> list[dict]:
|
|
_spot_or_404(db, spot_id)
|
|
now = datetime.now(timezone.utc)
|
|
buckets = []
|
|
for index in range(6):
|
|
start = now - timedelta(hours=(6 - index) * 12)
|
|
end = start + timedelta(hours=12)
|
|
count = db.scalar(select(func.count()).select_from(CatchReport).where(
|
|
CatchReport.spot_id == spot_id,
|
|
CatchReport.moderation_status == ModerationStatus.approved,
|
|
CatchReport.deleted_at.is_(None),
|
|
CatchReport.reported_at >= start, CatchReport.reported_at < end,
|
|
)) or 0
|
|
buckets.append({"start": start.isoformat(), "end": end.isoformat(), "count": count})
|
|
return buckets
|
|
|
|
|
|
def _report_source(report: CatchReport) -> str:
|
|
provenance = (report.raw_payload or {}).get("provenance", {})
|
|
if isinstance(provenance, dict) and provenance.get("source_system"):
|
|
return str(provenance["source_system"])
|
|
if report.source_type == SourceType.official_record:
|
|
return "rf4-official"
|
|
if report.source_type == SourceType.user:
|
|
return "players"
|
|
return "manual-import"
|
|
|
|
|
|
@app.get("/api/v1/community-observations", response_model=list[PublicObservationOut])
|
|
def community_observations(
|
|
db: Db, limit: int = Query(12, ge=1, le=50), offset: int = Query(0, ge=0),
|
|
waterbody: str | None = None, fish: str | None = None,
|
|
) -> list[PublicObservationOut]:
|
|
query = select(ExternalObservation).join(ExternalObservation.source).options(joinedload(ExternalObservation.source)).where(
|
|
ExternalObservation.catch_report_id.is_(None),
|
|
ExternalObservation.status != "rejected",
|
|
DataSource.enabled.is_(True),
|
|
)
|
|
if waterbody:
|
|
query = query.join(ExternalObservation.waterbody).where(Waterbody.slug == waterbody)
|
|
if fish:
|
|
query = query.join(ExternalObservation.fish).where(Fish.slug == fish)
|
|
items = list(db.scalars(query.order_by(ExternalObservation.last_seen_at.desc(), ExternalObservation.id.desc()).offset(offset).limit(limit)))
|
|
result: list[PublicObservationOut] = []
|
|
for item in items:
|
|
missing = []
|
|
if item.x is None or item.y is None:
|
|
missing.append("координаты")
|
|
if item.weight_g is None:
|
|
missing.append("вес")
|
|
result.append(PublicObservationOut(
|
|
id=item.id, source_system=item.source_system, source_name=item.source.name,
|
|
source_url=item.source_url, fish_name=item.fish_name, waterbody_name=item.waterbody_name,
|
|
x=item.x, y=item.y, weight_g=item.weight_g, last_seen_at=item.last_seen_at,
|
|
missing_fields=missing, quality="incomplete" if missing else "unverified",
|
|
))
|
|
return result
|
|
|
|
|
|
@app.get("/api/v1/source-status", response_model=list[SourceStatusOut])
|
|
def source_status(db: Db) -> list[SourceStatusOut]:
|
|
now = datetime.now(timezone.utc)
|
|
result = []
|
|
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)
|
|
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(SourceStatusOut(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,
|
|
observations=db.scalar(select(func.count()).select_from(ExternalObservation).where(ExternalObservation.source_system == source.key)) or 0))
|
|
return result
|
|
|
|
|
|
@app.get("/api/v1/records", response_model=PaginatedOfficialRecordOut)
|
|
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),
|
|
) -> PaginatedOfficialRecordOut:
|
|
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)
|
|
if category:
|
|
query = query.where(CatchReport.raw_payload["category"].as_string() == category)
|
|
# Count from the exact same filtered query (before pagination).
|
|
count_query = query.with_only_columns(func.count(CatchReport.id), maintain_column_froms=True).order_by(None)
|
|
total = db.scalar(count_query) or 0
|
|
items = list(db.scalars(query.order_by(CatchReport.caught_at.desc(), CatchReport.weight_g.desc(), CatchReport.id.desc()).offset(offset).limit(limit)))
|
|
return PaginatedOfficialRecordOut(
|
|
items=[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],
|
|
total=total, limit=limit, offset=offset,
|
|
)
|
|
|
|
|
|
def _admin(authorization: Annotated[str | None, Header()] = None) -> str:
|
|
expected = f"Bearer {settings.admin_token}"
|
|
if not authorization or not hmac.compare_digest(authorization, expected):
|
|
raise HTTPException(status_code=401, detail="invalid admin token", headers={"WWW-Authenticate": "Bearer"})
|
|
return "admin"
|
|
|
|
|
|
@app.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"})
|
|
|
|
|
|
@app.get("/api/v1/imports", response_model=list[ImportRunPublicOut])
|
|
def imports(db: Db, limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0)) -> list[OfficialRecordImport]:
|
|
return list(db.scalars(select(OfficialRecordImport).order_by(OfficialRecordImport.started_at.desc(), OfficialRecordImport.id.desc()).offset(offset).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 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:
|
|
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: Literal["staged", "mapped", "ready", "published", "rejected", "review"] | 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 == "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)
|
|
items = db.scalars(query.order_by(ExternalObservation.last_seen_at.desc(), ExternalObservation.id.desc()).offset(offset).limit(limit))
|
|
return [_external_out(item) for item in items]
|
|
|
|
|
|
@app.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,
|
|
)
|
|
|
|
|
|
@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
|
|
public_cache.invalidate()
|
|
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=CatchReportAccepted, status_code=201)
|
|
def create_catch_report(
|
|
payload: CatchReportCreate, request: Request, db: Db,
|
|
idempotency_key: Annotated[str | None, Header()] = None,
|
|
) -> CatchReportAccepted:
|
|
if payload.website:
|
|
raise HTTPException(status_code=400, detail="invalid submission")
|
|
payload_hash = hashlib.sha256(json.dumps(payload.model_dump(mode="json"), sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
# A05: Server-side idempotency — check BEFORE rate limit to avoid polluting table
|
|
if idempotency_key:
|
|
key_hash = hmac.new(settings.rate_limit_secret.encode(), idempotency_key.encode(), hashlib.sha256).hexdigest()
|
|
cutoff = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
# Force refresh from database to see committed data from previous requests
|
|
db.expire_all()
|
|
existing = db.scalar(
|
|
select(SubmissionAttempt).where(
|
|
SubmissionAttempt.idempotency_key == key_hash,
|
|
SubmissionAttempt.created_at >= cutoff,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
# Return 200 with idempotent flag — client can retry safely
|
|
logger.info("idempotent hit", extra={"idempotency_key": idempotency_key[:8]})
|
|
report = existing.catch_report
|
|
if existing.payload_hash and not hmac.compare_digest(existing.payload_hash, payload_hash):
|
|
raise HTTPException(status_code=409, detail="Idempotency-Key was already used with different payload")
|
|
if report is None:
|
|
raise HTTPException(status_code=409, detail="idempotency record is incomplete; retry with a new key")
|
|
# Re-derive the one-time upload token from the idempotency key;
|
|
# only its hash is persisted, so the secret is never stored.
|
|
replay_token = hmac.new(settings.rate_limit_secret.encode(), (key_hash + ":upload").encode(), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(hashlib.sha256(replay_token.encode()).hexdigest(), report.screenshot_upload_token_hash or ""):
|
|
raise HTTPException(status_code=409, detail="idempotency record token mismatch; retry with a new key")
|
|
return JSONResponse(status_code=200, content={"id": str(report.id), "moderation_status": report.moderation_status.value, "screenshot_upload_token": replay_token, "idempotent": True})
|
|
logger.info("idempotency check miss", extra={"idempotency_key": idempotency_key[:8]})
|
|
_check_rate_limit(request, 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)
|
|
upload_token = (hmac.new(settings.rate_limit_secret.encode(), (key_hash + ":upload").encode(), hashlib.sha256).hexdigest()
|
|
if idempotency_key else secrets.token_urlsafe(32))
|
|
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, screenshot_upload_token_hash=hashlib.sha256(upload_token.encode()).hexdigest())
|
|
db.add(report)
|
|
db.commit()
|
|
# Store idempotency key if provided
|
|
if idempotency_key:
|
|
key_hash = hmac.new(settings.rate_limit_secret.encode(), idempotency_key.encode(), hashlib.sha256).hexdigest()
|
|
db.add(SubmissionAttempt(client_hash="", idempotency_key=key_hash, catch_report_id=report.id, payload_hash=payload_hash, created_at=datetime.now(timezone.utc)))
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
# Another request won the same idempotency key race.
|
|
db.rollback()
|
|
winner = db.scalar(select(SubmissionAttempt).where(SubmissionAttempt.idempotency_key == key_hash))
|
|
if winner and winner.catch_report:
|
|
replay_token = hmac.new(settings.rate_limit_secret.encode(), (key_hash + ":upload").encode(), hashlib.sha256).hexdigest()
|
|
return JSONResponse(status_code=200, content={"id": str(winner.catch_report.id), "moderation_status": winner.catch_report.moderation_status.value, "screenshot_upload_token": replay_token, "idempotent": True})
|
|
raise
|
|
logger.info("idempotency key stored", extra={"idempotency_key": idempotency_key[:8]})
|
|
return CatchReportAccepted(id=report.id, moderation_status=report.moderation_status.value, screenshot_upload_token=upload_token, idempotent=False)
|
|
|
|
|
|
@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(),
|
|
upload_token: Annotated[str | None, Header(alias="X-Upload-Token")] = None,
|
|
) -> 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")
|
|
supplied_hash = hashlib.sha256((upload_token or "").encode()).hexdigest()
|
|
if not report.screenshot_upload_token_hash or not hmac.compare_digest(report.screenshot_upload_token_hash, supplied_hash):
|
|
raise HTTPException(status_code=401, detail="invalid screenshot upload token")
|
|
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
|
|
report.screenshot_upload_token_hash = None
|
|
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), 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) 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()
|
|
public_cache.invalidate()
|
|
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()
|
|
public_cache.invalidate()
|
|
return Response(status_code=204)
|
|
|
|
|
|
def _is_trusted_proxy(address: str, trusted_cidrs: list[str]) -> bool:
|
|
"""Check if address is in trusted proxy CIDRs."""
|
|
try:
|
|
addr = IPv4Address(address) if ":" not in address else IPv6Address(address)
|
|
except ValueError:
|
|
return False
|
|
for cidr in trusted_cidrs:
|
|
try:
|
|
network = IPv4Network(cidr) if ":" not in cidr else IPv6Network(cidr)
|
|
if addr in network:
|
|
return True
|
|
except ValueError:
|
|
continue
|
|
return False
|
|
|
|
|
|
def _check_rate_limit(request: Request, db: Session) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
cutoff = now - timedelta(minutes=10)
|
|
# Extract real client IP from forwarded headers
|
|
client = request.client.host if request.client else "unknown"
|
|
forwarded = request.headers.get("x-forwarded-for")
|
|
# Only trust X-Forwarded-For if connection came from a trusted proxy
|
|
if forwarded and request.client and _is_trusted_proxy(request.client.host, settings.trusted_proxy_cidrs):
|
|
client = forwarded.split(",")[0].strip()
|
|
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)
|