textmachine/eval/exp14/exp14_material.py

#!/usr/bin/env python3
"""exp14 — материал: trap-чанки, Stage-II главы (детерм. D), реконструкция глоссарий-
инъекции из rerun-store (rig-фиделити A1′, как exp12_blocks), P0-baseline paragraph-KPI.

$0. Выход: /home/ubuntu/books/gu-zhenren/exp14/material.json (ВНЕ git).
"""
from __future__ import annotations
import json, re, sqlite3, statistics
from pathlib import Path

RERUN = Path("/home/ubuntu/books/gu-zhenren/rerun")
OUT = Path("/home/ubuntu/books/gu-zhenren/exp14"); OUT.mkdir(parents=True, exist_ok=True)
DB = RERUN / "guzhenren-rerun.db"

# ── Заморожённый выбор (§1.0) ──────────────────────────────────────────────────────
TRAP_CHUNKS = [(7, 0), (8, 0), (8, 1), (9, 0), (10, 0), (11, 0), (11, 1), (24, 0), (24, 1)]
STAGE2 = [19, 17, 18]
EXCLUDE = {(5, 0), (20, 0)}  # экспорт-баг D35.4a

EDITOR_HEADER = ("КАНОНИЧЕСКИЕ ПЕРЕВОДЫ имён и терминов (в черновике эти сущности должны быть "
                 "переданы ИМЕННО этими формами — приводи к ним любые расхождения, склоняя по "
                 "контексту; не вводи иных вариантов и не меняй ничего другого):")
GLOSS_HEADER = ("ГЛОССАРИЙ (используй эти утверждённые переводы имён и терминов последовательно; "
                "строки с пометкой ⟨проверить⟩ — неподтверждённые кандидаты):")


def load_records():
    d = json.load(open(RERUN / "records.json"))
    return {(r["chapter"], r["chunk_idx"]): r for r in d}, d


def paras(t: str):
    return [p for p in re.split(r"\n\s*\n", t.strip()) if p.strip()]

def sents(p: str):
    return [s for s in re.split(r"(?<=[.!?…])\s+", p.strip()) if s.strip()]

def dialogue_D(source: str) -> float:
    """exp12 §1.1: доля символов внутри «“…”» (U+201C..U+201D) от не-пробельных."""
    nonspace = sum(1 for c in source if not c.isspace())
    inq, indq = 0, False
    for c in source:
        if c == "“":
            indq = True; continue
        if c == "”":
            indq = False; continue
        if indq and not c.isspace():
            inq += 1
    return inq / max(1, nonspace)


def reconstruct_injection():
    """editor_constraint (dst-only, approved, dedup) + bilingual_glossary (src→dst[+⟨проверить⟩])
    per chunk из injected_ids + glossary. Sticky union chunk0∪.. внутри главы (exp12 A5)."""
    con = sqlite3.connect(DB); con.row_factory = sqlite3.Row
    gl = {(r["src"], r["sense"], r["since_ch"], r["until_ch"]): (r["dst"], r["status"])
          for r in con.execute("SELECT src,sense,since_ch,until_ch,dst,status FROM glossary")}
    inj = {}
    for r in con.execute("SELECT chapter,chunk_idx,injected_ids,n_sticky,n_evicted FROM retrieval_state"):
        inj[(r["chapter"], r["chunk_idx"])] = (json.loads(r["injected_ids"] or "[]"),
                                               r["n_sticky"], r["n_evicted"])
    con.close()

    def keydec(k):
        src, sense, since, until = k.split("\x1f")
        return (src, sense, int(since), int(until))

    def block_for(keys):
        recs = []
        for k in keys:
            src, sense, since, until = keydec(k)
            dst, status = gl.get((src, sense, since, until), ("", "auto"))
            if dst.strip():
                recs.append((src, dst.strip(), status == "approved"))
        seen, el = set(), []
        for src, dst, conf in recs:
            if conf and dst not in seen:
                seen.add(dst); el.append(f"- «{dst}»")
        eb = (EDITOR_HEADER + "\n" + "\n".join(el)) if el else ""
        glines = [f"{src} → {dst}" + ("" if conf else "  ⟨проверить⟩") for src, dst, conf in recs]
        gb = (GLOSS_HEADER + "\n" + "\n".join(glines)) if glines else ""
        return eb, gb, len(el), len(glines)

    out = {}
    chapters = sorted(set(ch for ch, _ in inj))
    for ch in chapters:
        cidxs = sorted(c for (cc, c) in inj if cc == ch)
        chunk0keys = inj.get((ch, 0), ([], 0, 0))[0]
        for c in cidxs:
            keys, nst, nev = inj[(ch, c)]
            active = list(keys)
            for k in chunk0keys:  # sticky scene-inertia within chapter
                if c > 0 and k not in active:
                    active.append(k)
            eb, gb, ne, ng = block_for(active)
            out[f"{ch}/{c}"] = {"editor_constraint": eb, "bilingual_glossary": gb,
                                "n_constraint": ne, "n_gloss": ng, "n_sticky": nst, "n_evicted": nev}
        # chapter-level union (для P1b/P1c whole-chapter арма)
        allkeys = []
        for c in cidxs:
            for k in inj[(ch, c)][0]:
                if k not in allkeys:
                    allkeys.append(k)
        eb, gb, ne, ng = block_for(allkeys)
        out[f"{ch}/ALL"] = {"editor_constraint": eb, "bilingual_glossary": gb,
                            "n_constraint": ne, "n_gloss": ng}
    return out


def p0_kpi(recs):
    """P0 baseline: sentences/narrative-paragraph по 55 committed чанкам (freeze-число)."""
    sp = []
    for (ch, ck), r in recs.items():
        if (ch, ck) in EXCLUDE:
            continue
        for p in paras(r["final"]):
            if p.strip().startswith("—"):
                continue
            sp.append(len(sents(p)))
    return {"n_paras": len(sp), "median": statistics.median(sp),
            "mean": round(statistics.mean(sp), 3),
            "share_1sent": round(sum(1 for x in sp if x == 1) / len(sp), 3)}


def main():
    recs, _ = load_records()
    inj = reconstruct_injection()

    # verify: trap chunks have reconstructed blocks
    print("=== Инъекция реконструирована (trap-чанки) ===")
    for ch, ck in TRAP_CHUNKS:
        b = inj[f"{ch}/{ck}"]
        print(f"  {ch}.{ck}: {b['n_constraint']} constraint / {b['n_gloss']} gloss "
              f"(sticky={b['n_sticky']} evict={b['n_evicted']})")

    # Stage-II D
    print("\n=== Stage-II главы: детерм. D (диалого-плотность) ===")
    d_scores = {}
    for ch in STAGE2:
        src = "".join(recs[(ch, c)]["source"] for (cc, c) in recs if cc == ch)
        d_scores[ch] = round(dialogue_D(src), 3)
        print(f"  гл.{ch}: D={d_scores[ch]}")

    # P0 KPI
    kpi = p0_kpi(recs)
    print(f"\n=== P0 baseline paragraph-KPI (freeze-сверка) ===\n  {kpi}")

    # export material for arms
    mat = {"trap_chunks": [], "stage2": {}, "kpi_p0": kpi, "stage2_D": d_scores}
    for ch, ck in TRAP_CHUNKS:
        r = recs[(ch, ck)]
        mat["trap_chunks"].append({"chapter": ch, "chunk_idx": ck,
            "source": r["source"], "draft": r["draft"], "final_p0": r["final"],
            "editor_constraint": inj[f"{ch}/{ck}"]["editor_constraint"],
            "bilingual_glossary": inj[f"{ch}/{ck}"]["bilingual_glossary"]})
    for ch in STAGE2:
        mat["stage2"][ch] = [{"chunk_idx": c, "source": recs[(ch, c)]["source"],
            "draft": recs[(ch, c)]["draft"], "final_p0": recs[(ch, c)]["final"],
            "editor_constraint": inj[f"{ch}/{c}"]["editor_constraint"],
            "bilingual_glossary": inj[f"{ch}/{c}"]["bilingual_glossary"]}
            for (cc, c) in sorted(recs) if cc == ch]
    (OUT / "material.json").write_text(json.dumps(mat, ensure_ascii=False), encoding="utf-8")
    (OUT / "injection_blocks.json").write_text(json.dumps(inj, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"\n→ {OUT/'material.json'}  ({len(mat['trap_chunks'])} trap chunks, {len(STAGE2)} stage-II chapters)")


if __name__ == "__main__":
    main()