from __future__ import annotations import json import os from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from fastapi import FastAPI, Query from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from .updater import ensure_monthly_refresh, load_events, query_events ROOT = Path(__file__).resolve().parents[1] DATA_DIR = ROOT / "data" STATIC_DIR = ROOT / "static" app = FastAPI(title="AI Timeline", version="0.1.0") # Serve the SPA app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") @app.get("/", response_class=HTMLResponse) def index() -> HTMLResponse: return HTMLResponse((STATIC_DIR / "index.html").read_text(encoding="utf-8")) @app.get("/api/events") def api_events( q: Optional[str] = Query(default=None, description="Full-text search (title/description/org)."), category: Optional[str] = Query(default=None, description="open-weights | api | all"), org: Optional[str] = Query(default=None, description="Filter by org/author (HF)."), date_from: Optional[str] = Query(default=None, description="YYYY-MM-DD"), date_to: Optional[str] = Query(default=None, description="YYYY-MM-DD"), ) -> JSONResponse: """ Returns events for the timeline. Refresh logic: on every request we *cheap-check* whether the month changed since last refresh. If yes, we refresh the Hugging Face portion (open-weight models) and merge into events.json. """ ensure_monthly_refresh(data_dir=DATA_DIR) payload = load_events(DATA_DIR) events = query_events( payload["events"], q=q, category=category, org=org, date_from=date_from, date_to=date_to, ) return JSONResponse( { "as_of": datetime.now(timezone.utc).isoformat(), "count": len(events), "events": events, "meta": { "tracked_orgs": payload.get("tracked_orgs", []), "last_refresh_month": payload.get("last_refresh_month"), "last_refresh_at": payload.get("last_refresh_at"), }, } )