Adds visitor dashboard and related features

This commit is contained in:
2026-06-28 06:41:23 +00:00
parent a7cccbf170
commit 77d1c04d86
35 changed files with 1938 additions and 8 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+187
View File
@@ -0,0 +1,187 @@
"""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
]
+4
View File
@@ -14,6 +14,10 @@ class Settings(BaseSettings):
JWT_SECRET_KEY: str = "change-me-in-production"
JWT_EXPIRE_MINUTES: int = 1440 # 24 hours
# Visitor stats
LOGS_DIR: str = "/logs"
GEOLITE2_DB_PATH: str = "/app/geo/GeoLite2-City.mmdb"
model_config = {"env_prefix": "KMTN_"}
+317
View File
@@ -0,0 +1,317 @@
"""Nginx access log parser with IP geolocation enrichment."""
import asyncio
import json
import os
from datetime import date, datetime
from typing import Callable, Optional
import maxminddb
from app.config import settings
from app.database import async_session
from app.models import LogParseState, VisitorStatDaily, VisitorStatGeo
# ── Geo reader ──────────────────────────────────────────────
_geo_reader: Optional[maxminddb.Reader] = None
def get_geo_reader() -> maxminddb.Reader:
"""Return a singleton MaxMind GeoLite2 City reader."""
global _geo_reader
if _geo_reader is None:
db_path = settings.GEOLITE2_DB_PATH
if os.path.exists(db_path):
_geo_reader = maxminddb.open_database(db_path)
else:
raise FileNotFoundError(
f"GeoLite2 database not found at {db_path}. "
"Download from https://dev.maxmind.com/geoip/geolite2-free-geolocation-data "
"and place it at backend/geo/GeoLite2-City.mmdb"
)
return _geo_reader
# Country centroid fallback for free GeoLite2 (no city/coordinates).
# Approximate center coordinates per ISO country code — enough for map markers.
_COUNTRY_CENTROIDS: dict[str, tuple[float, float]] = {
"US": (39.8, -98.5),
"GB": (53.5, -1.5),
"DE": (51.2, 10.4),
"FR": (46.6, 2.3),
"JP": (36.2, 138.3),
"AU": (-25.3, 133.8),
"BR": (-14.2, -51.9),
"CA": (56.1, -106.3),
"IN": (20.6, 79.0),
"KR": (35.9, 127.8),
"NL": (52.1, 5.3),
"MX": (23.6, -102.5),
"IT": (41.9, 12.6),
"ES": (40.5, -3.7),
"SE": (62.0, 15.0),
"PL": (51.9, 19.1),
"AR": (-38.4, -63.6),
"ZA": (-30.6, 22.9),
"EG": (26.8, 30.8),
"TR": (39.0, 35.2),
"RU": (61.5, 105.3),
"SG": (1.35, 103.8),
"NZ": (-40.9, 174.9),
"CO": (4.6, -74.3),
"CL": (-35.7, -71.5),
"PE": (-9.2, -75.0),
"PH": (12.9, 121.8),
"ID": (-0.8, 113.9),
"TH": (15.9, 100.9),
"VN": (14.1, 108.3),
"MY": (4.2, 101.9),
"AE": (23.4, 53.8),
"SA": (23.9, 45.1),
"IL": (31.0, 34.9),
"NO": (60.5, 8.5),
"DK": (56.3, 9.5),
"FI": (64.1, 26.5),
"BE": (50.5, 4.5),
"AT": (47.5, 14.6),
"CH": (46.8, 8.2),
"PT": (39.4, -8.2),
"GR": (39.1, 21.8),
"CZ": (49.8, 15.5),
"RO": (45.9, 24.9),
"HU": (47.2, 19.5),
"UA": (48.4, 31.2),
"NG": (9.1, 8.7),
"KE": (-0.02, 37.9),
"GH": (7.9, -1.0),
"CN": (35.9, 104.2),
"PK": (30.4, 69.3),
"BD": (23.7, 90.4),
}
def lookup_geo(ip: str) -> dict:
"""Look up geolocation for an IP address.
Returns dict with country, country_code, city, latitude, longitude
or empty dict if lookup fails. Falls back to country centroids
when the free GeoLite2 DB lacks city/coordinate data.
"""
try:
reader = get_geo_reader()
result = reader.get(ip)
if result is None:
return {}
country = result.get("country", {})
city = result.get("city", {})
location = result.get("location", {})
country_code = country.get("iso_code")
lat = location.get("latitude")
lon = location.get("longitude")
# Free GeoLite2 City often lacks city/coordinates — use centroid fallback
if lat is None or lon is None:
centroid = _COUNTRY_CENTROIDS.get(country_code or "")
if centroid:
lat = lat if lat is not None else centroid[0]
lon = lon if lon is not None else centroid[1]
return {
"country": country.get("names", {}).get("en", country_code or ""),
"country_code": country_code,
"city": city.get("names", {}).get("en", ""),
"latitude": lat,
"longitude": lon,
}
except Exception:
return {}
# ── IP resolution ───────────────────────────────────────────
def resolve_client_ip(entry: dict) -> str:
"""Extract the real client IP from a log entry.
Prefers the leftmost X-Forwarded-For entry (production behind proxy),
falls back to remote_addr (dev).
"""
xff = entry.get("x_forwarded_for", "")
if xff and xff != "-":
# Leftmost entry is the original client
return xff.split(",")[0].strip()
return entry.get("remote_addr", "")
# ── Log parsing ─────────────────────────────────────────────
def parse_log_line(line: str) -> Optional[dict]:
"""Parse a single JSON log line. Returns dict or None if unparseable."""
line = line.strip()
if not line:
return None
try:
return json.loads(line)
except (json.JSONDecodeError, ValueError):
return None
def parse_log_file(filepath: str) -> dict:
"""Parse a single daily log file and return aggregated counts.
Returns a dict with:
- unique_ips: set of unique client IP addresses
- geo_counts: dict of (country_code, country, city, lat, lon) -> count
"""
unique_ips: set[str] = set()
geo_counts: dict[tuple, int] = {}
with open(filepath, "r") as f:
for line in f:
entry = parse_log_line(line)
if entry is None:
continue
ip = resolve_client_ip(entry)
if not ip:
continue
unique_ips.add(ip)
# Geo enrichment
geo = lookup_geo(ip)
if geo.get("country_code"):
geo_key = (
geo.get("country_code", ""),
geo.get("country", ""),
geo.get("city", ""),
geo.get("latitude"),
geo.get("longitude"),
)
geo_counts[geo_key] = geo_counts.get(geo_key, 0) + 1
return {
"unique_ips": unique_ips,
"geo_counts": geo_counts,
}
async def process_unparsed_logs(
progress_callback: Optional[Callable] = None,
) -> dict:
"""Scan /logs/ for unprocessed daily files, parse them, store aggregates.
Only processes files strictly older than today (file must be complete).
Advances the LogParseState marker after each successful file.
Returns a summary dict with files_processed and rows_inserted.
"""
logs_dir = settings.LOGS_DIR
if not os.path.isdir(logs_dir):
return {"files_processed": 0, "error": f"Logs directory not found: {logs_dir}"}
# Get current parse state
async with async_session() as session:
from sqlalchemy import select
result = await session.execute(
select(LogParseState).where(LogParseState.key == "default")
)
state = result.scalar_one_or_none()
if state is None:
# Initialize with a far-past date so all files get processed
state = LogParseState(key="default", last_parsed_date=date(2020, 1, 1))
session.add(state)
await session.commit()
last_parsed = state.last_parsed_date
# Discover daily log files
import glob
pattern = os.path.join(logs_dir, "access-*.log")
log_files = sorted(glob.glob(pattern))
files_processed = 0
rows_inserted = 0
today = date.today()
for filepath in log_files:
# Extract date from filename
basename = os.path.basename(filepath)
try:
date_str = basename.replace("access-", "").replace(".log", "")
file_date = date.fromisoformat(date_str)
except ValueError:
continue
# Only process files strictly before today (file must be complete)
if file_date >= today:
continue
# Only process files after the last parsed date
if file_date <= last_parsed:
continue
# Parse this file
try:
stats = parse_log_file(filepath)
except Exception as e:
print(f" ERROR parsing {filepath}: {e}")
continue
# Store aggregates in database
async with async_session() as session:
from sqlalchemy import select
# Upsert daily stats
result = await session.execute(
select(VisitorStatDaily).where(VisitorStatDaily.date == file_date)
)
daily = result.scalar_one_or_none()
if daily is None:
daily = VisitorStatDaily(date=file_date)
session.add(daily)
daily.unique_visitors = len(stats["unique_ips"])
# Geo stats
for geo_key, count in stats["geo_counts"].items():
geo_row = VisitorStatGeo(
date=file_date,
country_code=geo_key[0],
country=geo_key[1],
city=geo_key[2],
latitude=geo_key[3],
longitude=geo_key[4],
visitor_count=count,
)
session.add(geo_row)
rows_inserted += 1
# Advance parse state
state = (
await session.execute(
select(LogParseState).where(LogParseState.key == "default")
)
).scalar_one()
state.last_parsed_date = file_date
state.updated_at = datetime.utcnow()
await session.commit()
files_processed += 1
# Report progress via callback (for SSE streaming)
if progress_callback:
cb_result = progress_callback(
file_date.isoformat(), files_processed, len(log_files)
)
# Support both sync and async callbacks
if cb_result is not None and asyncio.iscoroutine(cb_result):
await cb_result
return {"files_processed": files_processed, "rows_inserted": rows_inserted}
+25 -1
View File
@@ -1,4 +1,6 @@
from contextlib import asynccontextmanager
import asyncio
import os
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
@@ -8,7 +10,7 @@ from app.config import settings
from app.database import async_session, engine
from app.models import Base, Show, StationConfig, HistoryEntry, TeamMember, CommunityHighlight, Underwriter
from app.user_models import User
from app.api import events, auth, station_config, history, team, community, shows, storage, underwriters
from app.api import events, auth, station_config, history, team, community, shows, storage, underwriters, stats
def _migrate_add_missing_columns(sync_conn):
@@ -117,6 +119,27 @@ async def lifespan(app: FastAPI):
await session.commit()
print(f" → Bootstrap admin created: {settings.ADMIN_USERNAME}")
# Check GeoLite2 database
geo_db = settings.GEOLITE2_DB_PATH
if os.path.exists(geo_db):
print(f" ✓ GeoLite2 database loaded from {geo_db}")
else:
print(f" WARNING: GeoLite2 database not found at {geo_db} — geo lookups will be disabled")
# Auto-parse any unprocessed logs on startup (background task)
async def _startup_parse():
try:
from app.log_parser import process_unparsed_logs
result = await process_unparsed_logs()
if result.get("files_processed"):
print(f" ✓ Parsed {result['files_processed']} log files ({result['rows_inserted']} geo rows)")
else:
print(" No unprocessed log files to parse")
except Exception as e:
print(f" ERROR during startup log parse: {e}")
asyncio.create_task(_startup_parse())
yield
@@ -145,6 +168,7 @@ app.include_router(community.router, prefix="/api/community", tags=["community"]
app.include_router(shows.router, prefix="/api/shows", tags=["shows"])
app.include_router(storage.router, prefix="/api/storage", tags=["storage"])
app.include_router(underwriters.router, prefix="/api/underwriters", tags=["underwriters"])
app.include_router(stats.router, prefix="/api/stats", tags=["stats"])
# Dev-only admin router (disabled in production)
if settings.ENV != "production":
+46
View File
@@ -3,6 +3,7 @@ from datetime import datetime
from sqlalchemy import (
Column,
DateTime,
Float,
Integer,
LargeBinary,
String,
@@ -11,6 +12,7 @@ from sqlalchemy import (
Date,
ForeignKey,
UniqueConstraint,
Index,
)
from sqlalchemy.orm import DeclarativeBase, relationship
@@ -159,3 +161,47 @@ class StorageBlob(Base):
size = Column(Integer, nullable=False)
data = Column(LargeBinary, nullable=False)
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
# ── Visitor Stats ──────────────────────────────────────────────
class VisitorStatDaily(Base):
"""Daily aggregated visitor statistics."""
__tablename__ = "visitor_stats_daily"
id = Column(Integer, primary_key=True, autoincrement=True)
date = Column(Date, nullable=False, unique=True)
unique_visitors = Column(Integer, nullable=False, default=0)
__table_args__ = (
Index("idx_visitor_stats_date", "date"),
)
class VisitorStatGeo(Base):
"""Geographic breakdown of visitors per day."""
__tablename__ = "visitor_stats_geo"
id = Column(Integer, primary_key=True, autoincrement=True)
date = Column(Date, nullable=False)
country = Column(String(100), nullable=True)
country_code = Column(String(10), nullable=True)
city = Column(String(200), nullable=True)
latitude = Column(Float, nullable=True)
longitude = Column(Float, nullable=True)
visitor_count = Column(Integer, nullable=False, default=0)
__table_args__ = (
Index("idx_visitor_stats_geo_date", "date"),
Index("idx_visitor_stats_geo_country", "country_code"),
)
class LogParseState(Base):
"""Tracks the last successfully parsed log date."""
__tablename__ = "log_parse_state"
id = Column(Integer, primary_key=True, autoincrement=True)
key = Column(String(20), unique=True, nullable=False, default="default")
last_parsed_date = Column(Date, nullable=False)
updated_at = Column(DateTime, nullable=False, default=datetime.utcnow)
+25 -1
View File
@@ -1,7 +1,7 @@
from datetime import date as _date
from typing import Optional
from pydantic import BaseModel, computed_field, field_validator
from pydantic import BaseModel, computed_field
# ── Shared ───────────────────────────────────────────────
@@ -284,3 +284,27 @@ class ShowUpdate(BaseModel):
display_order: Optional[int] = None
active: Optional[bool] = None
schedules: Optional[list[ShowScheduleCreate]] = None
# ── Stats ────────────────────────────────────────────────────────
class StatsSummaryResponse(BaseModel):
total_unique_visitors: int
class TimeSeriesPoint(BaseModel):
date: str
unique_visitors: int
class GeoStatResponse(BaseModel):
country: str
country_code: str
latitude: Optional[float] = None
longitude: Optional[float] = None
visitors: int
class ParseResultResponse(BaseModel):
files_processed: int
rows_inserted: int