devtools/e2e.py

"""End-to-end check of the real pipeline on sample texts (no UI):
    python devtools/e2e.py [detect|humanize|smart|all] [--full-ai] [--passes N] [--sample tech|social|human]
Prints the before/after AI score, per-sentence changes, integrity checks and timings."""
import argparse
import os
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(Path(__file__).resolve().parent))
for s in (sys.stdout, sys.stderr):
    s.reconfigure(encoding="utf-8", errors="replace")
import env  # noqa: E402
env.load()
os.environ.setdefault("LT_GRAMMAR", os.environ.get("LT_GRAMMAR", ""))

import samples  # noqa: E402

SAMPLES = {"tech": samples.AI_TECH, "social": samples.AI_SOCIAL, "human": samples.HUMAN}


def rep(frac, desc=None):
    if desc:
        print(f"    [{frac:4.0%}] {desc}", flush=True)


TIMES = {}


def profile(mod, name):
    """Accumulate wall time and call count of mod.name (printed at the end)."""
    fn = getattr(mod, name)
    key = f"{mod.__name__}.{name}"

    def wrapped(*a, **k):
        t = time.time()
        try:
            return fn(*a, **k)
        finally:
            c = TIMES.setdefault(key, [0, 0.0, 0])
            c[0] += 1
            c[1] += time.time() - t
            if a and isinstance(a[0], (list, tuple)):
                c[2] += len(a[0])
    setattr(mod, name, wrapped)


def integrity(orig, new):
    """Numbers, citations and technical tokens of the original that are missing in the result."""
    import re
    import textparse as TP
    nums = set(re.findall(r"\d+(?:[.,]\d+)*", orig)) - set(re.findall(r"\d+(?:[.,]\d+)*", new))
    cites = {m.group(0) for m in TP.CITE.finditer(orig)} - {m.group(0) for m in TP.CITE.finditer(new)}
    tech = {m.group(0) for m in TP.TECH_TOKEN.finditer(orig)} - {m.group(0) for m in TP.TECH_TOKEN.finditer(new)}
    return dict(numbers=sorted(nums), citations=sorted(cites), terms=sorted(tech))


def show_pairs(orig, new):
    import humanizer as H
    import meaning as M
    so, sn = H.split_sentences(orig.replace("\n", " ")), H.split_sentences(new.replace("\n", " "))
    if len(so) != len(sn):
        print(f"  !! sentence count changed: {len(so)} -> {len(sn)}")
        print("  OUT:", new)
        return
    sims = M.similarities(list(zip(so, sn)))
    for a, b, s in zip(so, sn, sims):
        if a.strip() == b.strip():
            print(f"  = {a}")
        else:
            print(f"  - {a}\n  + {b}\n      meaning {s:.2f} | new wording {H.words_changed(a, b):.0f}%")


def make_docx(path):
    """A small manuscript with the things a rewrite must not disturb: heading, bold / italic runs inside a
    sentence, a superscript citation, a table, and a reference list."""
    from docx import Document
    d = Document()
    d.add_heading("Hybrid Thermal Management of a 48 V Battery Module", 1)
    paras = samples.AI_TECH.split(chr(10) * 2)
    first = paras[0]
    a, b = first.split("phase change material (PCM)")
    p = d.add_paragraph()
    p.add_run(a)
    p.add_run("phase change material (PCM)").bold = True
    head, tail = b.split(" [12].")
    p.add_run(head)
    p.add_run("12").font.superscript = True
    p.add_run("." + tail)
    d.add_heading("2. Monitoring model", 2)
    p2 = d.add_paragraph()
    x, y = paras[1].split("ResNet-50")
    p2.add_run(x)
    p2.add_run("ResNet-50").italic = True
    p2.add_run(y)
    t = d.add_table(rows=2, cols=2)
    for (r, c), v in {(0, 0): "Cooling", (0, 1): "Max temperature (°C)", (1, 0): "Hybrid", (1, 1): "41.2"}.items():
        t.cell(r, c).text = v
    d.add_heading("References", 1)
    d.add_paragraph("[12] Kumar, A., Lee, B. (2021). Thermal runaway in lithium-ion cells. Journal of Power Sources, "
                    "vol. 400, pp. 1-12. https://doi.org/10.1000/xyz")
    d.save(path)
    return path


def docx_facts(path):
    """What must be identical before and after: headings, formatted runs, table cells, reference entries."""
    from docx import Document
    d = Document(path)
    return dict(headings=[p.text for p in d.paragraphs if p.style.name.startswith("Heading")],
                bold=[r.text for p in d.paragraphs for r in p.runs if r.bold],
                italic=[r.text for p in d.paragraphs for r in p.runs if r.italic],
                sup=[r.text for p in d.paragraphs for r in p.runs if r.font.superscript],
                table=[c.text for t in d.tables for row in t.rows for c in row.cells],
                refs=[p.text for p in d.paragraphs if p.text.startswith("[12]")],
                n_paras=len(d.paragraphs))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("what", nargs="?", default="all")
    ap.add_argument("--sample", default="tech")
    ap.add_argument("--full-ai", action="store_true")
    ap.add_argument("--passes", type=int, default=2)
    ap.add_argument("--vocab", type=float, default=0)
    a = ap.parse_args()
    text = SAMPLES[a.sample]
    t0 = time.time()
    import detector as D
    import humanizer as H
    import smart as S
    import groqai as G
    import meaning as M, fluency as FL, clarity as CL, lexsub as LX
    for mod, name in ((D, "score_texts"), (H, "_generate"), (G, "_chat"), (M, "similarities"), (FL, "scores"),
                      (CL, "assess"), (LX, "variants"), (G, "review"), (G, "polish"), (G, "grammar"), (G, "variants")):
        profile(mod, name)
    print(f"models loaded in {time.time() - t0:.0f}s | groq: {G.status()}")
    H.FULL_AI.set(a.full_ai)
    H.VOCAB_TARGET.set(a.vocab or None)

    if a.what in ("detect", "all"):
        t = time.time()
        for name, tx in SAMPLES.items():
            print(f"[detect] {name}: {S.detector_pct_text(tx)}% AI")
        print(f"[detect] {time.time() - t:.1f}s")

    if a.what in ("humanize", "all"):
        t = time.time()
        out, path, _, summary, _ = H._text_core(text, 0.9, 6, "Auto", rep, max_words=5000)
        print(f"[humanize] {time.time() - t:.1f}s\n{summary}")
        show_pairs(text, out)
        print("  AI score:", S.detector_pct_text(text), "->", S.detector_pct_text(out), "| missing:", integrity(text, out))

    if a.what in ("smart", "all"):
        t = time.time()
        r = S.smart_text(text, 0, a.passes, 0.9, "Auto", 0.80, 0.50, rep)
        print(f"[smart] {time.time() - t:.1f}s  AI {r['before']['pct']}% -> {r['after']['pct']}%")
        for line in r["log"]:
            print("   ", line)
        print("   ", r["note"])
        show_pairs(r["orig"], r["new"])
        print("  missing:", integrity(r["orig"], r["new"]))
    if a.what in ("docx", "all"):
        src = make_docx(str(ROOT / "devtools" / "logs" / "sample.docx"))
        before = docx_facts(src)
        t = time.time()
        r = S.smart_docx(src, 0, a.passes, 0.9, "Auto", 0.80, 0.50, rep)
        after = docx_facts(r["path"])
        print(f"[docx] {time.time() - t:.1f}s  AI {r['before']['pct']}% -> {r['after']['pct']}%  -> {r['path']}")
        for line in r["log"]:
            print("   ", line)
        print("   ", r["note"])
        for k in before:
            print(f"  {'OK  ' if before[k] == after[k] else 'DIFF'} {k}" + ("" if before[k] == after[k] else
                                                                        f": {before[k]} -> {after[k]}"))
        show_pairs(r["orig"], r["new"])
        print("  missing:", integrity(r["orig"], r["new"]))
    print("groq:", G.status())
    for key, (n, sec, items) in sorted(TIMES.items(), key=lambda x: -x[1][1]):
        print(f"  {sec:7.1f}s  {n:4d} calls  {items:5d} items  {key}")


if __name__ == "__main__":
    main()