Blame for export.py

"""Export analysis results in formats that are easy to process downstream.

Three files per download:
- summary.csv    one row per text (tidy, wide). Rows from different downloads can be
                 stacked, e.g. to follow a source over time or compare groups of texts.
- sentences.csv  one row per sentence, with its position, relevance and whether it was analysed.
- results.json   everything above, nested, plus the instrument (model, poles, axis overlap).

Every row carries the information needed to judge whether two results are comparable:
the encoder model, a fingerprint of the pole exemplars, the topic-filter settings and the
app version. Results are only directly comparable when model and poles fingerprint match.
"""

from __future__ import annotations

import hashlib
import json
import re
import tempfile
import uuid
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd


def fingerprint(path: Path) -> str:
    """Short SHA-256 of a file (e.g. poles.tsv), to check that two results used the same poles."""
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()[:12]


def parse_tags(tags: str) -> dict:
    """'source=Guardian; year=2024; draft' -> {'source': 'Guardian', 'year': '2024', 'draft': 'yes'}"""
    out = {}
    for item in re.split(r"[;\n]", tags or ""):
        item = item.strip()
        if not item:
            continue
        if "=" in item:
            k, v = item.split("=", 1)
            out[k.strip()] = v.strip()
        else:
            out[item] = "yes"
    return out


def _slug(s: str) -> str:
    return re.sub(r"[^0-9A-Za-z]+", "_", s).strip("_").lower()


def record(res: dict, sp, lm: pd.DataFrame, meta: dict) -> tuple[dict, list[dict], dict]:
    """Build the summary row, the sentence rows and the JSON record for one analysed text.

    meta: run_id, label, date, source_file, tags (dict), model, model_label, poles_sha,
          app_version, topic_filter, threshold, filter_model, topic_definition, topic_sha.
    """
    ids = [ax["id"] for ax in sp.axes]
    xyz = res["text_xyz"]
    sx = res["sent_xyz"]
    c = res["cloud"]
    raw = res["table"]["Before orthogonalisation"].to_numpy(dtype=float)

    W = lm[ids].to_numpy(dtype=float)
    dist = np.linalg.norm(xyz - W, axis=1)
    zone = int(np.argmin(np.linalg.norm(xyz[:2] - W[:, :2], axis=1)))   # eco × change map zone
    nearest = int(np.argmin(dist))

    row = {
        "run_id": meta["run_id"],
        "label": meta["label"],
        "date": meta["date"],
        "source_file": meta["source_file"],
    }
    for k, v in meta["tags"].items():
        row[f"tag_{_slug(k)}"] = v
    row.update({
        "n_sentences_total": res["n_total"],
        "n_sentences_analysed": len(res["sentences"]),
        "filter_fallback": res["fallback"],
        "n_negation_analysed": sum(bool(n) for n in res["sent_neg"]),
    })
    for k, i in enumerate(ids):
        row[f"{i}_pct"] = round(float(xyz[k]), 2)
    for k, i in enumerate(ids):
        row[f"{i}_sd"] = round(float(sx[:, k].std()), 2) if len(sx) > 1 else None
    for k, i in enumerate(ids):
        row[f"{i}_raw_pct"] = round(float(raw[k]), 2)
    row["spread_frobenius"] = round(c["frobenius"], 2) if "frobenius" in c else None
    row["variation_captured"] = round(c["captured"], 4) if "captured" in c else None
    if "eigvals" in c:
        tot = float(c["eigvals"].sum()) or float("nan")
        for j in range(3):
            row[f"eig{j + 1}_share"] = round(float(c["eigvals"][j] / tot), 4)
        row["main_direction_axis"] = ids[c["main_axis"]]
        row["main_direction_angle_deg"] = round(float(c["angles"][c["main_axis"]]), 1)
        for k, i in enumerate(ids):
            row[f"main_direction_{i}"] = round(float(c["eigvecs"][k, 0]), 4)
    else:
        for j in range(3):
            row[f"eig{j + 1}_share"] = None
        row["main_direction_axis"] = None
        row["main_direction_angle_deg"] = None
        for i in ids:
            row[f"main_direction_{i}"] = None
    row["nearest_worldview"] = lm["name"].iloc[nearest]
    row["nearest_worldview_distance"] = round(float(dist[nearest]), 2)
    row["map_zone"] = lm["name"].iloc[zone]
    for name, d in zip(lm["name"], dist):
        row[f"dist_{_slug(name)}"] = round(float(d), 2)
    row.update({
        "model": meta["model"],
        "poles_sha": meta["poles_sha"],
        "topic_filter": meta["topic_filter"],
        "threshold": meta["threshold"] if meta["topic_filter"] else None,
        "filter_model": meta["filter_model"],
        "topic_definition": meta["topic_definition"],
        "topic_sha": meta["topic_sha"],
        "app_version": meta["app_version"],
        "analysed_utc": meta["analysed_utc"],
    })

    # Sentences: analysed and filtered-out, in their original order
    sent_rows = []
    for n, (s, used, r, p, neg) in enumerate(zip(res["all_sentences"], res["keep"], res["all_rel"],
                                                 res["all_xyz"], res["all_neg"]), start=1):
        sr = {"run_id": meta["run_id"], "label": meta["label"], "sentence_no": n,
              "analysed": bool(used), "relevance": round(float(r), 4), "negation_cues": neg}
        for k, i in enumerate(ids):
            sr[f"{i}_pct"] = round(float(p[k]), 2)
        sr["sentence"] = s
        sent_rows.append(sr)

    js = {
        "run_id": meta["run_id"],
        "metadata": {"label": meta["label"], "date": meta["date"],
                     "source_file": meta["source_file"], "tags": meta["tags"]},
        "instrument": {
            "app_version": meta["app_version"],
            "model": meta["model"],
            "model_label": meta["model_label"],
            "poles_sha": meta["poles_sha"],
            "orthogonalisation": "Loewdin (symmetric)",
            "axes": sp.axes,
            "axis_overlap_cosine": sp.overlap.to_numpy().round(4).tolist(),
            "topic_filter": meta["topic_filter"],
            "threshold": meta["threshold"] if meta["topic_filter"] else None,
            "filter_model": meta["filter_model"],
            "topic_definition": meta["topic_definition"],
            "topic_sha": meta["topic_sha"],
        },
        "analysed_utc": meta["analysed_utc"],
        "summary": row,
        "sentences": sent_rows,
    }
    return row, sent_rows, js


def write_files(rows: list[dict], sent_rows: list[dict], records: list[dict], stem: str) -> list[str]:
    """Write summary.csv, sentences.csv and results.json into a fresh temp folder."""
    folder = Path(tempfile.mkdtemp(prefix="discoursegeometry_"))
    stem = _slug(stem) or "discoursegeometry"
    paths = [folder / f"{stem}_summary.csv", folder / f"{stem}_sentences.csv", folder / f"{stem}_results.json"]
    pd.DataFrame(rows).to_csv(paths[0], index=False, encoding="utf-8")
    pd.DataFrame(sent_rows).to_csv(paths[1], index=False, encoding="utf-8")
    payload = records[0] if len(records) == 1 else {"runs": records}
    paths[2].write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=_jsonable), encoding="utf-8")
    return [str(p) for p in paths]


def _jsonable(o):
    if isinstance(o, (np.integer,)):
        return int(o)
    if isinstance(o, (np.floating,)):
        return float(o)
    if isinstance(o, (np.bool_,)):
        return bool(o)
    if isinstance(o, np.ndarray):
        return o.tolist()
    return str(o)


def new_meta(**kw) -> dict:
    kw.setdefault("run_id", uuid.uuid4().hex[:8])
    kw.setdefault("analysed_utc", datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"))
    return kw