api.py

import json
import os
import threading
import time
from typing import Any, Dict, List, Optional

import pandas as pd
from fastapi import BackgroundTasks, FastAPI, 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

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


# In-memory store for total revisions per article (keyed by safe_title)
progress_map: Dict[str, Dict[str, int]] = {}

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

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()

# Initialize FastAPI app
app = FastAPI(title="Wiki Revision Classifier API")

# 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"
)

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

class ClassifyPayload(BaseModel):
    article: str
    model: Optional[str] = None
    api_key: Optional[str] = None


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)
    with progress_lock:
        progress_map[safe_title] = {"total": revision_count}

    return {
        "previewRows": preview,
        "csvUrl": f"/files/{csv_filename}",
        "diffs": diffs,
        "revisionCount": revision_count,
    }

@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 = payload.model or "gpt-5-mini"

    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
    )

    safe_title = title.replace(" ", "_")
    total = len(prompts)
    with progress_lock:
        progress_map[safe_title] = {
            "total": total,
            "processed": 0,
            "stage": "submitting",
            "started_at": time.time(),
            "batch": None,
        }
    print(f"Initialized progress tracking for {safe_title}: 0/{total}")

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

    return {
        "status": "processing",
        "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(model_name, api_key, prompts, df, title, taxonomy):
    """Run the OpenAI batch classification in the background."""
    safe_title = title.replace(" ", "_")
    total = len(prompts)

    def _update_batch_status(batch_info: dict) -> None:
        with progress_lock:
            entry = progress_map.setdefault(safe_title, {})
            entry["batch"] = batch_info
            entry["stage"] = batch_info.get("status") or entry.get("stage", "running")
            completed = batch_info.get("completed", 0) or 0
            entry["processed"] = min(completed, entry.get("total", 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})")
        with progress_lock:
            progress_map.setdefault(safe_title, {})["stage"] = "submitting"

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

        with progress_lock:
            progress_map.setdefault(safe_title, {})["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 = f"{safe_title}_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_dataset(title, 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 = f"{safe_title}_evolution.txt"
        evolution_path = os.path.join("visualizations", evolution_filename)
        try:
            with progress_lock:
                progress_map.setdefault(safe_title, {})["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: {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 = f"{safe_title}_distribution.png"
        dist_path = os.path.join("visualizations", dist_filename)
        plot_label_distribution(results_df, taxonomy, output_path=dist_path)

        cooc_filename = f"{safe_title}_cooccurrence.png"
        cooc_path = os.path.join("visualizations", cooc_filename)
        plot_label_cooccurrence(results_df, taxonomy, output_path=cooc_path)

        with progress_lock:
            progress_map[safe_title]["processed"] = total
            progress_map[safe_title]["stage"] = "complete"

        print(f"Background processing completed for {title}")
        with progress_lock:
            print(f"Final progress_map: {progress_map}")
    except Exception as e:
        print(f"Error in background processing: {e}")
        with progress_lock:
            progress_map.setdefault(safe_title, {})["stage"] = "error"
            progress_map[safe_title]["error"] = str(e)
        import traceback
        traceback.print_exc()


@app.get("/results")
async def get_results(article: str) -> Dict[str, Any]:
    """
    Get the results of a completed classification.
    """
    safe_title = article.replace(" ", "_")
    result_filename = f"{safe_title}_results.csv"
    result_path = os.path.join("visualizations", result_filename)

    # Visualization file paths
    dist_filename = f"{safe_title}_distribution.png"
    distribution_path = os.path.join("visualizations", dist_filename)

    cooc_filename = f"{safe_title}_cooccurrence.png"
    cooccurrence_path = os.path.join("visualizations", cooc_filename)

    # Return "processing" unless **all** files exist
    if not all(os.path.exists(p) for p in [result_path, distribution_path, cooccurrence_path]):
        return {
            "status": "processing",
            "message": "Results not available yet. Check progress with the /progress endpoint."
        }

    # Load results
    results_df = pd.read_csv(result_path)
    _store_dataset(article, 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_filename = f"{safe_title}_evolution.txt"
    evolution_path = os.path.join("visualizations", evolution_filename)
    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)

# 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)

    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] = None
    model: Optional[str] = "gpt-5.4"
    # Set only for the built-in demo pages; other summaries run on the user's
    # own key.
    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


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 = "gpt-5.4"
) -> 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="gpt-5.4",
        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="gpt-5.4",
        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


@app.post("/summarize")
async def summarize_edits(payload: SummarizePayload) -> Dict[str, Any]:
    """
    Summarize the Explanation text for all rows that share a (label, period,
    section-group) cell using a fast OpenAI model. Used by interactive chart
    drill-downs.
    """
    # The server key is limited to demo articles served from the bundled
    # dataset.
    if payload.use_server_key:
        if payload.article not in DEMOS:
            raise HTTPException(
                status_code=403,
                detail=(
                    "The server API key may only be used for built-in demo "
                    f"articles. '{payload.article}' is not a demo — supply your "
                    "own OpenAI key for this request."
                ),
            )
        if payload.explanations is not None:
            raise HTTPException(
                status_code=403,
                detail=(
                    "The server API key may not be used with client-supplied "
                    "explanations. Demo summaries are computed from the server's "
                    "bundled dataset only."
                ),
            )

    # 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:
        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 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)
        )

    billed_server_key = payload.use_server_key and not (payload.api_key or "").strip()
    try:
        import openai
        client = openai.OpenAI(
            api_key=_resolve_api_key(
                payload.api_key, allow_secret_fallback=payload.use_server_key
            )
        )
        completion = client.chat.completions.create(
            model="gpt-5.4",
            store=True,
            messages=[
                {
                    "role": "system",
                    "content": (
                        "You are a careful research assistant that summarizes "
                        "lists of Wikipedia edit explanations into compact, "
                        "high-signal prose."
                    ),
                },
                {"role": "user", "content": user_prompt},
            ],
        )
        summary_text = completion.choices[0].message.content or ""
        _log_openai_call(
            endpoint="summarize",
            model="gpt-5.4",
            article=payload.article,
            used_server_key=billed_server_key,
            prompt_chars=len(user_prompt),
            response_chars=len(summary_text),
            usage=_usage_dict(completion),
            extra={
                "labels": combined_labels,
                "period_key": payload.period_key,
                "snippet_count": len(snippets_for_prompt),
                "client_supplied_explanations": client_snippets is not None,
            },
        )
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"OpenAI request failed: {e}")

    return {
        "summary": summary_text.strip(),
        "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,
    }


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

    with progress_lock:
        article_progress = dict(progress_map.get(safe_title, {}))

    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(token: str, limit: int = 500) -> Dict[str, Any]:
    """Return recent usage-log entries. Requires a valid token."""
    expected = (
        os.environ.get("USAGE_LOG_TOKEN", "").strip()
        or os.environ.get("OPENAI_API_KEY", "").strip()
    )
    if not expected or token.strip() != 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"))