Spaces:

omerehrlich
/

WikiStar

Sleeping

TEMP: log summary cache hits/misses (revert after testing)
b608171 15 days ago
73.6 kB
import hashlib
import json
import logging
import os
import re
import secrets
import sqlite3
import threading
import time
import traceback
from collections import OrderedDict, deque
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any, Deque, Dict, List, Optional, Union

import pandas as pd
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from revision_history import fetch_revision_history
from flexible_revision_linker import add_flexible_revision_linking
from generate_prompts_from_linking import (
    create_taxonomy,
    generate_prompts_from_flexible_linking,
)
from Proccessor.parsers import parse_prompt
from visualization import plot_label_distribution, plot_label_cooccurrence

# Structural / non-scientific sections hidden from the aggregated views and
# results everywhere. Matched case-insensitively against a row's trimmed
# `Section` name. Mirrors HIDDEN_SECTIONS in the frontend's taxonomy.ts — the
# browser strips these at ingestion; this backend guard keeps server-generated
# summaries (which re-filter by grouped_idx) from leaking rows whose section
# happens to share a union-find group with a visible section.
HIDDEN_SECTIONS = {
    "see also",
    "further reading",
    "external links",
    "references",
    "other sources",
    "resources",
}


def _is_hidden_section(section: Any) -> bool:
    if section is None or (isinstance(section, float) and pd.isna(section)):
        return False
    return str(section).strip().lower() in HIDDEN_SECTIONS


# User-supplied OpenAI keys must never reach the logs. OpenAI error messages
# echo part of the key back ("Incorrect API key provided: sk-proj-****abcd"),
# so anything printed from an exception goes through _redact_secrets first.
_SECRET_PATTERN = re.compile(r"(sk-[A-Za-z0-9_\-*.]{4,}|Bearer\s+[A-Za-z0-9_\-*.]+)")


def _redact_secrets(text: Any) -> str:
    return _SECRET_PATTERN.sub("[REDACTED]", str(text))


class _RedactSecretsFilter(logging.Filter):
    """Scrubs keys from log records, including formatted tracebacks — covers
    uvicorn's "Exception in ASGI application" output for uncaught errors."""

    def filter(self, record: logging.LogRecord) -> bool:
        # Redact args in place rather than flattening into msg: uvicorn's
        # AccessFormatter unpacks record.args positionally.
        if isinstance(record.msg, str):
            record.msg = _redact_secrets(record.msg)
        if isinstance(record.args, tuple):
            record.args = tuple(
                _redact_secrets(a) if isinstance(a, str) else a for a in record.args
            )
        if record.exc_info:
            record.exc_text = _redact_secrets(
                "".join(traceback.format_exception(*record.exc_info))
            ).rstrip("\n")
            record.exc_info = None
        return True


def _install_log_redaction() -> None:
    redact = _RedactSecretsFilter()
    for name in ("", "uvicorn", "uvicorn.error", "uvicorn.access", "fastapi"):
        logger = logging.getLogger(name)
        logger.addFilter(redact)
        for handler in logger.handlers:
            handler.addFilter(redact)


# In-memory progress per classification run, keyed by job ID. Each /classify
# call mints a fresh unguessable ID, so two users classifying the same article
# get separate jobs and nobody can read a run they don't hold the ID for.
progress_map: Dict[str, Dict[str, Any]] = {}

# Add a lock for thread-safe access to progress_map
progress_lock = threading.Lock()

_JOB_ID_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}$")


def _new_job_id() -> str:
    return secrets.token_urlsafe(16)


# A running job heartbeats (updated_at) on every stage change and every batch
# poll (~10s). One that goes quiet for JOB_STALL_SECONDS, or runs longer than
# JOB_MAX_SECONDS (OpenAI's 24h batch window plus slack), is marked failed so
# clients stop waiting. Tunable via Space env vars.
JOB_STALL_SECONDS = int(os.environ.get("JOB_STALL_SECONDS", str(15 * 60)))
JOB_MAX_SECONDS = int(os.environ.get("JOB_MAX_SECONDS", str(25 * 3600)))


def _expire_if_stuck(entry: Dict[str, Any], now: float) -> None:
    """Flip a stalled or overlong job to the error stage. Caller holds
    progress_lock."""
    if entry.get("stage") in ("complete", "error"):
        return
    started = entry.get("started_at") or now
    last_beat = entry.get("updated_at") or started
    if now - started > JOB_MAX_SECONDS:
        message = (
            f"Classification timed out after {JOB_MAX_SECONDS // 3600} hours. "
            "Please try again."
        )
    elif now - last_beat > JOB_STALL_SECONDS:
        message = (
            f"Classification stopped responding (no progress for "
            f"{JOB_STALL_SECONDS // 60} minutes). Please try again."
        )
    else:
        return
    entry["stage"] = "error"
    entry["error"] = message
    entry["timed_out"] = True


def _set_job_fields(job_id: str, **fields: Any) -> None:
    """Update a running job and refresh its heartbeat. Ignored once the job
    has timed out, so a late worker can't resurrect it."""
    with progress_lock:
        entry = progress_map.setdefault(job_id, {})
        if entry.get("timed_out"):
            return
        entry.update(fields)
        entry["updated_at"] = time.time()


def _get_job(job_id: str) -> Dict[str, Any]:
    """Snapshot of a job's progress entry, or 404. The format check also keeps
    the ID safe to embed in file paths."""
    if not _JOB_ID_RE.match(job_id or ""):
        raise HTTPException(status_code=404, detail="Unknown job.")
    with progress_lock:
        entry = progress_map.get(job_id)
        if entry is None:
            raise HTTPException(status_code=404, detail="Unknown job.")
        _expire_if_stuck(entry, time.time())
        return dict(entry)


def _job_file(job: Dict[str, Any], suffix: str) -> str:
    """Filename under visualizations/ for one of a job's output files."""
    return f"{job['file_prefix']}_{suffix}"

USAGE_LOG_PATH = os.path.join("visualizations", "openai_usage_log.jsonl")
_usage_log_lock = threading.Lock()


def _log_openai_call(
    endpoint: str,
    model: str,
    article: Optional[str],
    used_server_key: bool,
    prompt_chars: int,
    response_chars: int,
    usage: Optional[Dict[str, Any]] = None,
    extra: Optional[Dict[str, Any]] = None,
) -> None:
    """Append one line describing an OpenAI call to the log. Best-effort:
    failures are swallowed so logging can never break the request."""
    record: Dict[str, Any] = {
        "ts": _utc_now_iso(),
        "endpoint": endpoint,
        "model": model,
        "article": article,
        "used_server_key": bool(used_server_key),
        "prompt_chars": int(prompt_chars),
        "response_chars": int(response_chars),
    }
    if usage:
        record["usage"] = usage
    if extra:
        record.update(extra)
    try:
        with _usage_log_lock:
            with open(USAGE_LOG_PATH, "a", encoding="utf-8") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception as log_err:  # pragma: no cover - logging must never throw
        print(f"[warn] failed to write usage log: {log_err}")


def _utc_now_iso() -> str:
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()


# Free summaries on the server's key, per browser. Keyed by an anonymous
# client ID the frontend keeps in localStorage and sends as X-Client-Id —
# deliberately not by IP, since a conference audience shares one network.
# In-memory, so counts reset when the Space restarts.
FREE_SUMMARY_LIMIT = 25
_free_summary_counts: Dict[str, int] = {}
_free_summary_lock = threading.Lock()

# Hard ceilings on server-key summaries across ALL clients. X-Client-Id is
# chosen by the caller, so rotating IDs dodges the per-browser quota; these
# caps bound the worst-case spend regardless. Tunable via Space env vars.
SERVER_KEY_HOURLY_LIMIT = int(os.environ.get("SERVER_KEY_HOURLY_LIMIT", "200"))
SERVER_KEY_DAILY_LIMIT = int(os.environ.get("SERVER_KEY_DAILY_LIMIT", "1000"))
_server_key_calls: Deque[float] = deque()


def _server_key_unavailable() -> HTTPException:
    return HTTPException(
        status_code=429,
        detail={
            "code": "server_key_unavailable",
            "message": (
                "Free summaries are temporarily unavailable. To "
                "keep getting summaries, enter your OpenAI API key"
            ),
        },
    )


def _free_quota_exceeded() -> HTTPException:
    return HTTPException(
        status_code=429,
        detail={
            "code": "free_quota_exceeded",
            "limit": FREE_SUMMARY_LIMIT,
            "message": (
                f"You've reached the free quota of {FREE_SUMMARY_LIMIT} "
                "summaries. To keep getting summaries, enter your OpenAI API key"
            ),
        },
    )


def _reserve_free_summary(client_id: str) -> float:
    """Claim one free summary for ``client_id`` or raise the quota error.
    Reserved up front so concurrent requests can't overshoot the limit;
    call ``_release_free_summary`` with the returned token if the OpenAI call
    then fails."""
    now = time.time()
    with _free_summary_lock:
        while _server_key_calls and now - _server_key_calls[0] > 86400:
            _server_key_calls.popleft()
        last_hour = sum(1 for t in _server_key_calls if now - t <= 3600)
        if (
            len(_server_key_calls) >= SERVER_KEY_DAILY_LIMIT
            or last_hour >= SERVER_KEY_HOURLY_LIMIT
        ):
            raise _server_key_unavailable()
        used = _free_summary_counts.get(client_id, 0)
        if used >= FREE_SUMMARY_LIMIT:
            raise _free_quota_exceeded()
        _free_summary_counts[client_id] = used + 1
        _server_key_calls.append(now)
    return now


def _release_free_summary(client_id: str, token: float) -> None:
    with _free_summary_lock:
        used = _free_summary_counts.get(client_id, 0)
        if used > 0:
            _free_summary_counts[client_id] = used - 1
        try:
            _server_key_calls.remove(token)
        except ValueError:
            pass


def _free_summaries_remaining(client_id: str) -> int:
    with _free_summary_lock:
        return max(0, FREE_SUMMARY_LIMIT - _free_summary_counts.get(client_id, 0))

# uvicorn configures its handlers before importing this module; run again at
# startup (via lifespan) in case a different launcher sets logging up later.
# Lifespan, not add_event_handler/on_event — newer Starlette removed those.
_install_log_redaction()


@asynccontextmanager
async def _lifespan(_app: "FastAPI"):
    _install_log_redaction()
    yield


# Initialize FastAPI app
# Public API docs disabled in production (/docs, /redoc, /openapi.json).
app = FastAPI(
    title="Wikistar API",
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
    lifespan=_lifespan,
)

# Compress JSON responses larger than 1 KB. Big win for /demo and /results
# (multi-MB JSON payloads shrink ~70% on the wire).
app.add_middleware(GZipMiddleware, minimum_size=1024)

# Enable CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Add cache control middleware
@app.middleware("http")
async def add_cache_control_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Pragma"] = "no-cache"
    response.headers["Expires"] = "0"
    return response

# Prepare backend file directory.
# Mounted at /files so it doesn't collide with CRA's /static asset paths.
os.makedirs("visualizations", exist_ok=True)
app.mount(
    "/files",
    StaticFiles(directory="visualizations"),
    name="files"
)

# OpenAI models a request may pick. Keep in sync with SUPPORTED_MODELS in the
# frontend's src/models.ts, which renders this list as the model dropdown.
DEFAULT_MODEL = "gpt-5-mini"
SUPPORTED_MODELS = ("gpt-5-mini", "gpt-5-nano", "gpt-5", "gpt-4o-mini", "gpt-4o", "o3-mini")


def _resolve_model(requested: Optional[str]) -> str:
    """Return the requested model, DEFAULT_MODEL when none was sent, or a
    400 when it isn't one of SUPPORTED_MODELS."""
    model = (requested or "").strip() or DEFAULT_MODEL
    if model not in SUPPORTED_MODELS:
        raise HTTPException(
            status_code=400,
            detail=(
                f"Unsupported model '{model}'. "
                f"Choose one of: {', '.join(SUPPORTED_MODELS)}."
            ),
        )
    return model


# Pydantic models
class PreparePayload(BaseModel):
    article: str

class ClassifyPayload(BaseModel):
    article: str
    model: Optional[str] = None
    # repr=False keeps the key out of any printed/logged payload.
    api_key: Optional[str] = Field(default=None, repr=False)


def _resolve_api_key(
    supplied: Optional[str], allow_secret_fallback: bool = True
) -> str:
    """Resolve the OpenAI key to use for a request.
    When ``allow_secret_fallback`` is True (the demo's summary feature), an
    empty supplied key falls back to the server's ``OPENAI_API_KEY`` secret.
    When False (the user-driven classification pipeline and summaries on a
    generated page), only the user-supplied key is accepted — the server
    secret is never used, so the caller must provide their own key.
    """
    supplied_key = (supplied or "").strip()
    if allow_secret_fallback:
        key = supplied_key or os.environ.get("OPENAI_API_KEY", "").strip()
    else:
        key = supplied_key
    if not key:
        if allow_secret_fallback:
            detail = "No OpenAI API key available — none supplied and OPENAI_API_KEY is unset."
        else:
            detail = "An OpenAI API key is required for this request. Please provide your own key."
        raise HTTPException(status_code=400, detail=detail)
    return key

@app.post("/prepare")
async def prepare_article(payload: PreparePayload) -> Dict[str, Any]:
    df, title = fetch_revision_history(payload.article)
    if df is None:
        raise HTTPException(status_code=404, detail="Article not found or no revisions")

    df["Timestamp"] = pd.to_datetime(df["Timestamp"])
    df = df.sort_values("Timestamp").reset_index(drop=True)

    safe_title = title.replace(" ", "_")

    sections_df = add_flexible_revision_linking(
        df,
        csv_filename=f"{title}.csv",
        title_threshold=0.8,
        content_threshold=0.99,
        lookback_window=1000,
    )

    csv_filename = f"{safe_title}_sections.csv"
    csv_path = os.path.join("visualizations", csv_filename)
    sections_df.to_csv(csv_path, index=False)

    diffs = []
    for _, row in sections_df.iterrows():
        url = row.get("Flexible_Revision_URL", "")
        if not isinstance(url, str) or url == "":
            continue
        diffs.append({
            "url": url,
            "section_before": row.get("Flexible_Previous_Section", "") or "",
            "section_after": row["Section"],
        })
    json_filename = f"{safe_title}_diffs.json"
    json_path = os.path.join("visualizations", json_filename)
    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(diffs, f, indent=2, ensure_ascii=False)

    preview = sections_df.head(5).to_dict(orient="records")
    revision_count = len(df)

    return {
        "previewRows": preview,
        "csvUrl": f"/files/{csv_filename}",
        "diffs": diffs,
        "revisionCount": revision_count,
        "estimate": _estimate_classification(sections_df, title),
    }


# Rough tokens-per-char ratio for English prose under OpenAI tokenizers.
_CHARS_PER_TOKEN = 4
# Per-line JSONL overhead around each prompt in the Batch API input file
# (custom_id, method, url, model, message wrapper).
_BATCH_LINE_OVERHEAD_BYTES = 200


def _estimate_classification(sections_df: pd.DataFrame, title: str) -> Dict[str, Any]:
    """Build the same prompts /classify would send and size them, so the
    frontend can show a cost/time warning before the user commits."""
    taxonomy = create_taxonomy("Proccessor/taxonomy.csv")
    taxonomy_first = create_taxonomy("Proccessor/taxonomy_first_time.csv")
    _, prompts = generate_prompts_from_flexible_linking(
        sections_df.copy(), title, taxonomy, taxonomy_first
    )
    input_chars = sum(len(p) for p in prompts)
    batch_bytes = sum(
        len(json.dumps(p)) + _BATCH_LINE_OVERHEAD_BYTES for p in prompts
    )
    return {
        "edits": len(prompts),
        "inputTokens": input_chars // _CHARS_PER_TOKEN,
        "batchBytes": batch_bytes,
    }

@app.post("/classify")
async def classify_article(payload: ClassifyPayload, background_tasks: BackgroundTasks) -> Dict[str, Any]:
    """
    Start classification in the background and return immediately.
    Uses the flexible-linker pipeline + OpenAI Batch API (single batch).
    """
    # Classification always runs on the user's own key.
    api_key = _resolve_api_key(payload.api_key, allow_secret_fallback=False)
    model_name = _resolve_model(payload.model)

    df, title = fetch_revision_history(payload.article)
    if df is None:
        raise HTTPException(status_code=404, detail="Article not found or no revisions")

    df["Timestamp"] = pd.to_datetime(df["Timestamp"])
    df = df.sort_values("Timestamp").reset_index(drop=True)

    taxonomy = create_taxonomy("Proccessor/taxonomy.csv")
    taxonomy_first = create_taxonomy("Proccessor/taxonomy_first_time.csv")

    linked_df = add_flexible_revision_linking(
        df,
        csv_filename=f"{title}.csv",
        title_threshold=0.8,
        content_threshold=0.99,
        lookback_window=1000,
    )
    prompted_df, prompts = generate_prompts_from_flexible_linking(
        linked_df, title, taxonomy, taxonomy_first
    )

    job_id = _new_job_id()
    # The job ID in the filename keeps output files under /files unguessable;
    # the sanitized title just makes downloads readable.
    file_title = re.sub(r"[^A-Za-z0-9_-]+", "_", title).strip("_")[:80] or "article"
    total = len(prompts)
    with progress_lock:
        progress_map[job_id] = {
            "title": title,
            "file_prefix": f"{file_title}_{job_id}",
            "total": total,
            "processed": 0,
            "stage": "submitting",
            "started_at": time.time(),
            "updated_at": time.time(),
            "batch": None,
        }
    print(f"Initialized progress tracking for {title} (job {job_id}): 0/{total}")

    background_tasks.add_task(
        process_classification,
        job_id=job_id,
        model_name=model_name,
        api_key=api_key,
        prompts=prompts,
        df=prompted_df,
        title=title,
        taxonomy=taxonomy,
    )

    return {
        "status": "processing",
        "jobId": job_id,
        "message": (
            f"Classification of {payload.article} started in the background "
            f"(model={model_name}). Check progress with the /progress endpoint."
        ),
    }


def compute_grouped_idx(results_df: pd.DataFrame) -> pd.Series:
    """
    Group rows into linear chains along the flexible-linker's row-to-row
    pointers: an edge from a row to the earlier row whose (Revision ID,
    Section) matches its (Flexible_Previous_Revision_ID,
    Flexible_Previous_Section). Returns an integer Series aligned with
    `results_df`'s original index.
    """
    from recompute_grouped_idx import recompute

    if len(results_df) == 0:
        return pd.Series([], index=results_df.index, dtype=int)

    working = results_df.reset_index(drop=False).rename(
        columns={"index": "_orig_idx"}
    )
    grouped = recompute(working)
    return (
        grouped.set_index("_orig_idx")["grouped_idx"]
        .reindex(results_df.index)
        .astype(int)
    )


def process_classification(job_id, model_name, api_key, prompts, df, title, taxonomy):
    """Run the OpenAI batch classification in the background."""
    total = len(prompts)
    with progress_lock:
        job = dict(progress_map[job_id])

    def _update_batch_status(batch_info: dict) -> None:
        completed = batch_info.get("completed", 0) or 0
        _set_job_fields(
            job_id,
            batch=batch_info,
            stage=batch_info.get("status") or "running",
            processed=min(completed, total),
        )

    try:
        import matplotlib
        matplotlib.use('Agg')

        from Model.GPT import GPT

        print(f"Starting background processing for {title} with {total} prompts (model={model_name})")
        _set_job_fields(job_id, stage="submitting")

        model = GPT(model_name=model_name, api_key=api_key)
        raw_responses = model.predict(prompts, status_callback=_update_batch_status)

        _set_job_fields(job_id, stage="parsing")

        labels_col = [None] * len(df)
        explanation_col = [None] * len(df)
        parsed_ok = 0
        for i, response in enumerate(raw_responses):
            parsed = parse_prompt(response, taxonomy)
            if parsed is None:
                print(f"[warn] Failed to parse response at row {i}")
                continue
            labels_col[i] = parsed["Labels"]
            explanation_col[i] = parsed["Explanation"]
            parsed_ok += 1
        print(f"Parsed {parsed_ok}/{total} responses successfully.")

        results_df = df.copy()
        results_df["Labels"] = labels_col
        results_df["Explanation"] = explanation_col
        results_df["grouped_idx"] = compute_grouped_idx(results_df)

        cols = list(results_df.columns)
        cols.remove("Labels")
        cols.remove("Explanation")
        insert_after = "Changed Content" if "Changed Content" in cols else cols[-1]
        insert_at = cols.index(insert_after) + 1
        cols = cols[:insert_at] + ["Labels", "Explanation"] + cols[insert_at:]
        results_df = results_df[cols]

        result_filename = _job_file(job, "results.csv")
        result_path = os.path.join("visualizations", result_filename)
        results_df.to_csv(result_path, index=False)

        # Stash the full DataFrame so /summarize can look up Explanations
        # for any (label, section, period) cell the user clicks.
        _store_job_dataset(job_id, results_df)

        # Generate the page-evolution summary now while we still have the
        # user's API key in scope. Persist alongside the results CSV.
        evolution_filename = _job_file(job, "evolution.txt")
        evolution_path = os.path.join("visualizations", evolution_filename)
        try:
            _set_job_fields(job_id, stage="summarizing")
            evolution_text = generate_evolution_summary(
                results_df, api_key=api_key, model=model_name
            )
            if evolution_text:
                with open(evolution_path, "w", encoding="utf-8") as f:
                    f.write(evolution_text)
        except Exception as e:
            # Don't fail the whole classification if the summary errors.
            print(f"[warn] evolution summary failed: {_redact_secrets(e)}")

        print("Creating visualizations...")
        taxonomy[11] = {
            "category": "None of the above",
            "description": "No meaningful scientific changes were detected between the revisions; applies to non-scientific, formatting, or trivial edits."
        }

        dist_filename = _job_file(job, "distribution.png")
        dist_path = os.path.join("visualizations", dist_filename)
        plot_label_distribution(results_df, taxonomy, output_path=dist_path)

        cooc_filename = _job_file(job, "cooccurrence.png")
        cooc_path = os.path.join("visualizations", cooc_filename)
        plot_label_cooccurrence(results_df, taxonomy, output_path=cooc_path)

        _set_job_fields(job_id, processed=total, stage="complete")

        print(f"Background processing completed for {title} (job {job_id})")
    except Exception as e:
        print(f"Error in background processing: {_redact_secrets(e)}")
        _set_job_fields(job_id, stage="error", error=_redact_secrets(e))
        print(_redact_secrets(traceback.format_exc()))


@app.get("/results")
async def get_results(job_id: str) -> Dict[str, Any]:
    """
    Get the results of a completed classification job.
    """
    job = _get_job(job_id)
    if job.get("stage") == "error":
        return {"status": "error", "message": job.get("error") or "Classification failed."}
    # The background task marks the job complete only after every output
    # file below has been written.
    if job.get("stage") != "complete":
        return {
            "status": "processing",
            "message": "Results not available yet. Check progress with the /progress endpoint."
        }

    result_filename = _job_file(job, "results.csv")
    result_path = os.path.join("visualizations", result_filename)
    dist_filename = _job_file(job, "distribution.png")
    cooc_filename = _job_file(job, "cooccurrence.png")

    # Load results
    results_df = pd.read_csv(result_path)
    _store_job_dataset(job_id, results_df)
    preview_rows = results_df.head(5).to_dict(orient="records")

    # URL paths for frontend
    csv_url = f"/files/{result_filename}"
    distribution_url = f"/files/{dist_filename}"
    cooccurrence_url = f"/files/{cooc_filename}"

    # Optional: page-evolution summary (label 1 + 6 across years).
    evolution_path = os.path.join("visualizations", _job_file(job, "evolution.txt"))
    evolution_text: Optional[str] = None
    if os.path.exists(evolution_path):
        try:
            with open(evolution_path, "r", encoding="utf-8") as f:
                evolution_text = f.read().strip() or None
        except Exception:
            evolution_text = None

    return {
        "status": "complete",
        "previewRows": preview_rows,
        "csvUrl": csv_url,
        "results": results_df.to_dict(orient="records"),
        "distributionUrl": distribution_url,
        "cooccurrenceUrl": cooccurrence_url,
        "evolutionSummary": evolution_text,
    }

DEMOS = {
    "CRISPR": {
        "csv": "demo/crispr_results.csv",
        "title": "CRISPR",
        "evolution": "demo/crispr_evolution.txt",
    },
    "Artificial intelligence": {
        "csv": "demo/artificial_intelligence_results.csv",
        "title": "Artificial intelligence",
        "evolution": "demo/artificial_intelligence_evolution.txt",
    },
    "Ising model": {
        "csv": "demo/ising_model_results.csv",
        "title": "Ising model",
        "evolution": "demo/ising_model_evolution.txt",
    },
    "Vaccine": {
        "csv": "demo/vaccine_results.csv",
        "title": "Vaccine",
        "evolution": "demo/vaccine_evolution.txt",
    },
    "Chaos theory": {
        "csv": "demo/chaos_theory_results.csv",
        "title": "Chaos theory",
        "evolution": "demo/chaos_theory_evolution.txt",
    },
    "Natural language processing": {
        "csv": "demo/natural_language_processing_results.csv",
        "title": "Natural language processing",
        "evolution": "demo/natural_language_processing_evolution.txt",
    },
}

# Full per-article DataFrames retained in memory so the summarize endpoint
# can look up Explanations for any (label, section, period) cell. Distinct
# from `_demo_cache`, which holds the slim JSON payload sent to the browser.
_dataset_cache: Dict[str, "pd.DataFrame"] = {}
_dataset_cache_lock = threading.Lock()


def _store_dataset(article: str, df: "pd.DataFrame") -> None:
    with _dataset_cache_lock:
        _dataset_cache[article] = df


def _get_dataset(article: str) -> Optional["pd.DataFrame"]:
    with _dataset_cache_lock:
        return _dataset_cache.get(article)


# Results of live classification runs, keyed by job ID so /summarize only
# reads a run's explanations for a caller holding its ID. Bounded LRU — an
# evicted run is reloaded from its results CSV on demand.
MAX_JOB_DATASETS = 20
_job_datasets: "OrderedDict[str, pd.DataFrame]" = OrderedDict()


def _store_job_dataset(job_id: str, df: "pd.DataFrame") -> None:
    with _dataset_cache_lock:
        _job_datasets[job_id] = df
        _job_datasets.move_to_end(job_id)
        while len(_job_datasets) > MAX_JOB_DATASETS:
            _job_datasets.popitem(last=False)


def _get_job_dataset(job_id: str) -> Optional["pd.DataFrame"]:
    job = _get_job(job_id)
    with _dataset_cache_lock:
        df = _job_datasets.get(job_id)
        if df is not None:
            _job_datasets.move_to_end(job_id)
            return df
    if job.get("stage") != "complete":
        return None
    result_path = os.path.join("visualizations", _job_file(job, "results.csv"))
    if not os.path.exists(result_path):
        return None
    df = pd.read_csv(result_path)
    _store_job_dataset(job_id, df)
    return df


# Bundled demo DataFrames only. Kept apart from `_dataset_cache` so
# server-key summaries always read explanations from the shipped CSV.
_demo_datasets: Dict[str, "pd.DataFrame"] = {}


def _get_demo_dataset(article: str) -> Optional["pd.DataFrame"]:
    demo = DEMOS.get(article)
    if demo is None:
        return None
    with _dataset_cache_lock:
        df = _demo_datasets.get(article)
    if df is None and os.path.exists(demo["csv"]):
        _build_demo_payload(demo)
        with _dataset_cache_lock:
            df = _demo_datasets.get(article)
    return df

# Columns shipped to the browser as the bulk `results` array. The Edits
# table reads the heavy text columns directly from these rows; charts only
# touch the slim ones. Hosted demo CSVs are LFS-tracked, so size isn't a
# constraint here — we send everything.
CHART_COLUMNS = [
    "Section",
    "Timestamp",
    "Labels",
    "grouped_idx",
    "Level",
    "Revision ID",
    "User",
    "Flexible_Previous_Section",
    "Changed Content",
    "Flexible_Previous_Content",
    "Explanation",
]

# In-memory cache for prepared demo payloads. Keyed by article name; cleared
# on container restart. Reading + serializing the CSV is the slow part on
# repeat clicks, so caching the final dict is the biggest win.
_demo_cache: Dict[str, Dict[str, Any]] = {}
_demo_cache_lock = threading.Lock()


def _build_demo_payload(demo: Dict[str, str]) -> Dict[str, Any]:
    results_df = pd.read_csv(demo["csv"])
    # The bundled demo CSVs may lack grouped_idx (e.g. older offline pipelines);
    # compute it if absent so the frontend charts work uniformly.
    if "grouped_idx" not in results_df.columns:
        results_df["grouped_idx"] = compute_grouped_idx(results_df)

    # Stash the full DataFrame so /summarize can look up Explanations.
    _store_dataset(demo["title"], results_df)
    with _dataset_cache_lock:
        _demo_datasets[demo["title"]] = results_df

    preview_rows = results_df.head(5).to_dict(orient="records")

    # Slim the bulk array: keep only columns the charts actually read.
    chart_cols = [c for c in CHART_COLUMNS if c in results_df.columns]
    slim_df = results_df[chart_cols]

    # Static, pre-generated evolution summary if the demo entry provides one.
    evolution_text: Optional[str] = None
    evo_path = demo.get("evolution")
    if evo_path and os.path.exists(evo_path):
        try:
            with open(evo_path, "r", encoding="utf-8") as f:
                evolution_text = f.read().strip() or None
        except Exception:
            evolution_text = None

    return {
        "status": "complete",
        "title": demo["title"],
        "previewRows": preview_rows,
        "results": slim_df.to_dict(orient="records"),
        "evolutionSummary": evolution_text,
        "csvUrl": f"/demo/download?article={demo['title']}",
    }


@app.get("/demo")
async def get_demo(article: str) -> Dict[str, Any]:
    """
    Return a pre-classified results dataset packaged with the Space.
    Lets users see the interactive visualizations without running their own
    classification.
    """
    demo = DEMOS.get(article)
    if demo is None:
        raise HTTPException(status_code=404, detail=f"No demo available for '{article}'")
    if not os.path.exists(demo["csv"]):
        raise HTTPException(status_code=500, detail=f"Demo file missing: {demo['csv']}")

    with _demo_cache_lock:
        cached = _demo_cache.get(article)
    if cached is not None:
        return cached

    payload = _build_demo_payload(demo)
    with _demo_cache_lock:
        _demo_cache[article] = payload
    return payload


@app.get("/demo/download")
async def download_demo_csv(article: str) -> FileResponse:
    """Serve the bundled demo CSV so users can save and re-load it later."""
    demo = DEMOS.get(article)
    if demo is None:
        raise HTTPException(status_code=404, detail=f"No demo available for '{article}'")
    if not os.path.exists(demo["csv"]):
        raise HTTPException(status_code=500, detail=f"Demo file missing: {demo['csv']}")
    return FileResponse(
        demo["csv"],
        media_type="text/csv",
        filename=f"{demo['title']}_classified_revisions.csv",
    )


# ---------------------------------------------------------------------------
# Summarize endpoint: small OpenAI call over the Explanation column for the
# rows behind a clicked chart cell or line-chart point.
# ---------------------------------------------------------------------------

class SummarizePayload(BaseModel):
    article: str
    label: int
    # Combined multi-label summary: when set (length > 1), rows are matched if
    # they carry ANY of these labels, and every covered label's explanation
    # snippet is summarized together into a single write-up. Sent by the
    # by-section heatmap's "All scientific edit types" cell clicks. `label`
    # still holds labels[0] for backward-compatible single-label callers.
    labels: Optional[List[int]] = None
    period: str  # one of "Y" | "2Q" | "Q" — matches the frontend Period type
    # e.g. "2018", "2018Q3", "2018H1". Omit / send null to summarize across
    # every period (the "Summarize all periods" affordance in Over-Time).
    period_key: Optional[str] = None
    section_ids: Optional[List[int]] = None  # grouped_idx values; None = all sections
    api_key: Optional[str] = Field(default=None, repr=False)
    model: Optional[str] = None
    # Ignored — kept so older frontends don't fail validation. The server
    # decides when its own key may be used (see summarize_edits).
    use_server_key: bool = False
    # When the dataset lives only in the user's browser (CSV they uploaded),
    # the frontend filters rows itself and sends just the matching explanation
    # snippets here. If provided, the server skips its own dataset lookup.
    explanations: Optional[List[str]] = None
    # Live classification run to summarize, as returned by /classify. Without
    # it, only the built-in demo datasets can be looked up server-side.
    job_id: Optional[str] = None
    # Set by the panel's "Regenerate" button: skip the summary cache lookup and
    # overwrite the stored entry with the fresh result.
    force_refresh: bool = False


def _period_key(timestamp: pd.Timestamp, period: str) -> str:
    """Mirror of the frontend's periodKey() in taxonomy.ts."""
    if pd.isna(timestamp):
        return ""
    if not isinstance(timestamp, pd.Timestamp):
        timestamp = pd.Timestamp(timestamp)
    if timestamp.tzinfo is None:
        timestamp = timestamp.tz_localize("UTC")
    else:
        timestamp = timestamp.tz_convert("UTC")
    y = timestamp.year
    m = timestamp.month - 1
    if period == "Y":
        return str(y)
    q = m // 3
    if period == "Q":
        return f"{y}Q{q + 1}"
    # half-year ("2Q")
    return f"{y}H1" if q < 2 else f"{y}H2"


def _parse_labels_field(raw: Any) -> List[int]:
    if raw is None or (isinstance(raw, float) and pd.isna(raw)) or raw == "":
        return []
    if isinstance(raw, list):
        return [int(x) for x in raw if str(x).strip().lstrip("-").isdigit()]
    s = str(raw).strip()
    if not s:
        return []
    try:
        import ast as _ast
        parsed = _ast.literal_eval(s.replace("'", '"'))
        if isinstance(parsed, list):
            return [int(x) for x in parsed if str(x).strip().lstrip("-").isdigit()]
    except Exception:
        pass
    inner = s.strip("[]")
    out: List[int] = []
    for tok in inner.split(","):
        tok = tok.strip()
        if tok and tok.lstrip("-").isdigit():
            out.append(int(tok))
    return out


def _explanation_for_label(raw: Any, label: int) -> Optional[str]:
    """Explanation column stores a dict-like JSON string keyed by label."""
    if raw is None or (isinstance(raw, float) and pd.isna(raw)) or raw == "":
        return None
    s = str(raw).strip()
    if not s:
        return None
    try:
        import ast as _ast
        parsed = _ast.literal_eval(s)
        if isinstance(parsed, dict):
            # Keys may be int or str depending on how it was serialized.
            for k in (label, str(label)):
                if k in parsed:
                    return str(parsed[k])
    except Exception:
        pass
    return None


# Labels considered "scientifically significant" for the evolution summary.
EVOLUTION_LABELS = {1, 6}
# Years with fewer than this many matching edits are dropped from the prompt.
EVOLUTION_MIN_PER_YEAR = 5
# Cap explanations sent per year to keep the prompt bounded on huge corpora.
EVOLUTION_MAX_PER_YEAR = 100


# Lazily-loaded taxonomy definitions, keyed by label number. The taxonomy CSV
# is bundled with the Space; if it's missing we fall back to the label name.
_TAXONOMY_DEFS: Optional[Dict[int, str]] = None


def _label_definition(label: int, article: Optional[str] = None) -> str:
    global _TAXONOMY_DEFS
    if _TAXONOMY_DEFS is None:
        try:
            tax = create_taxonomy(os.path.join("Proccessor", "taxonomy.csv"))
            _TAXONOMY_DEFS = {
                int(k): str(v.get("definition", "")).strip() for k, v in tax.items()
            }
        except Exception:
            _TAXONOMY_DEFS = {}
    definition = _TAXONOMY_DEFS.get(int(label), "(definition unavailable)")
    # Definitions for labels 1 and 4 contain a {PAGE_NAME} template token that
    # mirrors what the classifier saw at inference time. Substitute the actual
    # article so the summary prompt never surfaces the raw placeholder.
    # TODO: switch this fallback off "Artificial intelligence" once every caller
    # reliably passes the article name.
    return definition.replace("{PAGE_NAME}", article or "Artificial intelligence")


def generate_evolution_summary(
    df: "pd.DataFrame", api_key: str, model: str = DEFAULT_MODEL
) -> Optional[str]:
    """
    Build a prose paragraph describing scientifically significant edits across
    years, grouped by year. Uses rows whose Labels include 1 (New Scientific
    Information) or 6 (Change in Scientific Narrative). Drops sparse years
    (< EVOLUTION_MIN_PER_YEAR matches) and caps very dense years.
    Returns None if there's nothing to summarize, or raises on OpenAI errors.
    """
    if "Labels" not in df.columns or "Timestamp" not in df.columns:
        return None
    if "Explanation" not in df.columns:
        return None

    # Pre-parse once.
    timestamps = pd.to_datetime(df["Timestamp"], utc=True, errors="coerce")
    parsed_labels = df["Labels"].apply(_parse_labels_field)
    has_section = "Section" in df.columns

    # Group explanations by year, only for rows touching label 1 or 6.
    by_year: Dict[int, List[str]] = {}
    for idx, ts in timestamps.items():
        if pd.isna(ts):
            continue
        if has_section and _is_hidden_section(df.at[idx, "Section"]):
            continue
        labels = parsed_labels.iloc[idx] if hasattr(parsed_labels, "iloc") else parsed_labels[idx]
        relevant = EVOLUTION_LABELS.intersection(labels)
        if not relevant:
            continue
        raw = df.at[idx, "Explanation"]
        snippets: List[str] = []
        for label in relevant:
            snippet = _explanation_for_label(raw, label)
            if snippet:
                snippets.append(f"(label {label}) {snippet}")
        if not snippets:
            continue
        year = int(ts.year)
        by_year.setdefault(year, []).extend(snippets)

    if not by_year:
        return None

    # Drop sparse years and cap dense ones.
    kept_years = sorted(y for y, items in by_year.items() if len(items) >= EVOLUTION_MIN_PER_YEAR)
    if not kept_years:
        return None

    # Build a year-headered prompt.
    lines: List[str] = []
    total_used = 0
    for year in kept_years:
        items = by_year[year][:EVOLUTION_MAX_PER_YEAR]
        total_used += len(items)
        lines.append(f"\n## {year} ({len(by_year[year])} matching edits)")
        for item in items:
            lines.append(f"- {item}")
    grouped_text = "\n".join(lines)

    user_prompt = (
        "You are summarizing the scientific evolution of a Wikipedia article "
        "based on per-revision edit explanations. The explanations below are "
        "grouped by year, drawn only from edits classified as introducing "
        "new scientific information (label 1) or changing the scientific "
        "narrative (label 6).\n\n"
        "Write a short markdown summary (about 4–6 short paragraphs, "
        "separated by blank lines) that traces how the article's scientific "
        "content evolved across years. The FIRST paragraph should be a "
        "1–2 sentence high-level overview so a reader sees the gist before "
        "expanding the rest. Each subsequent paragraph should cover a "
        "different era or theme. Highlight the most dominant scientific "
        "changes per year, weaving in concrete topics (e.g., specific "
        "discoveries, mechanisms, applications, or controversies) by year "
        "(e.g., \"In **2017**, findings about X were added\"). Mention years "
        "explicitly and wrap every year or year-range in markdown bold "
        "(**2017**, **2010–2012**). Do not list every edit — synthesize the "
        "patterns. Do not invent details that aren't in the explanations. "
        "Output markdown prose only — no bullet points and no headings.\n\n"
        "After the paragraphs, add a blank line and then a final line that "
        "begins exactly with 'TLDR: ' followed by a 1–2 sentence takeaway "
        "capturing the overall arc of the article's scientific evolution. "
        "The TLDR line must be the very last line of the output.\n\n"
        f"Explanations grouped by year ({total_used} total snippets):\n"
        f"{grouped_text}"
    )

    import openai
    client = openai.OpenAI(api_key=api_key)
    completion = client.chat.completions.create(
        model=model,
        store=True,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a careful research assistant that writes compact, "
                    "evidence-grounded prose summaries of how Wikipedia articles "
                    "evolved scientifically over time."
                ),
            },
            {"role": "user", "content": user_prompt},
        ],
    )
    text = completion.choices[0].message.content or ""
    _log_openai_call(
        endpoint="evolution",
        model=model,
        article=None,
        used_server_key=False,
        prompt_chars=len(user_prompt),
        response_chars=len(text),
        usage=_usage_dict(completion),
    )
    return text.strip() or None


def _usage_dict(completion: Any) -> Optional[Dict[str, Any]]:
    """Pull token counts out of a chat completion response, if present."""
    usage = getattr(completion, "usage", None)
    if usage is None:
        return None
    try:
        return {
            "prompt_tokens": getattr(usage, "prompt_tokens", None),
            "completion_tokens": getattr(usage, "completion_tokens", None),
            "total_tokens": getattr(usage, "total_tokens", None),
        }
    except Exception:
        return None


# ---------------------------------------------------------------------------
# Summary cache: generated /summarize outputs keyed by a hash of the exact
# OpenAI request (model + system prompt + user prompt). The user prompt embeds
# every matched explanation snippet, so any request that would make the same
# call is served from the DB, and a changed dataset or prompt template can
# never hit a stale entry. Precomputed default-view summaries are committed as
# SUMMARY_SEED_PATH (JSONL — plain text, so no LFS) by
# scripts/precompute_summaries.py and merged into the runtime DB on startup.
# ---------------------------------------------------------------------------

# Default summary model. The server's own key only ever runs models in
# SERVER_KEY_MODELS; a user-supplied key may pick any of SUPPORTED_MODELS.
SUMMARY_MODEL = DEFAULT_MODEL
SERVER_KEY_MODELS = {DEFAULT_MODEL}
# Older seed entries were generated with these models. Cached text costs
# nothing to serve, so lookups fall back to them before calling OpenAI.
CACHE_FALLBACK_MODELS = ("gpt-5.4",)
SUMMARY_SYSTEM_PROMPT = (
    "You are a careful research assistant that summarizes "
    "lists of Wikipedia edit explanations into compact, "
    "high-signal prose."
)
SUMMARY_SEED_PATH = os.path.join("demo", "summary_cache.jsonl")


def _default_summary_db_path() -> str:
    env_path = os.environ.get("SUMMARY_DB_PATH", "").strip()
    if env_path:
        return env_path
    # HF persistent storage (when enabled) survives restarts; otherwise the
    # runtime DB lives with the other generated artifacts and is rebuilt from
    # the committed seed on each boot.
    if os.path.isdir("/data") and os.access("/data", os.W_OK):
        return "/data/summary_cache.sqlite"
    return os.path.join("visualizations", "summary_cache.sqlite")


SUMMARY_DB_PATH = _default_summary_db_path()
_summary_db_lock = threading.Lock()


def _summary_db() -> sqlite3.Connection:
    return sqlite3.connect(SUMMARY_DB_PATH, timeout=10)


def _init_summary_db() -> None:
    """Create the summaries table and load the committed seed into it."""
    try:
        with _summary_db_lock, _summary_db() as conn:
            conn.execute(
                "CREATE TABLE IF NOT EXISTS summaries ("
                " cache_key TEXT PRIMARY KEY,"
                " article TEXT,"
                " labels TEXT,"
                " period TEXT,"
                " period_key TEXT,"
                " section_count INTEGER,"
                " model TEXT,"
                " summary TEXT NOT NULL,"
                " created_at TEXT)"
            )
            if not os.path.exists(SUMMARY_SEED_PATH):
                return
            loaded = 0
            with open(SUMMARY_SEED_PATH, "r", encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        rec = json.loads(line)
                    except Exception:
                        continue
                    _upsert_summary_row(conn, rec)
                    loaded += 1
            print(f"Loaded {loaded} precomputed summaries into {SUMMARY_DB_PATH}")
    except Exception as e:  # pragma: no cover - cache must never block startup
        print(f"[warn] summary cache init failed: {e}")


def _upsert_summary_row(conn: sqlite3.Connection, rec: Dict[str, Any]) -> None:
    conn.execute(
        "INSERT OR REPLACE INTO summaries (cache_key, article, labels, period,"
        " period_key, section_count, model, summary, created_at)"
        " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (
            rec["cache_key"],
            rec.get("article"),
            json.dumps(rec.get("labels")),
            rec.get("period"),
            rec.get("period_key"),
            rec.get("section_count"),
            rec.get("model"),
            rec["summary"],
            rec.get("created_at"),
        ),
    )


def _summary_cache_key(user_prompt: str, model: str = SUMMARY_MODEL) -> str:
    blob = json.dumps([model, SUMMARY_SYSTEM_PROMPT, user_prompt], ensure_ascii=False)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()


def get_cached_summary(cache_key: str) -> Optional[str]:
    try:
        with _summary_db_lock, _summary_db() as conn:
            row = conn.execute(
                "SELECT summary FROM summaries WHERE cache_key = ?", (cache_key,)
            ).fetchone()
        return row[0] if row else None
    except Exception as e:
        print(f"[warn] summary cache read failed: {e}")
        return None


def lookup_summary(job: "SummaryJob") -> Optional[str]:
    """Cached summary for this job under its own model, else under any
    CACHE_FALLBACK_MODELS entry."""
    cached = get_cached_summary(job.cache_key)
    if cached is not None:
        return cached
    for model in CACHE_FALLBACK_MODELS:
        if model == job.model:
            continue
        cached = get_cached_summary(_summary_cache_key(job.user_prompt, model))
        if cached is not None:
            return cached
    return None


def store_cached_summary(job: "SummaryJob", summary: str) -> Optional[Dict[str, Any]]:
    """Upsert a generated summary. Returns the stored record (also the seed
    JSONL line format), or None if the write failed."""
    rec = {
        "cache_key": job.cache_key,
        "article": job.article,
        "labels": job.labels,
        "period": job.period,
        "period_key": job.period_key,
        "section_count": job.section_count,
        "model": job.model,
        "summary": summary,
        "created_at": _utc_now_iso(),
    }
    try:
        with _summary_db_lock, _summary_db() as conn:
            _upsert_summary_row(conn, rec)
        return rec
    except Exception as e:
        print(f"[warn] summary cache write failed: {e}")
        return None


_init_summary_db()


@dataclass
class SummaryJob:
    """Everything needed to generate (or look up) one summary."""
    user_prompt: str
    model: str
    cache_key: str
    article: str
    labels: List[int]
    period: str
    period_key: Optional[str]
    section_count: Optional[int]
    snippet_count: int
    client_supplied: bool
    # Response fields other than "summary" / "cached".
    response: Dict[str, Any]


def prepare_summary(
    payload: SummarizePayload,
    server_key: bool = False,
    model: str = SUMMARY_MODEL,
) -> Union[Dict[str, Any], SummaryJob]:
    """
    Filter the rows behind a (label, period, section-group) request and build
    the summary prompt. Returns a finished response dict when there is nothing
    to summarize, otherwise a SummaryJob. With ``server_key`` the rows always
    come from the bundled demo CSV and client explanations are ignored.
    """
    # Normalize the label set. A combined request (by-section "All edit types"
    # cell) carries multiple labels; a single-label request carries just one.
    # `combined_labels` is the deduped, sorted set we match/summarize over;
    # `is_combined` gates the multi-label prompt and response labelling.
    if payload.labels:
        combined_labels = sorted({int(l) for l in payload.labels})
    else:
        combined_labels = [int(payload.label)]
    is_combined = len(combined_labels) > 1

    # Client-supplied path: the dataset is in the user's browser, not cached
    # on the server. The frontend has already filtered rows by (label, period,
    # sections) and sent the relevant explanation snippets. Skip the server
    # dataset lookup entirely.
    client_snippets: Optional[List[str]] = None
    matched_count: Optional[int] = None
    if payload.explanations is not None and not server_key:
        client_snippets = [s for s in payload.explanations if s and s.strip()]
        matched_count = len(payload.explanations)
        if not client_snippets:
            return {
                "summary": None,
                "edit_count": matched_count,
                "explanation_count": 0,
                "message": "Matched edits had no explanation text to summarize.",
            }

    df = None
    if server_key:
        df = _get_demo_dataset(payload.article)
        if df is None:
            raise HTTPException(
                status_code=404,
                detail=f"No demo dataset for '{payload.article}'.",
            )
    elif client_snippets is None and payload.job_id:
        df = _get_job_dataset(payload.job_id)
        if df is None:
            raise HTTPException(
                status_code=404,
                detail="No results loaded for this classification run.",
            )
    elif client_snippets is None:
        df = _get_dataset(payload.article)
        if df is None:
            # Lazy-load the demo dataset if its cache got cleared (e.g. container restart).
            demo = DEMOS.get(payload.article)
            if demo is not None and os.path.exists(demo["csv"]):
                _build_demo_payload(demo)
                df = _get_dataset(payload.article)
        if df is None:
            raise HTTPException(
                status_code=404,
                detail=f"No dataset loaded for '{payload.article}'.",
            )
        if "Explanation" not in df.columns:
            raise HTTPException(
                status_code=400,
                detail="This dataset has no Explanation column to summarize.",
            )

    if client_snippets is not None:
        snippets = client_snippets
        # We don't know the unmatched count on the client path; use the count
        # the client reported. On this path each reported snippet is one
        # non-empty explanation, so the "with explanation text" count is just
        # the snippet count.
        match_count_for_response = matched_count or len(snippets)
        rows_with_expl = len(snippets)
    else:
        # Filter to rows that contain any of the requested label(s). For a
        # single-label request this is just `payload.label`; for a combined
        # request it's any of the covered labels.
        label_set = set(combined_labels)
        mask = df["Labels"].apply(
            lambda v: bool(label_set.intersection(_parse_labels_field(v)))
        )

        # Never summarize hidden structural sections, even if the union-find
        # grouped_idx lumped one in with a visible section the user selected.
        if "Section" in df.columns:
            mask = mask & ~df["Section"].apply(_is_hidden_section)

        # Filter by period bucket — skipped when period_key is None
        # ("Summarize all periods" affordance).
        if payload.period_key is not None:
            timestamps = pd.to_datetime(df["Timestamp"], utc=True, errors="coerce")
            period_mask = timestamps.apply(
                lambda t: _period_key(t, payload.period) == payload.period_key
            )
            mask = mask & period_mask

        # Filter by section group(s) if provided.
        if payload.section_ids is not None and "grouped_idx" in df.columns:
            section_set = set(int(s) for s in payload.section_ids)
            mask = mask & df["grouped_idx"].apply(
                lambda v: int(v) in section_set if pd.notna(v) else False
            )

        matched = df.loc[mask]
        if matched.empty:
            return {
                "summary": None,
                "edit_count": 0,
                "message": (
                    "No edits found for this label and section selection."
                    if payload.period_key is None
                    else "No edits found for this cell."
                ),
            }

        # Pull the relevant Explanation snippet(s) for each matched row. For a
        # combined request, gather every covered label's snippet the row
        # carries so the summary spans all edit types in the cell.
        snippets = []
        rows_with_expl = 0
        for _, row in matched.iterrows():
            raw_expl = row.get("Explanation")
            row_had_snippet = False
            for lbl in combined_labels:
                text = _explanation_for_label(raw_expl, lbl)
                if text:
                    snippets.append(text)
                    row_had_snippet = True
            if row_had_snippet:
                rows_with_expl += 1

        if not snippets:
            return {
                "summary": None,
                "edit_count": int(len(matched)),
                "message": "Matched edits had no explanation text to summarize.",
            }
        match_count_for_response = int(len(matched))

    # Send every matched snippet to the model. Capping was previously used to
    # bound latency / cost but it skewed summaries toward the earliest rows
    # (especially noticeable with "Summarize all periods"). Modern context
    # windows comfortably fit thousands of short explanations.
    snippets_for_prompt = snippets
    truncated = False

    # Compose the prompt.
    label_name_map = {
        1: "New Scientific Information",
        2: "Scientific Information Removed",
        3: "Scientific Clarification Added",
        4: "Scientific Technical Terms Added",
        5: "Researcher Names Added",
        6: "Change in Scientific Narrative",
        7: "Academic References Added",
        8: "Academic References Removed",
        9: "Wikilink Added",
        10: "Addition or Modification of Quantitative Information",
        11: "None of the above",
    }
    label_name = label_name_map.get(payload.label, f"Label {payload.label}")
    label_definition = _label_definition(payload.label, payload.article)

    section_note = (
        "across all sections"
        if not payload.section_ids
        else f"across {len(payload.section_ids)} section group(s)"
    )

    period_note = (
        f"during period {payload.period_key}"
        if payload.period_key
        else "across the article's full revision history"
    )

    if is_combined:
        # Combined summary: one write-up spanning every scientific edit type
        # present in the cell. The scope is the union of the covered labels'
        # definitions rather than a single one, so the model synthesizes
        # across edit types instead of being told to omit "other labels".
        covered_names = [
            label_name_map.get(l, f"Label {l}") for l in combined_labels
        ]
        definitions_block = "\n\n".join(
            f"'{label_name_map.get(l, f'Label {l}')}' (label {l}):\n"
            f"{_label_definition(l, payload.article)}"
            for l in combined_labels
        )
        response_label_name = "All scientific edit types"
        user_prompt = (
            f"You are summarizing Wikipedia revision edits for the article "
            f"'{payload.article}'.\n\n"
            f"Below are explanations from {len(snippets_for_prompt)} edit "
            f"observations {period_note}, {section_note}. They span every "
            f"scientific edit type present in this selection: "
            f"{', '.join(covered_names)}.\n\n"
            f"Definitions of the covered edit types:\n{definitions_block}\n\n"
            f"Write a tight, scannable summary as 3–6 short bullets that "
            f"synthesize what changed across ALL of these edit types together. "
            f"Each bullet must:\n"
            f"- Start with '- '.\n"
            f"- Be one sentence, ideally under 20 words.\n"
            f"- Cover a distinct theme or grouping — do NOT restate the same "
            f"point twice and do NOT enumerate every edit.\n"
            f"- Lead with the substance, not filler (avoid openers like "
            f"'This edit…' or 'The article…').\n\n"
            f"Scope rules — read carefully:\n"
            f"- Summarize the edits as a whole. Group related changes across "
            f"edit types; you do not need one bullet per edit type.\n"
            f"- Only discuss evidence that satisfies at least one of the "
            f"definitions above. Ignore anything in the explanations that "
            f"falls outside all of them.\n\n"
            f"Formatting requirements:\n"
            f"- Use Markdown **bold** (double asterisks) only around concrete "
            f"evidence — e.g. the specific technical terms, references, "
            f"wikilinks, numeric values, or researcher names mentioned in the "
            f"explanations. Bold only the evidence phrase itself, not the "
            f"surrounding prose.\n"
            f"- After the bullets, add a separate final line beginning with "
            f"'Bottom line: ' that gives a one-sentence takeaway about what "
            f"these edits collectively did to the article during this period.\n"
            f"- Do not output any prose outside the bullets and the Bottom "
            f"line line. No headings, no intro sentence, no closing remark.\n\n"
            f"Explanations:\n"
            + "\n\n".join(f"- {s}" for s in snippets_for_prompt)
        )
    else:
        response_label_name = label_name
        user_prompt = (
            f"You are summarizing Wikipedia revision edits for the article "
            f"'{payload.article}'.\n\n"
            f"Below are explanations from {len(snippets_for_prompt)} edits "
            f"classified as '{label_name}' (label {payload.label}) {period_note}, "
            f"{section_note}.\n\n"
            f"Definition of '{label_name}':\n{label_definition}\n\n"
            f"Write a tight, scannable summary as 3–6 short bullets that "
            f"synthesize the patterns. Each bullet must:\n"
            f"- Start with '- '.\n"
            f"- Be one sentence, ideally under 20 words.\n"
            f"- Cover a distinct theme or grouping — do NOT restate the same "
            f"point twice and do NOT enumerate every edit.\n"
            f"- Lead with the substance, not filler (avoid openers like "
            f"'This edit…' or 'The article…').\n\n"
            f"Scope rules — read carefully:\n"
            f"- Only discuss evidence that directly satisfies the definition of "
            f"'{label_name}' given above. Do NOT mention edits or details that "
            f"appeared in the explanations but fall outside this definition "
            f"(e.g. for an 'Academic References Added' summary, do not mention "
            f"named-reference tags, formatting cleanup, or non-academic links — "
            f"the definition is specifically about new academic papers with "
            f"identifiers like DOI/PMID/PMC/arXiv/ISSN).\n"
            f"- If something feels noteworthy but doesn't match the definition, "
            f"omit it. It will be covered by a different label's summary.\n\n"
            f"Formatting requirements:\n"
            f"- Use Markdown **bold** (double asterisks) only around concrete "
            f"evidence that directly satisfies the definition above — e.g. the "
            f"specific technical terms, references, wikilinks, numeric values, "
            f"or researcher names mentioned in the explanations. Bold only the "
            f"evidence phrase itself, not the surrounding prose. Do not bold "
            f"anything that isn't itself an instance of the label.\n"
            f"- After the bullets, add a separate final line beginning with "
            f"'Bottom line: ' that gives a one-sentence takeaway about what "
            f"these edits collectively did to the article during this period.\n"
            f"- Do not output any prose outside the bullets and the Bottom line "
            f"line. No headings, no intro sentence, no closing remark.\n\n"
            f"Explanations:\n"
            + "\n\n".join(f"- {s}" for s in snippets_for_prompt)
        )

    return SummaryJob(
        user_prompt=user_prompt,
        model=model,
        cache_key=_summary_cache_key(user_prompt, model),
        article=payload.article,
        labels=combined_labels,
        period=payload.period,
        period_key=payload.period_key,
        section_count=len(payload.section_ids) if payload.section_ids else None,
        snippet_count=len(snippets_for_prompt),
        client_supplied=client_snippets is not None,
        response={
            "edit_count": match_count_for_response,
            # Rows that contributed at least one explanation snippet. In
            # combined mode a single row may contribute several snippets, so
            # this stays a per-row count rather than the raw snippet total.
            "explanation_count": rows_with_expl,
            "truncated": truncated,
            "label": payload.label,
            "labels": combined_labels if is_combined else None,
            "label_name": response_label_name,
            "period_key": payload.period_key,
        },
    )


def generate_summary(job: SummaryJob, api_key: str, used_server_key: bool) -> str:
    """Run the OpenAI call for a SummaryJob. Raises on OpenAI errors."""
    import openai
    client = openai.OpenAI(api_key=api_key)
    completion = client.chat.completions.create(
        model=job.model,
        store=True,
        messages=[
            {"role": "system", "content": SUMMARY_SYSTEM_PROMPT},
            {"role": "user", "content": job.user_prompt},
        ],
    )
    summary_text = completion.choices[0].message.content or ""
    _log_openai_call(
        endpoint="summarize",
        model=job.model,
        article=job.article,
        used_server_key=used_server_key,
        prompt_chars=len(job.user_prompt),
        response_chars=len(summary_text),
        usage=_usage_dict(completion),
        extra={
            "labels": job.labels,
            "period_key": job.period_key,
            "snippet_count": job.snippet_count,
            "client_supplied_explanations": job.client_supplied,
        },
    )
    return summary_text.strip()


@app.post("/summarize")
async def summarize_edits(
    payload: SummarizePayload,
    x_client_id: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
    """
    Summarize the Explanation text for all rows that share a (label, period,
    section-group) cell. Used by interactive chart drill-downs. Served from the
    summary cache when an identical request was generated before, unless
    `force_refresh` is set.
    """
    # The server — not the browser's `use_server_key` flag — decides when its
    # own key pays: only when no user key is supplied, for a built-in demo
    # article, on an allow-listed model, with explanations loaded from the
    # bundled CSV (prepare_summary ignores client-supplied ones).
    billed_server_key = not (payload.api_key or "").strip()
    model = _resolve_model(payload.model)
    if billed_server_key:
        if payload.article not in DEMOS:
            raise HTTPException(
                status_code=403,
                detail=(
                    "An OpenAI API key is required to summarize "
                    f"'{payload.article}'. Free summaries are only available "
                    "for the built-in demo articles."
                ),
            )
        if model not in SERVER_KEY_MODELS:
            raise HTTPException(
                status_code=403,
                detail=(
                    f"Free summaries run on {', '.join(sorted(SERVER_KEY_MODELS))} "
                    f"only. Supply your own OpenAI key to use '{model}'."
                ),
            )

    # Requests without the header (non-browser callers) share one bucket.
    client_id = (x_client_id or "").strip()[:128] or "anonymous"
    # Reported back so the panel can show how many free summaries are left.
    quota_info: Dict[str, Any] = {}

    job = prepare_summary(payload, server_key=billed_server_key, model=model)
    if isinstance(job, dict):
        return job

    # TEMP (revert after cache testing): log every summary cache hit/miss.
    _cache_log_ctx = (
        f"article={job.article!r} labels={job.labels} period={job.period}"
        f" period_key={job.period_key} sections={job.section_count}"
        f" snippets={job.snippet_count} model={job.model}"
        f" server_key={billed_server_key} key={job.cache_key[:12]}"
    )
    _cache_log_t0 = time.time()

    if not payload.force_refresh:
        cached = lookup_summary(job)
        print(f"[summary-cache] {'HIT' if cached is not None else 'MISS'} {_cache_log_ctx}")
        if cached is not None:
            # Cache hits cost nothing, so they don't count against the quota.
            if billed_server_key:
                quota_info["free_remaining"] = _free_summaries_remaining(client_id)
            return {"summary": cached, "cached": True, **job.response, **quota_info}

    api_key = _resolve_api_key(
        payload.api_key, allow_secret_fallback=billed_server_key
    )
    reservation = _reserve_free_summary(client_id) if billed_server_key else None
    try:
        summary_text = generate_summary(job, api_key, billed_server_key)
    except Exception as e:
        print(
            f"[summary-cache] GENERATION_FAILED took={time.time() - _cache_log_t0:.1f}s"
            f" error={type(e).__name__} {_cache_log_ctx}"
        )
        if reservation is not None:
            _release_free_summary(client_id, reservation)
            # The server's own OpenAI account is rate-limited or out of
            # credit. Ask for the user's key instead of showing the raw error.
            import openai
            if isinstance(e, openai.RateLimitError):
                raise _server_key_unavailable()
        raise HTTPException(
            status_code=502, detail=f"OpenAI request failed: {_redact_secrets(e)}"
        )

    print(
        f"[summary-cache] GENERATED force_refresh={payload.force_refresh}"
        f" took={time.time() - _cache_log_t0:.1f}s chars={len(summary_text)} {_cache_log_ctx}"
    )
    if billed_server_key:
        quota_info["free_remaining"] = _free_summaries_remaining(client_id)
    if summary_text:
        store_cached_summary(job, summary_text)
    return {"summary": summary_text, "cached": False, **job.response, **quota_info}


@app.get("/progress")
async def get_progress(job_id: str) -> Dict[str, Any]:
    """
    Return processed/total counts plus the latest OpenAI batch status (if any)
    for the given classification job.
    """
    article_progress = _get_job(job_id)

    return {
        "processed": article_progress.get("processed", 0),
        "total": article_progress.get("total", 0),
        "stage": article_progress.get("stage"),
        "started_at": article_progress.get("started_at"),
        "batch": article_progress.get("batch"),
        "error": article_progress.get("error"),
    }


@app.get("/usage-log")
async def get_usage_log(
    limit: int = 500,
    x_usage_log_token: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
    """Return recent usage-log entries. Requires a valid X-Usage-Log-Token header.
    The token is read from a header rather than the query string so it doesn't
    end up in access logs, proxy logs, or browser history.
    """
    expected = (
        os.environ.get("USAGE_LOG_TOKEN", "").strip()
        or os.environ.get("OPENAI_API_KEY", "").strip()
    )
    token = (x_usage_log_token or "").strip()
    if not expected or not secrets.compare_digest(token, expected):
        raise HTTPException(status_code=403, detail="Invalid or missing token.")

    if not os.path.exists(USAGE_LOG_PATH):
        return {"entries": [], "total_entries": 0, "server_key_calls": 0}

    entries: List[Dict[str, Any]] = []
    server_key_calls = 0
    with _usage_log_lock:
        with open(USAGE_LOG_PATH, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    rec = json.loads(line)
                except Exception:
                    continue
                entries.append(rec)
                if rec.get("used_server_key"):
                    server_key_calls += 1

    return {
        "total_entries": len(entries),
        "server_key_calls": server_key_calls,
        "entries": entries[-max(0, limit):],
    }

# Serve the built React app at "/".
# This must be registered AFTER all API routes so /prepare, /classify, /progress, /results,
# /files still take precedence.
from fastapi.responses import FileResponse  # noqa: E402

FRONTEND_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "frontend_build")

if os.path.isdir(FRONTEND_DIR):
    # CRA's compiled JS/CSS lives under build/static/...
    app.mount(
        "/static",
        StaticFiles(directory=os.path.join(FRONTEND_DIR, "static")),
        name="frontend-assets",
    )

    @app.get("/")
    async def serve_index() -> FileResponse:
        return FileResponse(os.path.join(FRONTEND_DIR, "index.html"))

    @app.get("/{full_path:path}")
    async def spa_fallback(full_path: str) -> FileResponse:
        # Try a real file under frontend_build first (favicon.ico, manifest.json, etc.).
        candidate = os.path.join(FRONTEND_DIR, full_path)
        if os.path.isfile(candidate):
            return FileResponse(candidate)
        # Otherwise fall back to index.html (no client-side router today, but harmless).
        return FileResponse(os.path.join(FRONTEND_DIR, "index.html"))