fundi-scraper-api/api/main.py

"""REST API over the scraped Fundbureau events.

Datasource is the JSON file produced by ``scraper/scraper.py``
(``fundi-scraped-output.json`` by default; override with the DATA_FILE env var).

Run with:
    uvicorn main:app --reload    # from inside the api/ folder
"""

import datetime
import os
from collections import defaultdict
from typing import List, Literal, Optional

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import PlainTextResponse

from calendar_feed import build_ics
from datasource import EventStore, iter_artist_names, parse_event_date, sort_date_strings, sort_events
from models import ArtistSchedule, Event, EventList
from rate_limit import Cooldown
from scrape_schedule import discover_cron_expr, get_schedule_info

SortOrder = Literal["asc", "desc", "az", "za"]
_SORT_DESCRIPTION = (
    "Sort order: asc/desc by event date, az/za alphabetically by event name"
)

app = FastAPI(
    title="Fundi Scraper API",
    description="REST API over the scraped upcoming/past events at the Fundbureau Hamburg.",
    version="1.0.0",
)

store = EventStore()

_RELOAD_MIN_INTERVAL_SECONDS = float(os.environ.get("RELOAD_MIN_INTERVAL_SECONDS", "10"))
_reload_cooldown = Cooldown(_RELOAD_MIN_INTERVAL_SECONDS)


def _matches(
    event: Event,
    free: Optional[bool],
    name: Optional[str],
    artist: Optional[str],
    event_date: Optional[str],
    upcoming: Optional[bool],
    today: datetime.date,
) -> bool:
    if free is not None and event.event_free is not free:
        return False
    if name is not None and name.lower() not in event.event_name.lower():
        return False
    if artist is not None:
        needle = artist.lower()
        if not any(needle in n.lower() for n in iter_artist_names(event)):
            return False
    if event_date is not None and event.event_date != event_date:
        return False
    if upcoming is not None:
        parsed = parse_event_date(event.event_date)
        if parsed is None:
            return False
        is_upcoming = parsed >= today
        if is_upcoming is not upcoming:
            return False
    return True


@app.get("/health")
def health():
    scrape_cron = os.environ.get("SCRAPE_CRON") or discover_cron_expr(
        "scraper.py", archive=False
    )
    archive_scrape_cron = (
        os.environ.get("ARCHIVE_SCRAPE_CRON")
        or discover_cron_expr("scraper.py", archive=True)
        or scrape_cron
    )
    return {
        "status": "ok",
        "events_loaded": len(store.all()),
        "data_file": str(store.path),
        "current_events": store.current_count,
        "archive_file": str(store.archive_path),
        "archive_file_found": store.archive_path.exists(),
        "archive_events": store.archive_count,
        "scrape": get_schedule_info(store.path, scrape_cron),
        "archive_scrape": get_schedule_info(store.archive_path, archive_scrape_cron),
    }


@app.post("/reload")
def reload():
    """Re-read the data file(s) from disk (e.g. after a fresh scrape).

    Rate-limited to one call per RELOAD_MIN_INTERVAL_SECONDS (default 10s)
    to stop it being hammered - a fresh scrape doesn't land any more often
    than that anyway.
    """
    remaining = _reload_cooldown.try_acquire()
    if remaining is not None:
        raise HTTPException(
            status_code=429,
            detail=f"Reload rate-limited; try again in {remaining:.1f}s",
            headers={"Retry-After": str(int(remaining) + 1)},
        )
    try:
        count = store.reload()
    except FileNotFoundError as exc:
        raise HTTPException(status_code=500, detail=str(exc))
    return {
        "reloaded": True,
        "events_loaded": count,
        "current_events": store.current_count,
        "archive_events": store.archive_count,
    }


@app.get("/events", response_model=EventList)
def list_events(
    free: Optional[bool] = Query(None, description="Filter by free admission"),
    name: Optional[str] = Query(None, description="Case-insensitive substring match on event name"),
    artist: Optional[str] = Query(None, description="Case-insensitive substring match on any artist name"),
    date: Optional[str] = Query(None, description="Exact match on the raw event date (dd.mm.yy)"),
    upcoming: Optional[bool] = Query(None, description="true = today or later, false = past events"),
    sort: Optional[SortOrder] = Query(None, description=_SORT_DESCRIPTION),
):
    today = datetime.date.today()
    events = [
        e
        for e in store.all()
        if _matches(e, free, name, artist, date, upcoming, today)
    ]
    events = sort_events(events, sort)
    return EventList(count=len(events), events=events)


@app.get("/artists", response_model=List[ArtistSchedule])
def list_artists():
    """Deduplicated, sorted list of all artists, each with every date
    (past or upcoming) they appear in the line-up.
    """
    appearances: dict = defaultdict(set)
    for event in store.all():
        for artist_name in iter_artist_names(event):
            appearances[artist_name].add(event.event_date)
    return [
        ArtistSchedule(artist_name=artist_name, dates=sort_date_strings(dates))
        for artist_name, dates in sorted(appearances.items(), key=lambda kv: kv[0].lower())
    ]


@app.get("/events/archive", response_model=EventList)
def list_archive_events(
    free: Optional[bool] = Query(None, description="Filter by free admission"),
    name: Optional[str] = Query(None, description="Case-insensitive substring match on event name"),
    artist: Optional[str] = Query(None, description="Case-insensitive substring match on any artist name"),
    date: Optional[str] = Query(None, description="Exact match on the raw event date (dd.mm.yy)"),
    sort: Optional[SortOrder] = Query(None, description=_SORT_DESCRIPTION),
):
    """Past events only - shorthand for ``/events?upcoming=false``."""
    today = datetime.date.today()
    events = [
        e
        for e in store.all()
        if _matches(e, free, name, artist, date, upcoming=False, today=today)
    ]
    events = sort_events(events, sort)
    return EventList(count=len(events), events=events)


@app.get("/calendar.ics")
def calendar_ics():
    """iCalendar feed of all known events (upcoming + archived).

    Subscribe from a calendar app with ``webcal://<host>/calendar.ics``
    (or the plain https URL) to get an auto-refreshing calendar.
    """
    ics = build_ics(store.all())
    return PlainTextResponse(
        content=ics,
        media_type="text/calendar",
        headers={"Content-Disposition": "inline; filename=fundbureau-events.ics"},
    )


@app.get("/events/{index}", response_model=Event)
def get_event(index: int):
    event = store.get(index)
    if event is None:
        raise HTTPException(status_code=404, detail=f"No event at index {index}")
    return event