refactor: extract activity and spot router
This commit is contained in:
+5
-105
@@ -1,6 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from ipaddress import IPv4Address, IPv6Address, IPv4Network, IPv6Network
|
||||
import hashlib
|
||||
@@ -21,7 +20,6 @@ from sqlalchemy import delete, func, select, text
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
|
||||
from .activity import activity_rows
|
||||
from .config import settings
|
||||
from .dependencies import Db
|
||||
from .community_review import ExternalReviewError, map_observation, publish_observation, reject_observation, suggest_aliases
|
||||
@@ -29,9 +27,11 @@ from .importer import ImportAlreadyRunning, ImportSourceError, import_records, n
|
||||
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 .routers.activity import router as activity_router
|
||||
from .routers.catalog import router as catalog_router
|
||||
from .time_utils import aware
|
||||
from .public_cache import public_cache
|
||||
from .schemas import ActivityOut, AdminCatchReportOut, CatchOut, CatchReportAccepted, CatchReportCreate, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, ImportRunOut, ImportRunPublicOut, ModerationUpdate, OfficialRecordOut, PaginatedActivityOut, PaginatedOfficialRecordOut, PublicObservationOut, SourceStatusOut, SpotOut
|
||||
from .schemas import ActivityOut, AdminCatchReportOut, CatchReportAccepted, CatchReportCreate, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, ImportRunOut, ImportRunPublicOut, ModerationUpdate, OfficialRecordOut, PaginatedOfficialRecordOut, PublicObservationOut, SourceStatusOut
|
||||
from .storage import ScreenshotError, client as storage_client, delete_screenshot, signed_screenshot_url, upload_screenshot
|
||||
|
||||
|
||||
@@ -99,103 +99,7 @@ def ready(db: Db) -> JSONResponse:
|
||||
|
||||
|
||||
app.include_router(catalog_router)
|
||||
|
||||
|
||||
@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.include_router(activity_router)
|
||||
|
||||
|
||||
@app.get("/api/v1/community-observations", response_model=list[PublicObservationOut])
|
||||
@@ -243,7 +147,7 @@ def source_status(db: Db) -> list[SourceStatusOut]:
|
||||
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):
|
||||
elif aware(latest.started_at) < now - timedelta(seconds=settings.community_import_interval_seconds * 2):
|
||||
state = "stale"
|
||||
else:
|
||||
state = "healthy"
|
||||
@@ -609,7 +513,3 @@ def _check_rate_limit(request: Request, db: Session) -> None:
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user