Spaces:
Running
Running
unknown
Claude Opus 5.5
Harden rewrite integrity, stop the final polish undoing humanization, add tests
585f232 | """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() | |