241 lines
12 KiB
Python
241 lines
12 KiB
Python
#!/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()
|