"""Stats API: trigger log parsing and query aggregated visitor data.""" import asyncio from datetime import date, timedelta from typing import AsyncGenerator, Optional from fastapi import APIRouter, Depends, Query, HTTPException, status from fastapi.responses import StreamingResponse from sqlalchemy import desc, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.auth import decode_token, get_current_admin_user from app.database import get_session from app.log_parser import process_unparsed_logs from app.models import LogParseState, VisitorStatDaily, VisitorStatGeo from app.user_models import User router = APIRouter() # ── Helpers ──────────────────────────────────────────────────── def _default_date_range() -> tuple[date, date]: """Return (start_date, end_date) for last 30 days.""" end = date.today() - timedelta(days=1) # today isn't parsed yet start = end - timedelta(days=29) return start, end async def _verify_admin_from_token(token: str) -> User: """Verify admin from a raw JWT token string (for SSE query-param auth).""" from app.database import async_session from sqlalchemy import select as sa_select payload = decode_token(token) user_id = int(payload["sub"]) if not payload.get("is_admin"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") async with async_session() as session: result = await session.execute(sa_select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found") return user # ── SSE parse endpoint ─────────────────────────────────────── @router.get("/parse") async def trigger_parse(token: Optional[str] = Query(None)): """Trigger log file parsing. Streams progress via Server-Sent Events. Accepts JWT via `token` query parameter (EventSource can't send headers). """ # Authenticate — SSE can't send Authorization headers, so use query param if token: await _verify_admin_from_token(token) else: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token required") async def event_generator() -> AsyncGenerator[str, None]: # Collect progress events, then yield them events: list[tuple[str, int, int]] = [] result = await process_unparsed_logs(lambda fd, c, t: events.append((fd, c, t))) for file_date, current, total in events: yield f'event: progress\ndata: {{"file": "{file_date}", "current": {current}, "total": {total}}}\n\n' yield f'event: done\ndata: {{"files_processed": {result["files_processed"]}, "rows_inserted": {result["rows_inserted"]}}}\n\n' return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Disable nginx buffering for SSE }, ) # ── Summary ──────────────────────────────────────────────────── @router.get("/summary") async def get_summary( start_date: Optional[date] = Query(None), end_date: Optional[date] = Query(None), session: AsyncSession = Depends(get_session), _: User = Depends(get_current_admin_user), ): """Get total unique visitors for a date range.""" if start_date is None or end_date is None: start_date, end_date = _default_date_range() result = await session.execute( select(func.coalesce(func.sum(VisitorStatDaily.unique_visitors), 0)).where( VisitorStatDaily.date >= start_date, VisitorStatDaily.date <= end_date, ) ) total_unique = result.scalar() return {"total_unique_visitors": total_unique} # ── Time series ──────────────────────────────────────────────── @router.get("/timeseries") async def get_timeseries( start_date: Optional[date] = Query(None), end_date: Optional[date] = Query(None), session: AsyncSession = Depends(get_session), _: User = Depends(get_current_admin_user), ): """Get daily unique visitor counts for a date range.""" if start_date is None or end_date is None: start_date, end_date = _default_date_range() stmt = ( select( VisitorStatDaily.date, VisitorStatDaily.unique_visitors, ) .where( VisitorStatDaily.date >= start_date, VisitorStatDaily.date <= end_date, ) .order_by(VisitorStatDaily.date) ) result = await session.execute(stmt) rows = result.all() return [ {"date": row.date.isoformat(), "unique_visitors": row.unique_visitors} for row in rows ] # ── Geo ──────────────────────────────────────────────────────── @router.get("/geo") async def get_geo_stats( start_date: Optional[date] = Query(None), end_date: Optional[date] = Query(None), session: AsyncSession = Depends(get_session), _: User = Depends(get_current_admin_user), ): """Get geographic distribution of visitors for a date range. Aggregates by country (country, country_code, lat, lon), summing visitor_count. Returns one row per country/region for the map to render. """ if start_date is None or end_date is None: start_date, end_date = _default_date_range() stmt = ( select( VisitorStatGeo.country, VisitorStatGeo.country_code, VisitorStatGeo.latitude, VisitorStatGeo.longitude, func.sum(VisitorStatGeo.visitor_count).label("total_visitors"), ) .where( VisitorStatGeo.date >= start_date, VisitorStatGeo.date <= end_date, ) .group_by( VisitorStatGeo.country, VisitorStatGeo.country_code, VisitorStatGeo.latitude, VisitorStatGeo.longitude, ) .order_by(desc(func.sum(VisitorStatGeo.visitor_count))) ) result = await session.execute(stmt) rows = result.all() return [ { "country": row.country or "Unknown", "country_code": row.country_code or "", "latitude": row.latitude, "longitude": row.longitude, "visitors": row.total_visitors, } for row in rows ]