#!/usr/bin/env python3 """§E-достройка по вопросу владельца 04.08: переносятся ли выводы §C (инжект) и §B4 (батч-граница семьи) с пары zh→ru на ja→ru и en→ru. Книги: ~/books/isekai_majutsushi_jp.txt · fifty_shades_1_en.txt. Части: inject-ja: 2 окна × 4 арма (A/B/C/D как в inject_probe) — послушание · вред · пустота маркера; inject-en: 1 окно × 4 арма — то же на en→ru; family-ja: семья 魔術/魔術師 (общий корень, РАЗНАЯ длина) НАМЕРЕННО разрезана на 2 батча терминолог-структурного промпта (source_lang=ja, поливанов) — проверка батч-граничной химеры конвенции корня вне zh. Голда нет — скорится СВЯЗНОСТЬ, не правильность. deepseek-v4-flash, reasoning_effort low, temp 0. Сырьё → ~/books/gu-zhenren/bank-arbitration/ (общий durable-синк пака). НЕ engine-faithful по паре (ja/en-паков в репо нет) — это проба ФЕНОМЕНА, не движка. """ from __future__ import annotations import argparse import json import os import re import time import unicodedata from pathlib import Path from dotenv import load_dotenv from openai import OpenAI REPO = Path("/home/ubuntu/projects/textmachine") RAW = Path.home() / "books" / "gu-zhenren" / "bank-arbitration" load_dotenv(REPO / "eval" / ".env") PRICE_IN, PRICE_CACHED, PRICE_OUT = 0.14, 0.0028, 0.28 GLOSSARY_HEADER = ("ГЛОССАРИЙ (используй эти утверждённые переводы имён и терминов последовательно; " "строки с пометкой ⟨проверить⟩ — неподтверждённые кандидаты):") MARKER = " ⟨проверить⟩" BOOKS = { "ja": dict( path=Path.home() / "books" / "isekai_majutsushi_jp.txt", src_lang="японский", transcription="поливанов", terms={ "勇者": ("герой", "храбрец", r"геро(й|я|ю|ем|е|и|ев)", r"храбрец"), "魔力": ("магическая сила", "мана", r"магическ\w* сил", r"\bман(а|ы|у|ой|е)\b"), "国王": ("король", "государь", r"корол(ь|я|ю|ем|е|ев)", r"государ"), "魔術": ("магия", "чародейство", r"маги(я|и|ю|ей)", r"чародейств"), "洗脳": ("промывание мозгов", "зомбирование", r"промыв\w* мозг", r"зомбир"), }, wrong_priority=["魔力", "国王", "洗脳", "魔術", "勇者"], n_windows=2, win_lo=400, win_hi=900, skip=25, ), "en": dict( path=Path.home() / "books" / "fifty_shades_1_en.txt", src_lang="английский", transcription="практическая транскрипция", terms={ "Christian Grey": ("Кристиан Грей", "Кристиан Грэй", r"кристиан\w* грей", r"грэй"), "Kate": ("Кейт", "Катя", r"\bкейт\b", r"\bкат(я|и|е|ю|ей)\b"), "Anastasia": ("Анастейша", "Анастасия", r"анастейш", r"анастаси"), }, wrong_priority=["Kate", "Christian Grey", "Anastasia"], n_windows=1, win_lo=600, win_hi=1200, skip=0, ), } TRANSLATOR_SYS = """Ты — профессиональный литературный переводчик с языка «{src}» на русский. Жанр: художественная проза. Переводи ВСЕ предложения исходника: ничего не пропускай и не добавляй. Пиши живым литературным русским языком. Транскрипция имён: {tr}. Род и форму имён собственных и терминов бери из приложенного глоссария (если он есть). Выведи ТОЛЬКО перевод, без преамбул, комментариев и markdown.""" FAMILY_SYS = """Ты — терминолог издательского перевода с языка «японский» на язык «русский». Тебе дан список терминов книги с контекстами из исходника. Выбери ОДИН перевод на термин, единый для всей книги. Сохраняй родство однокоренных терминов. Имена — транскрипцией ({tr}); понятия — по смыслу. Давай словарную форму без пояснений. Формат ответа: по одной строке на термин, ровно два поля через ТАБУЛЯЦИЮ: термин и перевод.""" # family-ja: 魔術 в батче 1, 魔術師 в батче 2 — граница режет семью общего корня (разной длины). FAMILY_BATCHES = [["魔術", "国王", "洗脳"], ["魔術師", "魔力", "勇者"]] def norm(s: str) -> str: return " ".join(unicodedata.normalize("NFKC", s).casefold().replace("ё", "е").split()) def pick_windows(cfg): text = cfg["path"].read_text(encoding="utf-8") paras = [p for p in text.split("\n") if p.strip() and not p.startswith(("http", "===", "URL"))] wins, i = [], cfg.get("skip", 0) while i < len(paras): buf, j = "", i while j < len(paras) and len(buf) < cfg["win_lo"]: buf += paras[j] + "\n" j += 1 if len(buf) > cfg["win_hi"]: buf = buf[:cfg["win_hi"]] terms = [t for t in cfg["terms"] if t in buf] if len(terms) >= 2: wins.append((i, buf, terms)) i = j + 5 else: i += 1 wins.sort(key=lambda w: (-len(w[2]), w[0])) chosen, used = [], set() for w in wins: if any(abs(w[0] - u) < 40 for u in used): continue chosen.append(w) used.add(w[0]) if len(chosen) == cfg["n_windows"]: break return chosen def block_for(cfg, terms, wrong, marked): lines = [] for t in terms: gold, bad, _, _ = cfg["terms"][t] if t == wrong: lines.append(f"{t} → {bad}" + (MARKER if marked else "")) else: lines.append(f"{t} → {gold}") return GLOSSARY_HEADER + "\n" + "\n".join(lines) def call(cl, system, inj, user, tag): msgs = [{"role": "system", "content": system}] if inj: msgs.append({"role": "system", "content": inj}) msgs.append({"role": "user", "content": user}) t0 = time.time() r = cl.chat.completions.create(model="deepseek-v4-flash", messages=msgs, max_tokens=16000, temperature=0, extra_body={"reasoning_effort": "low"}) ch = r.choices[0] content = ch.message.content or "" u = r.usage pt, ct = u.prompt_tokens or 0, u.completion_tokens or 0 cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0 cost = (pt - cached) / 1e6 * PRICE_IN + cached / 1e6 * PRICE_CACHED + ct / 1e6 * PRICE_OUT rec = dict(tag=tag, model_returned=r.model, effort="low", finish=ch.finish_reason, ts=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), prompt_tokens=pt, cached_tokens=cached, completion_tokens=ct, cost_usd=round(cost, 6), latency_s=round(time.time() - t0, 1), injection=inj, user=user, content=content) (RAW / f"{tag}.json").write_text(json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8") print(f"[{tag}] finish={ch.finish_reason} out={ct} ${cost:.6f} {rec['latency_s']}s") return rec def kwic(text, term, per=3, width=40): out, frm = [], 0 while len(out) < per: i = text.find(term, frm) if i < 0: break out.append(text[max(0, i - width):i + len(term) + width].replace("\n", " ")) frm = i + len(term) return out def run_inject(cl, lang): cfg = BOOKS[lang] total = 0.0 for k, (i, buf, terms) in enumerate(pick_windows(cfg)): cand = [t for t in cfg["wrong_priority"] if t in terms] wrong = cand[(2 * k + 1) % len(cand)] system = TRANSLATOR_SYS.format(src=cfg["src_lang"], tr=cfg["transcription"]) user = "Переведи следующий фрагмент:\n\n" + buf for arm, inj in {"A": None, "B": block_for(cfg, terms, None, False), "C": block_for(cfg, terms, wrong, False), "D": block_for(cfg, terms, wrong, True)}.items(): total += call(cl, system, inj, user, f"ml-{lang}-w{k}-{arm}")["cost_usd"] time.sleep(0.5) return total def run_family(cl): text = BOOKS["ja"]["path"].read_text(encoding="utf-8") total = 0.0 for bi, batch in enumerate(FAMILY_BATCHES): blocks = [] for t in batch: blocks.append(f"### {t}\nkey: {t}\ntype: term\nfreq: {text.count(t)}\n" + "\n".join(f"ctx: {c}" for c in kwic(text, t))) user = "Термины книги:\n\n" + "\n\n".join(blocks) total += call(cl, FAMILY_SYS.format(tr="поливанов"), None, user, f"ml-fam-b{bi}")["cost_usd"] time.sleep(0.5) return total def score(): rows = [] for lang in BOOKS: cfg = BOOKS[lang] for k, (i, buf, terms) in enumerate(pick_windows(cfg)): cand = [t for t in cfg["wrong_priority"] if t in terms] wrong = cand[(2 * k + 1) % len(cand)] for arm in "ABCD": f = RAW / f"ml-{lang}-w{k}-{arm}.json" if not f.exists(): continue out = norm(json.load(open(f, encoding="utf-8"))["content"]) for t in terms: gold, bad, grx, brx = cfg["terms"][t] rows.append(dict(lang=lang, win=k, arm=arm, term=t, is_wrong=(t == wrong), gold_hit=bool(re.search(grx, out)), wrong_hit=bool(re.search(brx, out)))) for lang in BOOKS: print(f"=== {lang} ===") for arm in "ABCD": ok = [r for r in rows if r["lang"] == lang and r["arm"] == arm and not r["is_wrong"]] wr = [r for r in rows if r["lang"] == lang and r["arm"] == arm and r["is_wrong"]] print(f" arm {arm}: верные {sum(r['gold_hit'] for r in ok)}/{len(ok)} gold; " f"wrong-терм: gold {sum(r['gold_hit'] for r in wr)}/{len(wr)}, WRONG {sum(r['wrong_hit'] for r in wr)}/{len(wr)}") print("=== family-ja (связность корня 魔術/魔術師 через границу батча) ===") for bi in (0, 1): f = RAW / f"ml-fam-b{bi}.json" if f.exists(): print(f" batch {bi}:") for ln in json.load(open(f, encoding="utf-8"))["content"].split("\n"): if ln.strip(): print(" ", ln) def main(): ap = argparse.ArgumentParser() ap.add_argument("--plan", action="store_true") ap.add_argument("--score", action="store_true") a = ap.parse_args() if a.plan: for lang in BOOKS: for k, (i, buf, terms) in enumerate(pick_windows(BOOKS[lang])): cand = [t for t in BOOKS[lang]["wrong_priority"] if t in terms] print(f"{lang} win{k} @{i} len={len(buf)} terms={terms} wrong={cand[(2 * k + 1) % len(cand)]}") print(" " + buf[:110].replace("\n", " / ")) return if a.score: score() return cl = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com/v1", timeout=600) total = run_inject(cl, "ja") + run_inject(cl, "en") + run_family(cl) print(f"TOTAL multilang: ${total:.6f}") if __name__ == "__main__": main()