thermograph/backend/scripts/generate_narratives.py

#!/usr/bin/env python3
"""Fill the narrative cache — the batch half of the plain-English day summaries.

This is the only thing in the codebase that calls the LLM. It walks the city
list, asks the running API for each city's graded day, has a local model phrase
it, and writes the sentence to the derived store. ``/api/v2/narrative`` then
serves those rows without ever touching a model.

    # on the desktop, where Ollama is
    python scripts/generate_narratives.py --limit 25

    # from a host that reaches the desktop across the mesh
    THERMOGRAPH_OLLAMA_URL=http://10.x.x.x:11434 \
        python scripts/generate_narratives.py

Two connections, deliberately separate:

* ``--api-base`` (or ``THERMOGRAPH_API_BASE_INTERNAL``) — where to read graded
  days from. Reusing the real endpoint means this script inherits the whole
  fetch/grade/cache pipeline instead of reimplementing it, and warms the grade
  cache as a side effect.
* ``THERMOGRAPH_OLLAMA_URL`` — where inference runs. It needs no access to
  anything else, and nothing waits on it but this process.

Safe to interrupt and safe to re-run: work is per-city and committed as it goes,
and an existing valid row is skipped unless ``--force``. Nothing here writes to
climate data — the only table it touches is ``narrative``.
"""
import argparse
import datetime
import os
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import httpx  # noqa: E402

from data import cities  # noqa: E402
from data import grid  # noqa: E402
from data import narrator  # noqa: E402
from data import store  # noqa: E402

DEFAULT_API = os.environ.get("THERMOGRAPH_API_BASE_INTERNAL", "http://127.0.0.1:8000").rstrip("/")

# The app mounts its API under THERMOGRAPH_BASE (default "/thermograph"), but a
# deployment can and does run with it empty — the LAN dev container serves
# /api/v2/... flat. Guessing wrong turns every city into a 404, so the prefix is
# probed once against the I/O-free /api/version rather than assumed.
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
_PREFIXES = tuple(dict.fromkeys([f"/{_BASE}" if _BASE else "", ""]))


def _detect_prefix(client: httpx.Client, api: str) -> str | None:
    for prefix in _PREFIXES:
        try:
            if client.get(f"{api}{prefix}/api/version", timeout=15).status_code == 200:
                return prefix
        except Exception:  # noqa: BLE001 - try the next candidate
            continue
    return None


def _grade(client: httpx.Client, api: str, prefix: str, lat: float, lon: float,
           target: datetime.date) -> dict | None:
    """One city's /grade payload, or None if the API couldn't produce it.

    ``after=0`` and a short history window keep the response small: the narrative
    only ever describes the target day, so grading two weeks either side of it
    would be work this script pays for and then discards."""
    try:
        r = client.get(f"{api}{prefix}/api/v2/grade",
                       params={"lat": lat, "lon": lon, "date": target.isoformat(),
                               "days": 1, "after": 0})
        r.raise_for_status()
        return r.json()
    except Exception as exc:  # noqa: BLE001
        print(f"    grade failed: {type(exc).__name__}", flush=True)
        return None


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--api-base", default=DEFAULT_API,
                    help=f"backend base URL to read graded days from (default {DEFAULT_API})")
    ap.add_argument("--date", default=None, help="target date YYYY-MM-DD (default today)")
    ap.add_argument("--limit", type=int, default=0, help="stop after N cities (0 = all)")
    ap.add_argument("--slug", action="append", default=[],
                    help="generate only this city slug (repeatable)")
    ap.add_argument("--force", action="store_true",
                    help="regenerate even when a valid row already exists")
    ap.add_argument("--dry-run", action="store_true",
                    help="print the sentence instead of storing it")
    args = ap.parse_args()

    target = (datetime.date.fromisoformat(args.date) if args.date
              else datetime.date.today())
    token = narrator.narrative_token()
    key = narrator.narrative_key(target)

    # Fail fast and loudly rather than timing out a thousand times in a row. The
    # processor line is the one people actually need: an accelerator that has
    # silently dropped out turns a minutes-long pass into an all-day one.
    hp = narrator.health()
    print(f"ollama   : {hp['url']}  reachable={hp['reachable']}  "
          f"model={hp['model']} present={hp['model_present']}")
    if not hp["reachable"]:
        print("ERROR: cannot reach Ollama. Set THERMOGRAPH_OLLAMA_URL.", file=sys.stderr)
        return 1
    if not hp["model_present"]:
        print(f"ERROR: model {hp['model']!r} not pulled. Run: ollama pull {hp['model']}",
              file=sys.stderr)
        return 1

    work = cities.all_cities()
    if args.slug:
        wanted = set(args.slug)
        work = [c for c in work if c["slug"] in wanted]
        missing = wanted - {c["slug"] for c in work}
        if missing:
            print(f"ERROR: unknown slug(s): {', '.join(sorted(missing))}", file=sys.stderr)
            return 1
    if args.limit > 0:
        work = work[:args.limit]

    print(f"target   : {target}  token={token}")
    print(f"cities   : {len(work)}\n")

    made = skipped = failed = 0
    started = time.time()
    with httpx.Client(timeout=120.0) as api_client, httpx.Client(timeout=narrator.TIMEOUT) as llm:
        prefix = _detect_prefix(api_client, args.api_base)
        if prefix is None:
            print(f"ERROR: no Thermograph API at {args.api_base} "
                  f"(tried {', '.join(repr(p) for p in _PREFIXES)})", file=sys.stderr)
            return 1
        print(f"api      : {args.api_base}{prefix}/api/v2\n")

        for n, city in enumerate(work, 1):
            label = f"{city['name']}, {city['country']}"
            cell = grid.snap(city["lat"], city["lon"])

            if not args.force and store.get_narrative(cell["id"], key, token):
                skipped += 1
                continue

            print(f"[{n}/{len(work)}] {label}", flush=True)
            payload = _grade(api_client, args.api_base, prefix,
                             city["lat"], city["lon"], target)
            if payload is None:
                failed += 1
                continue

            # Name the city we asked about, not the cell's reverse-geocoded
            # neighbourhood — see narrator.facts_from_grade.
            facts = narrator.facts_from_grade(payload, target,
                                              place=cities.display_name(city))
            if facts is None:
                # No gradeable metric for this day — usually a cell whose recent
                # bundle hasn't synced yet. Not an error; it'll be picked up on a
                # later pass once the data lands.
                print("    no graded data for the target day — skipping", flush=True)
                skipped += 1
                continue

            text = narrator.generate(facts, client=llm)
            if not text:
                print("    generation failed or was rejected by the sanitizer", flush=True)
                failed += 1
                continue

            print(f"    {text}", flush=True)
            if not args.dry_run:
                store.put_narrative(cell["id"], key, token, text, narrator.MODEL)
            made += 1

    elapsed = time.time() - started
    print(f"\ndone in {elapsed:.0f}s — {made} written, {skipped} skipped, {failed} failed")
    if args.dry_run:
        print("(dry run — nothing was stored)")
    # A run that produced nothing while having work to do is a failure worth a
    # non-zero exit, so a cron wrapper notices.
    return 1 if made == 0 and failed > 0 else 0


if __name__ == "__main__":
    raise SystemExit(main())