#!/usr/bin/env -S uv run --script --quiet
# /// script
# requires-python = ">=3.12,<3.14"
# # numpy<2.2 + python<3.14: umap-learn's transitive numba/llvmlite have no wheel
# # for numpy>=2.2 or py3.14, so an unpinned solve backtracks to ancient numba
# # 0.53 and tries a source build that fails the py-version guard. keep both pins.
# dependencies = ["httpx", "numpy<2.2", "scikit-learn", "umap-learn", "hdbscan", "pydantic-settings", "anthropic"]
# ///
"""
Build atlas.json — 2D semantic map of the document index.

Exports vectors from turbopuffer, projects to 2D via PCA+UMAP,
clusters with HDBSCAN at two granularities, labels via c-TF-IDF.

Usage:
    ./scripts/build-atlas           # writes site/atlas.json
    ./scripts/build-atlas --output out.json
"""

import argparse
import asyncio
import gc
import json
import os
import re
import sys
import time
from collections import Counter
from pathlib import Path

import httpx
import numpy as np
from pydantic import ValidationError
from pydantic_settings import BaseSettings, SettingsConfigDict
from sklearn.decomposition import PCA
from sklearn.feature_extraction.text import TfidfVectorizer


# avatar cache: persists did → avatar URL across runs.
# lives in site/ so it gets deployed alongside atlas.json — the deployed copy
# is what the prefect flow (which clones fresh each run) reads on cold start.
AVATAR_CACHE_PATH = Path(__file__).resolve().parent.parent / "site" / "atlas-avatar-cache.json"
AVATAR_CACHE_REMOTE = "https://pub-search.waow.tech/atlas-avatar-cache.json"
AVATAR_CACHE_TTL_SECONDS = 7 * 86400  # refresh entries weekly so changed avatars trickle in

# theme cache: persists basePath → leaflet publication theme colors.
# same deployment-roundtrip pattern as the avatar cache above.
THEME_CACHE_PATH = Path(__file__).resolve().parent.parent / "site" / "atlas-theme-cache.json"
THEME_CACHE_REMOTE = "https://pub-search.waow.tech/atlas-theme-cache.json"


def log(msg: str) -> None:
    print(msg, flush=True)


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
    )

    turbopuffer_api_key: str
    turbopuffer_namespace: str = "leaflet-search"
    anthropic_api_key: str = ""  # optional: enables LLM-refined cluster labels
    turso_url: str = ""  # optional: enables publication data (e.g. https://leaf-....turso.io)
    turso_token: str = ""

    @classmethod
    def settings_customise_sources(cls, settings_cls, **kwargs):
        """Dotenv file wins over environment variables."""
        return (
            kwargs["init_settings"],
            kwargs["dotenv_settings"],
            kwargs["env_settings"],
            kwargs["file_secret_settings"],
        )


def tpuf_export(client: httpx.Client, settings: Settings) -> list[dict]:
    """Export all vectors from turbopuffer via paginated query."""
    url = f"https://api.turbopuffer.com/v2/namespaces/{settings.turbopuffer_namespace}/query"
    headers = {
        "Authorization": f"Bearer {settings.turbopuffer_api_key}",
        "Content-Type": "application/json",
    }
    attrs = ["uri", "title", "platform", "base_path", "path", "did", "vector"]
    all_rows = []
    last_id = None

    while True:
        body: dict = {
            "rank_by": ["id", "asc"],
            "limit": 10000,
            "include_attributes": attrs,
        }
        if last_id is not None:
            body["filters"] = ["id", "Gt", last_id]

        resp = client.post(url, headers=headers, json=body, timeout=120)
        resp.raise_for_status()
        rows = resp.json().get("rows", [])
        if not rows:
            break

        all_rows.extend(rows)
        last_id = rows[-1]["id"]
        log(f"  fetched {len(all_rows)} vectors so far...")

        if len(rows) < 10000:
            break

    return all_rows


def turso_query_publications(settings: Settings) -> dict[str, dict]:
    """Fetch publication metadata from Turso. Returns basePath → {name, did, coverImage}."""
    if not settings.turso_url or not settings.turso_token:
        return {}

    host = settings.turso_url.replace("https://", "").replace("http://", "").replace("libsql://", "")
    url = f"https://{host}/v3/pipeline"
    headers = {"Authorization": f"Bearer {settings.turso_token}"}

    body = {
        "requests": [
            {
                "type": "execute",
                "stmt": {
                    "sql": (
                        "SELECT p.name, p.base_path, p.did, c.cover_image "
                        "FROM publications p "
                        "LEFT JOIN ("
                        "  SELECT publication_uri, MIN(cover_image) as cover_image "
                        "  FROM documents WHERE cover_image IS NOT NULL "
                        "  GROUP BY publication_uri"
                        ") c ON c.publication_uri = p.uri"
                    )
                },
            },
            {"type": "close"},
        ]
    }

    try:
        resp = httpx.post(url, headers=headers, json=body, timeout=30)
        resp.raise_for_status()
        results = resp.json().get("results", [])
        if not results or "response" not in results[0]:
            return {}
        rows = results[0]["response"]["result"]["rows"]
        cols = [c["name"] for c in results[0]["response"]["result"]["cols"]]
    except httpx.HTTPError as e:
        log(f"  warning: turso request failed: {e}")
        return {}
    except (KeyError, IndexError) as e:
        log(f"  warning: unexpected turso response: {e}")
        return {}
    lookup = {}
    for row in rows:
        values = {cols[i]: (row[i]["value"] if row[i]["type"] != "null" else None) for i in range(len(cols))}
        bp = values.get("base_path")
        if bp:
            lookup[bp] = {
                "name": values.get("name") or bp,
                "did": values.get("did") or "",
                "coverImage": values.get("cover_image") or "",
            }
    return lookup


def extract_terms(title: str) -> list[str]:
    """Extract lowercase alphanumeric tokens from a title."""
    if not title:
        return []
    return [t for t in re.findall(r"[a-z0-9]+", title.lower()) if len(t) > 1]


def cluster_labels(titles_per_cluster: dict[int, list[str]], n_terms: int = 3) -> dict[int, str]:
    """Compute c-TF-IDF labels for clusters from document titles."""
    cluster_ids = sorted(titles_per_cluster.keys())
    if not cluster_ids:
        return {}

    # build one "document" per cluster: concatenated titles
    docs = []
    for cid in cluster_ids:
        docs.append(" ".join(titles_per_cluster[cid]))

    # TF-IDF across cluster pseudo-documents
    vectorizer = TfidfVectorizer(
        max_features=5000,
        stop_words="english",
        token_pattern=r"[a-z0-9]{2,}",
        lowercase=True,
    )
    try:
        tfidf = vectorizer.fit_transform(docs)
    except ValueError:
        return {cid: f"cluster {cid}" for cid in cluster_ids}

    feature_names = vectorizer.get_feature_names_out()
    labels = {}
    for i, cid in enumerate(cluster_ids):
        row = tfidf[i].toarray().flatten()
        top_idx = row.argsort()[-n_terms:][::-1]
        top_terms = [feature_names[j] for j in top_idx if row[j] > 0]
        labels[cid] = " ".join(top_terms) if top_terms else f"cluster {cid}"

    return labels


def llm_refine_labels(
    tfidf_labels: dict[int, str],
    titles_per_cluster: dict[int, list[str]],
    cluster_counts: dict[int, int],
    api_key: str,
    tier: str = "coarse",
    batch_size: int = 15,
) -> dict[int, str]:
    """Refine c-TF-IDF labels with Claude Haiku for human-readable topic names."""
    from anthropic import AsyncAnthropic

    client = AsyncAnthropic(api_key=api_key)
    rng = np.random.default_rng(42)

    cluster_ids = sorted(tfidf_labels.keys())
    batches = [cluster_ids[i : i + batch_size] for i in range(0, len(cluster_ids), batch_size)]

    # Split fixed instructions (system, cached) from variable per-batch payload
    # (user). Each rebuild runs ~18 batches (5 coarse + 13 fine); the
    # instructions are identical across all of them, so caching the system
    # block drops input cost to ~10% on every batch after the first within
    # the 5m TTL.
    system_instructions = (
        "You are labeling topic clusters for a document atlas — a 2D map where each cluster "
        "represents a group of semantically similar blog posts and articles.\n\n"
        "For each cluster you receive, write a short descriptive label (2-4 lowercase words) "
        "that captures the theme. The label should read like a topic name on a map.\n\n"
        "Good labels: \"ai language models\", \"bluesky development\", \"personal journals\", "
        "\"spanish poetry\", \"game reviews\", \"home cooking\"\n"
        "Bad labels: \"ai llms llm\", \"results april june\", \"39 let english\"\n\n"
        "Respond with ONLY a JSON object mapping cluster ID (as string) to label. No other text."
    )

    async def label_batch(batch: list[int]) -> dict[int, str]:
        parts = []
        for cid in batch:
            titles = titles_per_cluster.get(cid, [])
            sample = list(rng.choice(titles, size=min(12, len(titles)), replace=False)) if titles else []
            parts.append(
                f"Cluster {cid} ({cluster_counts.get(cid, 0)} docs):\n"
                f"  Keywords: {tfidf_labels[cid]}\n"
                f"  Sample titles: {', '.join(repr(t) for t in sample)}"
            )

        resp = await client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=2048,
            system=[
                {
                    "type": "text",
                    "text": system_instructions,
                    "cache_control": {"type": "ephemeral"},
                }
            ],
            messages=[{"role": "user", "content": "\n\n".join(parts)}],
        )
        text = resp.content[0].text.strip()
        # extract JSON from response (handle markdown code fences)
        if text.startswith("```"):
            text = text.split("\n", 1)[1].rsplit("```", 1)[0].strip()
        try:
            parsed = json.loads(text)
            return {int(k): str(v) for k, v in parsed.items()}
        except (json.JSONDecodeError, ValueError):
            log(f"  warning: failed to parse LLM response for batch, using TF-IDF fallback")
            return {}

    async def run_all():
        results = await asyncio.gather(*[label_batch(b) for b in batches])
        merged = {}
        for r in results:
            merged.update(r)
        return merged

    log(f"  refining {tier} labels with LLM ({len(batches)} batches, {len(cluster_ids)} clusters)...")
    refined = asyncio.run(run_all())

    # merge: use LLM labels where available, fall back to c-TF-IDF
    final = {}
    for cid in cluster_ids:
        final[cid] = refined.get(cid, tfidf_labels[cid])
    log(f"  {tier}: {len(refined)}/{len(cluster_ids)} labels refined by LLM")
    return final


def assign_outliers(coords_2d: np.ndarray, labels: np.ndarray, centroids: dict[int, np.ndarray]) -> np.ndarray:
    """Assign outlier points (label == -1) to nearest cluster centroid."""
    result = labels.copy()
    outlier_mask = result == -1
    if not outlier_mask.any() or not centroids:
        return result

    centroid_ids = sorted(centroids.keys())
    centroid_arr = np.array([centroids[c] for c in centroid_ids])
    outlier_coords = coords_2d[outlier_mask]

    # nearest centroid for each outlier
    dists = np.linalg.norm(outlier_coords[:, None] - centroid_arr[None, :], axis=2)
    nearest = dists.argmin(axis=1)
    result[outlier_mask] = np.array(centroid_ids)[nearest]

    return result


def load_avatar_cache(local_path: Path, remote_url: str) -> dict[str, dict]:
    """Load avatar cache. Prefer local file (dev re-runs); fall back to the
    previously-deployed copy on the CDN (prefect flow clones fresh each run).
    Returns {} on any failure — caller treats that as a cold start."""
    if local_path.exists():
        try:
            data = json.loads(local_path.read_text())
            log(f"  loaded {len(data)} avatar cache entries from {local_path.name}")
            return data
        except (json.JSONDecodeError, OSError) as e:
            log(f"  warning: local avatar cache unreadable ({e}), trying remote")

    try:
        resp = httpx.get(remote_url, timeout=15)
        if resp.status_code == 200:
            data = resp.json()
            if isinstance(data, dict):
                log(f"  loaded {len(data)} avatar cache entries from {remote_url}")
                return data
            log(f"  warning: remote avatar cache has unexpected shape, starting cold")
        else:
            log(f"  remote avatar cache returned {resp.status_code}, starting cold")
    except (httpx.HTTPError, json.JSONDecodeError) as e:
        log(f"  warning: failed to fetch remote avatar cache ({e}), starting cold")

    return {}


def save_avatar_cache(path: Path, entries: dict[str, dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(entries, separators=(",", ":"), sort_keys=True))


async def resolve_avatars_async(
    dids: list[str], cache: dict[str, dict], concurrency: int = 20
) -> dict[str, dict]:
    """Resolve bsky avatars for the given DIDs.

    Uses cached entries within TTL, fetches the rest concurrently. Returns the
    new cache state — exactly the DIDs passed in, so unused entries from prior
    runs naturally drop out and the file stays roughly the size of the corpus.
    Empty-string avatars are cached too, so we don't hammer the API every run
    for authors who have no avatar set."""
    now = time.time()
    cutoff = now - AVATAR_CACHE_TTL_SECONDS

    fresh: dict[str, dict] = {}
    missing: list[str] = []
    for did in dids:
        entry = cache.get(did)
        if isinstance(entry, dict) and float(entry.get("ts", 0)) > cutoff:
            fresh[did] = entry
        else:
            missing.append(did)

    log(f"  cache: {len(fresh)} hits, {len(missing)} to fetch")
    if not missing:
        return fresh

    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(10.0),
        limits=httpx.Limits(
            max_connections=concurrency,
            max_keepalive_connections=concurrency,
        ),
    ) as client:
        async def fetch_one(did: str) -> tuple[str, str]:
            async with sem:
                try:
                    r = await client.get(
                        f"https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile?actor={did}"
                    )
                    if r.status_code == 200:
                        return did, (r.json().get("avatar") or "")
                except Exception:
                    pass
                return did, ""

        results = await asyncio.gather(*(fetch_one(d) for d in missing))

    for did, avatar in results:
        fresh[did] = {"avatar": avatar, "ts": now}

    return fresh


def _theme_color_hex(color: dict | None) -> str:
    """pub.leaflet.theme.color#rgb{a} object → #rrggbb, or '' if absent."""
    if not isinstance(color, dict):
        return ""
    try:
        return "#%02x%02x%02x" % (int(color["r"]), int(color["g"]), int(color["b"]))
    except (KeyError, ValueError, TypeError):
        return ""


async def resolve_themes_async(
    pubs: list[dict], cache: dict[str, dict], concurrency: int = 16
) -> dict[str, dict]:
    """Resolve leaflet publication theme colors from their PDS records.

    pub.leaflet.publication records carry the author's actual palette
    (backgroundColor / accentBackground); the atlas uses it to paint that
    pub's planets. One listRecords per DID covers all of its publications.
    Empty results are cached too. Returns {basePath: {bg, accent, ts}} for
    exactly the leaflet pubs passed in."""
    now = time.time()
    cutoff = now - AVATAR_CACHE_TTL_SECONDS

    targets = [p for p in pubs if p.get("platform") == "leaflet" and str(p.get("did", "")).startswith("did:plc:")]
    fresh: dict[str, dict] = {}
    by_did: dict[str, list[str]] = {}
    for p in targets:
        entry = cache.get(p["basePath"])
        if isinstance(entry, dict) and float(entry.get("ts", 0)) > cutoff:
            fresh[p["basePath"]] = entry
        else:
            by_did.setdefault(p["did"], []).append(p["basePath"])

    log(f"  theme cache: {len(fresh)} hits, {len(by_did)} DIDs to fetch")
    if not by_did:
        return fresh

    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(10.0),
        limits=httpx.Limits(
            max_connections=concurrency,
            max_keepalive_connections=concurrency,
        ),
    ) as client:
        async def fetch_did(did: str, base_paths: list[str]) -> dict[str, dict]:
            out = {bp: {"bg": "", "accent": "", "ts": now} for bp in base_paths}
            async with sem:
                try:
                    doc = (await client.get(f"https://plc.directory/{did}")).json()
                    pds = next(
                        s["serviceEndpoint"] for s in doc.get("service", [])
                        if s.get("id") == "#atproto_pds"
                    )
                    r = await client.get(
                        f"{pds}/xrpc/com.atproto.repo.listRecords",
                        params={"repo": did, "collection": "pub.leaflet.publication", "limit": 50},
                    )
                    if r.status_code != 200:
                        return out
                    for rec in r.json().get("records", []):
                        value = rec.get("value", {})
                        bp = value.get("base_path", "")
                        if bp not in out:
                            continue
                        theme = value.get("theme") or {}
                        out[bp] = {
                            "bg": _theme_color_hex(theme.get("backgroundColor") or theme.get("pageBackground")),
                            "accent": _theme_color_hex(theme.get("accentBackground") or theme.get("primary")),
                            "ts": now,
                        }
                except Exception:
                    pass
            return out

        results = await asyncio.gather(*(fetch_did(d, bps) for d, bps in by_did.items()))

    for chunk in results:
        fresh.update(chunk)
    return fresh


def main():
    parser = argparse.ArgumentParser(description="Build atlas.json")
    parser.add_argument(
        "--output", "-o",
        default=str(Path(__file__).resolve().parent.parent / "site" / "atlas.json"),
        help="Output path (default: site/atlas.json)",
    )
    args = parser.parse_args()

    try:
        settings = Settings()  # type: ignore
    except ValidationError as e:
        # never print the exception directly: pydantic's str(ValidationError)
        # embeds input_value — the full dict of ingested env/dotenv vars,
        # including unrelated secrets — which would land in the prefect logs.
        # surface only the offending field names, never their values.
        fields = ", ".join(str(err["loc"][0]) for err in e.errors()) or "config"
        print(f"error loading settings; missing/invalid: {fields}", file=sys.stderr)
        print("required env vars: TURBOPUFFER_API_KEY", file=sys.stderr)
        sys.exit(1)

    t_start = time.monotonic()

    # --- step 1: export vectors from turbopuffer ---
    log("exporting vectors from turbopuffer...")
    client = httpx.Client()
    rows = tpuf_export(client, settings)
    client.close()
    log(f"  got {len(rows)} documents")

    if len(rows) < 10:
        print("too few documents to build atlas", file=sys.stderr)
        sys.exit(1)

    # Parse into a dense float32 matrix without retaining a second Python list
    # of vector objects. Turbopuffer returns vectors as JSON lists; holding
    # rows + vectors + X at once was enough to OOM the Prefect pod as the
    # corpus grew.
    valid_rows = [row for row in rows if row.get("vector")]
    first_vec = valid_rows[0]["vector"]
    X = np.empty((len(valid_rows), len(first_vec)), dtype=np.float32)
    metadata = []
    for i, row in enumerate(valid_rows):
        vec = row.pop("vector")
        X[i] = vec
        metadata.append({
            "uri": row.get("uri", row.get("id", "")),
            "title": row.get("title", ""),
            "platform": row.get("platform", "other"),
            "basePath": row.get("base_path", ""),
            "path": row.get("path", ""),
            "did": row.get("did", ""),
        })
    del rows, valid_rows
    gc.collect()
    log(f"  {X.shape[0]} vectors, {X.shape[1]} dims")

    # --- step 2: dimensionality reduction ---
    log("PCA 1024 → 50...")
    n_components_pca = min(50, X.shape[0] - 1, X.shape[1])
    pca = PCA(n_components=n_components_pca, random_state=42)
    X_pca = pca.fit_transform(X)
    log(f"  variance explained: {pca.explained_variance_ratio_.sum():.2%}")

    log("UMAP 50 → 2...")
    import umap
    reducer = umap.UMAP(
        n_components=2,
        metric="cosine",
        n_neighbors=15,
        min_dist=0.1,
        random_state=42,
    )
    X_2d = reducer.fit_transform(X_pca)

    # normalize to [-1, 1] range
    for dim in range(2):
        lo, hi = X_2d[:, dim].min(), X_2d[:, dim].max()
        if hi > lo:
            X_2d[:, dim] = 2 * (X_2d[:, dim] - lo) / (hi - lo) - 1

    log(f"  2D range: x=[{X_2d[:,0].min():.2f}, {X_2d[:,0].max():.2f}] y=[{X_2d[:,1].min():.2f}, {X_2d[:,1].max():.2f}]")

    # --- step 3: clustering ---
    import hdbscan

    log("HDBSCAN coarse (min_cluster_size=50)...")
    clusterer_coarse = hdbscan.HDBSCAN(min_cluster_size=50, min_samples=10)
    labels_coarse_raw = clusterer_coarse.fit_predict(X_2d)
    n_coarse = len(set(labels_coarse_raw)) - (1 if -1 in labels_coarse_raw else 0)
    n_outliers_coarse = (labels_coarse_raw == -1).sum()
    log(f"  {n_coarse} clusters, {n_outliers_coarse} outliers")

    # compute centroids for coarse
    coarse_centroids = {}
    for cid in set(labels_coarse_raw):
        if cid == -1:
            continue
        mask = labels_coarse_raw == cid
        coarse_centroids[cid] = X_2d[mask].mean(axis=0)

    labels_coarse = assign_outliers(X_2d, labels_coarse_raw, coarse_centroids)
    # recompute centroids after outlier assignment
    for cid in set(labels_coarse):
        mask = labels_coarse == cid
        coarse_centroids[cid] = X_2d[mask].mean(axis=0)

    log("HDBSCAN fine (min_cluster_size=20)...")
    clusterer_fine = hdbscan.HDBSCAN(min_cluster_size=20, min_samples=3)
    labels_fine_raw = clusterer_fine.fit_predict(X_2d)
    n_fine = len(set(labels_fine_raw)) - (1 if -1 in labels_fine_raw else 0)
    n_outliers_fine = (labels_fine_raw == -1).sum()
    log(f"  {n_fine} clusters, {n_outliers_fine} outliers")

    fine_centroids = {}
    for cid in set(labels_fine_raw):
        if cid == -1:
            continue
        mask = labels_fine_raw == cid
        fine_centroids[cid] = X_2d[mask].mean(axis=0)

    labels_fine = assign_outliers(X_2d, labels_fine_raw, fine_centroids)
    for cid in set(labels_fine):
        mask = labels_fine == cid
        fine_centroids[cid] = X_2d[mask].mean(axis=0)

    # --- step 4: labels via c-TF-IDF + optional LLM refinement ---
    log("computing cluster labels (c-TF-IDF on titles)...")

    coarse_titles: dict[int, list[str]] = {}
    fine_titles: dict[int, list[str]] = {}
    for i, meta in enumerate(metadata):
        title = meta.get("title", "")
        if title:
            coarse_titles.setdefault(int(labels_coarse[i]), []).append(title)
            fine_titles.setdefault(int(labels_fine[i]), []).append(title)

    coarse_labels = cluster_labels(coarse_titles)
    fine_labels = cluster_labels(fine_titles)

    if settings.anthropic_api_key:
        log("refining labels with LLM...")
        coarse_counts = {int(cid): int((labels_coarse == cid).sum()) for cid in set(labels_coarse)}
        fine_counts = {int(cid): int((labels_fine == cid).sum()) for cid in set(labels_fine)}
        try:
            coarse_labels = llm_refine_labels(
                coarse_labels, coarse_titles, coarse_counts,
                settings.anthropic_api_key, tier="coarse",
            )
            fine_labels = llm_refine_labels(
                fine_labels, fine_titles, fine_counts,
                settings.anthropic_api_key, tier="fine",
            )
        except Exception as e:
            log(f"  LLM labeling failed, using c-TF-IDF fallback: {e}")
    else:
        log("  (no ANTHROPIC_API_KEY — using c-TF-IDF labels only)")

    # map fine clusters to their parent coarse cluster (majority vote)
    fine_to_coarse = {}
    for fine_id in set(labels_fine):
        fine_mask = labels_fine == fine_id
        coarse_for_fine = labels_coarse[fine_mask]
        counts = Counter(coarse_for_fine)
        fine_to_coarse[int(fine_id)] = int(counts.most_common(1)[0][0])

    log(f"  coarse: {len(coarse_labels)} labels")
    log(f"  fine: {len(fine_labels)} labels")

    # --- step 5: publication centroids ---
    log("computing publication centroids...")
    pub_lookup = turso_query_publications(settings)
    if pub_lookup:
        log(f"  fetched {len(pub_lookup)} publications from turso")
    else:
        log("  (no TURSO_URL/TURSO_TOKEN — using basePath grouping only)")

    # group documents by basePath → list of 2D positions
    bp_groups: dict[str, list[int]] = {}
    for i, meta in enumerate(metadata):
        bp = meta.get("basePath", "")
        if bp:
            bp_groups.setdefault(bp, []).append(i)

    publications = []
    for bp, indices in bp_groups.items():
        if len(indices) < 2:
            continue
        coords = X_2d[indices]
        centroid = coords.mean(axis=0)
        info = pub_lookup.get(bp, {})
        # get DID from turso lookup first, fall back to documents
        did = info.get("did", "")
        if not did:
            for idx in indices:
                d = metadata[idx].get("did", "")
                if d:
                    did = d
                    break
        pub = {
            "name": info.get("name") or bp,
            "basePath": bp,
            "did": did,
            "cx": round(float(centroid[0]), 4),
            "cy": round(float(centroid[1]), 4),
            "count": len(indices),
            "coverImage": info.get("coverImage", ""),
        }
        # determine dominant platform for this publication
        plat_counts: dict[str, int] = {}
        for idx in indices:
            p = metadata[idx].get("platform", "other")
            plat_counts[p] = plat_counts.get(p, 0) + 1
        pub["platform"] = max(plat_counts, key=plat_counts.get)  # type: ignore
        publications.append(pub)

    publications.sort(key=lambda p: p["count"], reverse=True)
    log(f"  {len(publications)} publications with 2+ documents")

    # fetch author avatars for publications (fallback when no cover image).
    # see resolve_avatars_async — concurrent fetch + persistent cache keep this
    # step at ~5s steady-state instead of ~85s serial.
    unique_dids = {p["did"] for p in publications if p["did"]}
    log(f"  resolving avatars for {len(unique_dids)} unique authors...")

    cache = load_avatar_cache(AVATAR_CACHE_PATH, AVATAR_CACHE_REMOTE)
    entries = asyncio.run(resolve_avatars_async(sorted(unique_dids), cache))
    save_avatar_cache(AVATAR_CACHE_PATH, entries)

    did_avatars = {did: e["avatar"] for did, e in entries.items() if e.get("avatar")}
    log(f"  resolved {len(did_avatars)} avatars ({len(entries) - len(did_avatars)} authors have none)")

    # attach avatar URLs to publications
    for pub in publications:
        pub["avatar"] = did_avatars.get(pub["did"], "")

    # fetch leaflet publication theme colors — the author's actual palette,
    # used by the atlas to paint that publication's planets
    log("  resolving leaflet publication themes...")
    theme_cache = load_avatar_cache(THEME_CACHE_PATH, THEME_CACHE_REMOTE)
    theme_entries = asyncio.run(resolve_themes_async(publications, theme_cache))
    save_avatar_cache(THEME_CACHE_PATH, theme_entries)
    themed = 0
    for pub in publications:
        t = theme_entries.get(pub["basePath"]) or {}
        if t.get("accent"):
            pub["themeAccent"] = t["accent"]
            if t.get("bg"):
                pub["themeBg"] = t["bg"]
            themed += 1
    log(f"  {themed} publications have theme colors")

    # --- step 6: build output ---
    log("building output...")

    points = []
    for i, meta in enumerate(metadata):
        points.append({
            "x": round(float(X_2d[i, 0]), 4),
            "y": round(float(X_2d[i, 1]), 4),
            "uri": meta["uri"],
            "title": meta["title"],
            "platform": meta["platform"],
            "basePath": meta["basePath"],
            "path": meta["path"],
            "clusterCoarse": int(labels_coarse[i]),
            "clusterFine": int(labels_fine[i]),
        })

    coarse_clusters = []
    for cid in sorted(coarse_centroids.keys()):
        count = int((labels_coarse == cid).sum())
        coarse_clusters.append({
            "id": int(cid),
            "label": coarse_labels.get(int(cid), f"cluster {cid}"),
            "cx": round(float(coarse_centroids[cid][0]), 4),
            "cy": round(float(coarse_centroids[cid][1]), 4),
            "count": count,
        })

    fine_clusters = []
    for cid in sorted(fine_centroids.keys()):
        count = int((labels_fine == cid).sum())
        fine_clusters.append({
            "id": int(cid),
            "label": fine_labels.get(int(cid), f"cluster {cid}"),
            "cx": round(float(fine_centroids[cid][0]), 4),
            "cy": round(float(fine_centroids[cid][1]), 4),
            "count": count,
            "parent": fine_to_coarse.get(int(cid), 0),
        })

    output = {
        "points": points,
        "clusters": {
            "coarse": coarse_clusters,
            "fine": fine_clusters,
        },
        "publications": publications,
        "meta": {
            "generatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "nDocuments": len(points),
            "nPublications": len(publications),
        },
    }

    out_path = Path(args.output)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(output, separators=(",", ":")))

    size_kb = out_path.stat().st_size / 1024
    elapsed = time.monotonic() - t_start
    log(f"\nwrote {out_path} ({size_kb:.0f} KB, {len(points)} points)")
    log(f"done in {elapsed:.1f}s")


if __name__ == "__main__":
    main()
