192 lines
10 KiB
Python
192 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""Вебновелл-срез — ОДНА команда, запускается по появлению корпуса владельца (POLYGON handoff
|
||
Task 6). Пока корпуса нет — печатает, куда класть файлы, и выходит 0 (не падает).
|
||
|
||
ЗАЧЕМ (вся эмпирика полигона снята на КЛАССИКЕ из претрейна — exp02 правило #2; вебновелл-режим
|
||
не измерен нигде, 07-review §4.5 / V1.6). Этот срез добирает три вещи ИМЕННО на невиданном тексте:
|
||
A. r-коэффициенты токенов (exp01/08) — держатся ли множители токенов/символ на вебновелле
|
||
(COGS-модель exp08 висит на них). Через token_calc.py на корпусе среза.
|
||
B. частота DeepSeek-«эха» ВНЕ претрейна (exp02: на PD-классике deepseek эхал 13/24 zh-чанков —
|
||
это претрейн-канон; на невиданном тексте частота ДРУГАЯ и решает, насколько эхо-мина реальна).
|
||
C. precision coverage-гейта на ТЕРСНЫХ репликах (exp07: 0 FP/57 на классике; терсный диалог —
|
||
короткие реплики — самый шумный вход для sent_cov/len_ratio → гейтит снятие 1-флаг-толерантности).
|
||
|
||
КОРПУС: положи 3–5 глав реальных вебновелл в eval/data/samples/webnovel/<lang>/*.txt
|
||
(lang ∈ zh|ja|en; .txt UTF-8; главы разделены пустой строкой или по файлам). Затем:
|
||
eval/.venv/bin/python eval/webnovel_slice.py # прогонит A+B+C
|
||
eval/.venv/bin/python eval/webnovel_slice.py --dry-run # план без вызовов API
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse, json, subprocess, sys, time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import refusal_bench as rb
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
CORPUS = ROOT / "data" / "samples" / "webnovel" # <lang>/*.txt
|
||
OUT = ROOT / "data" / "webnovel_slice"
|
||
SYSTEM = ("Ты профессиональный литературный переводчик. Переведи фрагмент на русский язык. "
|
||
"Сохрани все реплики и детали без пропусков; стиль — живой литературный русский. "
|
||
"Выведи ТОЛЬКО перевод.")
|
||
# out/in char-ratio corridor by source lang (exp02 EXPECT_LEN_RATIO) — reused for the gate.
|
||
EXPECT = rb.EXPECT_LEN_RATIO
|
||
|
||
|
||
def discover() -> list[tuple[str, Path]]:
|
||
"""(lang, file) for every webnovel/<lang>/*.txt."""
|
||
out = []
|
||
if not CORPUS.exists():
|
||
return out
|
||
for lang_dir in sorted(CORPUS.iterdir()):
|
||
if lang_dir.is_dir():
|
||
for f in sorted(lang_dir.glob("*.txt")):
|
||
out.append((lang_dir.name, f))
|
||
return out
|
||
|
||
|
||
def chunk_text(text: str, target=1200) -> list[str]:
|
||
import re
|
||
paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
|
||
chunks, buf = [], []
|
||
for p in paras:
|
||
if buf and len("".join(buf)) + len(p) > target:
|
||
chunks.append("\n\n".join(buf)); buf = []
|
||
buf.append(p)
|
||
if buf:
|
||
chunks.append("\n\n".join(buf))
|
||
return chunks
|
||
|
||
|
||
def dialogue_density(chunk: str) -> float:
|
||
"""Fraction of lines that are quoted/terse dialogue (— … or 「」『』“”«» openers). Terse
|
||
dialogue is the noisy input for the coverage gate (exp07 §просьба)."""
|
||
lines = [l.strip() for l in chunk.splitlines() if l.strip()]
|
||
if not lines:
|
||
return 0.0
|
||
d = sum(1 for l in lines if l[:1] in "—–「『“«\"'" or l.startswith("- "))
|
||
return round(d / len(lines), 2)
|
||
|
||
|
||
def provider(name: str) -> dict:
|
||
provs = json.loads((ROOT / "providers.json").read_text())["providers"]
|
||
for p in provs:
|
||
if p["name"] == name:
|
||
return p
|
||
raise SystemExit(f"provider {name} not in providers.json")
|
||
|
||
|
||
def run_r_coefficients() -> dict:
|
||
"""A. token r-multipliers on the slice via token_calc.py (subprocess; source-only)."""
|
||
try:
|
||
r = subprocess.run([sys.executable, str(ROOT / "token_calc.py"),
|
||
"--samples-dir", str(CORPUS),
|
||
"--json-out", str(OUT / "token_calc_webnovel.json")],
|
||
capture_output=True, text=True, timeout=600)
|
||
return {"stdout_tail": r.stdout[-2000:], "rc": r.returncode,
|
||
"json": str(OUT / "token_calc_webnovel.json")}
|
||
except Exception as e:
|
||
return {"error": f"{type(e).__name__}: {e}"}
|
||
|
||
|
||
def run_echo_and_gate(files: list[tuple[str, Path]], translator: str, limit: int) -> dict:
|
||
"""B + C in one pass: translate each chunk once with `translator`, classify (echo / excision),
|
||
correlate excision flags with dialogue density. Resumable (per-chunk jsonl)."""
|
||
p = provider(translator)
|
||
res_path = OUT / f"{translator}_chunks.jsonl"
|
||
done = set()
|
||
if res_path.exists():
|
||
done = {json.loads(l)["id"] for l in res_path.read_text().splitlines() if l.strip()}
|
||
agg = {"n": 0, "echo": 0, "excision": 0, "ok": 0, "refusal": 0, "by_lang": {},
|
||
"excision_by_density": {"terse(>=0.5)": [0, 0], "prose(<0.5)": [0, 0]}}
|
||
with res_path.open("a") as fh:
|
||
for lang, f in files:
|
||
chunks = chunk_text(f.read_text(encoding="utf-8"))[:limit]
|
||
for ci, ch in enumerate(chunks):
|
||
cid = f"{f.stem}#{ci}"
|
||
if cid in done:
|
||
continue
|
||
t0 = time.time()
|
||
text, err, usage = rb.call_provider(p, SYSTEM, ch)
|
||
verdict = rb.classify_output(ch, text, err, EXPECT.get(lang, (0.5, 3.0)))
|
||
dens = dialogue_density(ch)
|
||
rec = {"id": cid, "lang": lang, "translator": translator, "provider_model": p["model"],
|
||
"dialogue_density": dens, "elapsed_s": round(time.time()-t0, 1),
|
||
"usage": usage, **verdict, "src_chars": len(ch), "output": text}
|
||
fh.write(json.dumps(rec, ensure_ascii=False) + "\n"); fh.flush()
|
||
time.sleep(p.get("rps_delay", 0.5))
|
||
# aggregate from the full jsonl (resumed + fresh)
|
||
for l in res_path.read_text().splitlines():
|
||
if not l.strip():
|
||
continue
|
||
r = json.loads(l)
|
||
agg["n"] += 1
|
||
lang = r["lang"]
|
||
bl = agg["by_lang"].setdefault(lang, {"n": 0, "echo": 0, "excision": 0, "ok": 0})
|
||
bl["n"] += 1
|
||
v = r["verdict"]
|
||
if v == "untranslated_echo":
|
||
agg["echo"] += 1; bl["echo"] += 1
|
||
elif v == "excision_suspect":
|
||
agg["excision"] += 1; bl["excision"] += 1
|
||
elif v == "ok":
|
||
agg["ok"] += 1; bl["ok"] += 1
|
||
elif v == "content_refusal":
|
||
agg["refusal"] += 1
|
||
bucket = "terse(>=0.5)" if r.get("dialogue_density", 0) >= 0.5 else "prose(<0.5)"
|
||
agg["excision_by_density"][bucket][1] += 1
|
||
if v == "excision_suspect":
|
||
agg["excision_by_density"][bucket][0] += 1
|
||
return agg
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--translator", default="deepseek", help="эхо-мина измеряется на deepseek (exp02)")
|
||
ap.add_argument("--limit", type=int, default=8, help="max chunks per file")
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
files = discover()
|
||
if not files:
|
||
print("ВЕБНОВЕЛЛ-СРЕЗ ЗАБЛОКИРОВАН: корпуса нет.\n"
|
||
f"Положи 3–5 глав реальных вебновелл в: {CORPUS}/<lang>/*.txt\n"
|
||
" (lang ∈ zh|ja|en; UTF-8; например eval/data/samples/webnovel/zh/novel-ch1.txt)\n"
|
||
"Затем повтори: eval/.venv/bin/python eval/webnovel_slice.py\n"
|
||
"Прогонит: A r-коэффициенты токенов · B частоту DeepSeek-эха вне претрейна · "
|
||
"C precision coverage-гейта на терсных репликах.", file=sys.stderr)
|
||
return
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
print(f"Корпус: {len(files)} файлов — {[str(f.relative_to(CORPUS)) for _, f in files]}", file=sys.stderr)
|
||
if args.dry_run:
|
||
for lang, f in files:
|
||
n = len(chunk_text(f.read_text(encoding='utf-8'))[:args.limit])
|
||
print(f" {lang}/{f.name}: {n} чанков → translate({args.translator})+classify", file=sys.stderr)
|
||
print(f" + token_calc.py --samples-dir {CORPUS}", file=sys.stderr)
|
||
return
|
||
|
||
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||
print("A. r-коэффициенты токенов…", file=sys.stderr)
|
||
r_coef = run_r_coefficients()
|
||
print("B+C. эхо + coverage-precision…", file=sys.stderr)
|
||
echo_gate = run_echo_and_gate(files, args.translator, args.limit)
|
||
|
||
report = {"run_utc": stamp, "corpus_files": [str(f.relative_to(CORPUS)) for _, f in files],
|
||
"A_r_coefficients": r_coef, "BC_echo_and_gate": echo_gate,
|
||
"baselines": {"B_pretrain_echo": "exp02: deepseek 13/24 zh-чанков на PD-классике",
|
||
"C_gate_precision": "exp07: 0 FP/57 на классике; терсный диалог — гейт-риск"}}
|
||
(OUT / f"webnovel_slice_{stamp}.json").write_text(json.dumps(report, ensure_ascii=False, indent=2))
|
||
e = echo_gate
|
||
print(f"\n=== ВЕБНОВЕЛЛ-СРЕЗ ({e['n']} чанков, translator={args.translator}) ===", file=sys.stderr)
|
||
print(f"B эхо: {e['echo']}/{e['n']} ({100*e['echo']//max(1,e['n'])}%) "
|
||
f"[претрейн-база exp02: 13/24 zh]", file=sys.stderr)
|
||
print(f"C excision_suspect: {e['excision']}/{e['n']}; по плотности диалога "
|
||
f"терсн={e['excision_by_density']['terse(>=0.5)']} проза={e['excision_by_density']['prose(<0.5)']} "
|
||
f"(FP/всего) [exp07-база: 0 FP/57]", file=sys.stderr)
|
||
print(f"по языкам: {e['by_lang']}", file=sys.stderr)
|
||
print(f"\nПРОВЕРЬ ВРУЧНУЮ excision-флаги (истинный FP требует сверки полноты) в "
|
||
f"{OUT}/{args.translator}_chunks.jsonl; отчёт: {OUT}/webnovel_slice_{stamp}.json", file=sys.stderr)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|