257 lines
13 KiB
Python
257 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Ф0.5 — БАНК ТЕРМИНОВ СРЕЗА: майнинг кандидатов → терминолог-прогон → ручной канон.
|
||
|
||
Зачем банк нужен ДО первой панели. Закон-блок D39.104 обязателен во всех армах с переводом или
|
||
редактурой, а метрика ПОКРЫТИЯ КАНОНА — единственный контур, которым эксп-21 поймал, что редактор
|
||
без банка ломает терминологию черновика (0.917 → 0.566). Без банка эксп-22 остался бы без этого
|
||
контура на новой книге.
|
||
|
||
⚠ ГРАНИЦА, объявляю: это НЕ ось эксперимента. Наличие банка как ФАКТОР (строки 5/36б бэклога) —
|
||
чужой слот и здесь не варьируется: банк собирается один раз и подаётся ВСЕМ армам одинаково.
|
||
|
||
⚠ «РУЧНОЙ КАНОН ≠ ПОДПИСЬ ВЛАДЕЛЬЦА» (образец D39.47). Терминолог предлагает, сессия правит явные
|
||
промахи, и получившийся банк помечен `signed: false`. Он годен как ЗАКОН внутри замера (все армы
|
||
видят одно) и НЕ годен как продуктовый канон книги.
|
||
|
||
Запуск: --mine кандидаты и KWIC, $0
|
||
--ask терминолог-прогон (Ф0, платно)
|
||
--canon свести ответ + ручные правки в bank.json, $0
|
||
--show что в банке, $0
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
ZONE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(ZONE))
|
||
sys.path.insert(0, str(REPO / "eval" / "bank_arbitration"))
|
||
|
||
from dotenv import load_dotenv # noqa: E402
|
||
from openai import OpenAI # noqa: E402
|
||
|
||
import material as M # noqa: E402
|
||
import money as MONEY # noqa: E402
|
||
import prompts as PR # noqa: E402
|
||
from roster import CANDIDATES # noqa: E402
|
||
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
|
||
OUT = MONEY.OUT
|
||
TERMINOLOGIST = REPO / "backend" / "prompts" / "zh-ru" / "terminologist.md"
|
||
# Терминолог — вспомогательный инструмент, не арм. Берётся дешёвая модель с управляемым
|
||
# размышлением: у боевого `deepseek-v4-flash` бэнк-роли ручки эффорта не имеют и упираются в
|
||
# стену (quirks §10 п.10, 4 батча из 5 пустые), а гасить thinking у DeepSeek запрещено.
|
||
TERM_MODEL = "gpt-5.6-luna"
|
||
N_TERMS = 26
|
||
MIN_COUNT = 6
|
||
KWIC_N, KWIC_W = 3, 46
|
||
|
||
# Служебные и грамматические знаки: n-грамма, начинающаяся или кончающаяся на них, термином не
|
||
# бывает. Список закрытый и объявлен ДО майнинга.
|
||
STOP_EDGE = set("的了是在和有我你他她们不就都而及与也很到说着过又却把被将从对为以之其这那一二三四五"
|
||
"六七八九十个人上下中大小时年日月来去出可能会要没什么样如此但只还更最")
|
||
STOP_WHOLE = {"自己", "已经", "现在", "眼中", "心中", "一个", "什么", "这样", "那些", "如今",
|
||
"顿时", "此刻", "随后", "所以", "因为", "如果", "虽然", "而且", "然后", "似乎"}
|
||
|
||
|
||
def source_text() -> str:
|
||
return "\n".join(c["text"] for c in M.chapters())
|
||
|
||
|
||
def mine() -> list[tuple[str, int]]:
|
||
"""Кандидаты в термины: частые CJK n-граммы, не поглощённые более длинной n-граммой.
|
||
|
||
Метод грубый и объявлен таким: его задача — дать терминологу СПИСОК, решение принимает он
|
||
плюс ручная правка. Точность майнинга на банк не влияет — влияет полнота.
|
||
"""
|
||
text = source_text()
|
||
cnt: Counter[str] = Counter()
|
||
for n in (4, 3, 2):
|
||
for m in re.finditer(r"[㐀-鿿]{%d}" % n, text):
|
||
cnt[m.group(0)] += 1
|
||
cand = {w: c for w, c in cnt.items()
|
||
if c >= MIN_COUNT and w not in STOP_WHOLE
|
||
and w[0] not in STOP_EDGE and w[-1] not in STOP_EDGE}
|
||
# Поглощение: короткая n-грамма выбрасывается, если длинная её содержит и почти так же часта.
|
||
out = {}
|
||
for w, c in sorted(cand.items(), key=lambda kv: (-len(kv[0]), -kv[1])):
|
||
if any(w in longer and c <= 1.25 * cc for longer, cc in out.items()):
|
||
continue
|
||
out[w] = c
|
||
return sorted(out.items(), key=lambda kv: -kv[1])[:N_TERMS]
|
||
|
||
|
||
def kwic(term: str, n: int = KWIC_N) -> list[str]:
|
||
text = source_text()
|
||
out, pos = [], 0
|
||
for _ in range(n):
|
||
i = text.find(term, pos)
|
||
if i < 0:
|
||
break
|
||
out.append(text[max(0, i - KWIC_W):i + len(term) + KWIC_W].replace("\n", " "))
|
||
pos = i + len(term) + 200
|
||
return out
|
||
|
||
|
||
def _payload() -> str:
|
||
rows = []
|
||
for t, c in mine():
|
||
rows.append(f"term: {t}\ncount: {c}\n" + "\n".join(f"ctx: {x}" for x in kwic(t)))
|
||
return "\n\n".join(rows)
|
||
|
||
|
||
def cmd_mine() -> None:
|
||
ms = mine()
|
||
print(f"кандидатов {len(ms)} (порог частоты {MIN_COUNT}, знаков в срезе {len(source_text()):,})")
|
||
for t, c in ms:
|
||
print(f" {t:6s} {c:4d} {kwic(t, 1)[0][:70] if kwic(t, 1) else ''}")
|
||
|
||
|
||
def cmd_ask() -> None:
|
||
core, _few, user = PR.load_template(TERMINOLOGIST)
|
||
brief = PR.BRIEF_ZH
|
||
msgs = [{"role": "system", "content": PR.render(core, brief, "", "")},
|
||
{"role": "user", "content": PR.render(user, brief, _payload(), "")}]
|
||
led = MONEY.Guarded("Ф0", default_expect=0.02)
|
||
base, env, *_ = CANDIDATES[TERM_MODEL]
|
||
cl = OpenAI(api_key=os.environ[env], base_url=base, timeout=600)
|
||
kw = dict(model=TERM_MODEL, messages=msgs, max_completion_tokens=4000,
|
||
extra_body={"reasoning_effort": "low"})
|
||
rec = MONEY.purchase(led, "tp0-bank-terminolog", TERM_MODEL, cl, kw, role="terminologist")
|
||
if rec.get("skipped"):
|
||
print("⛔ потолок Ф0 — терминолог не куплен")
|
||
return
|
||
print(rec.get("content", "")[:4000])
|
||
print(f"\nfinish={rec.get('finish')} ${rec.get('cost_usd', 0):.6f} · "
|
||
f"потрачено Ф0 ${led.spent():.6f}")
|
||
|
||
|
||
def _rx(dst: str) -> str:
|
||
"""Регекс канонной формы под русское словоизменение.
|
||
|
||
⚠ Правка ПОСЛЕ фриза Ф0 и ПОСЛЕ покупки терминолога (то есть $0 и на результат прогона не
|
||
влияет), объявляется как девиация. Первая версия усекала только слова от 6 знаков, и короткие
|
||
склоняемые слова канона («род Цинь») не находились в форме «рода Цинь» — метрика покрытия
|
||
занижалась бы на всех армах одинаково, но занижалась.
|
||
"""
|
||
parts = []
|
||
for raw in dst.split():
|
||
w = re.escape(raw)
|
||
if len(raw) >= 4:
|
||
parts.append(w[:-2] + r"\w*")
|
||
elif len(raw) == 3:
|
||
parts.append(w[:-1] + r"\w*")
|
||
else:
|
||
parts.append(w)
|
||
return r"\s+".join(parts)
|
||
|
||
|
||
# ⚠ РУЧНЫЕ ПРАВКИ КАНОНА. Пусто = терминолог принят как есть. Каждая строка — решение СЕССИИ,
|
||
# не владельца; основания печатаются `--canon` и едут в отчёт.
|
||
MANUAL: dict[str, tuple[str, str]] = {}
|
||
# Выброшено из банка: это общеязыковая лексика, а не термины книги. Канон на них сделал бы
|
||
# метрику покрытия замером словарного совпадения, а не терминологической дисциплины.
|
||
DROP = {"瞬间", "声音", "仿佛", "庭院"}
|
||
|
||
|
||
def cmd_canon() -> None:
|
||
f = OUT / "tp0-bank-terminolog.json"
|
||
if not f.exists():
|
||
print("нет ответа терминолога — сначала --ask")
|
||
return
|
||
ans = json.loads(f.read_text(encoding="utf-8"))["content"]
|
||
got: dict[str, str] = {}
|
||
for line in ans.splitlines():
|
||
if "\t" not in line:
|
||
continue
|
||
src, dst = (x.strip() for x in line.split("\t", 1))
|
||
if src and dst and "TM-NO-DST" not in dst:
|
||
got[src] = dst
|
||
terms = {}
|
||
for src, dst in got.items():
|
||
if src in DROP:
|
||
continue
|
||
if src in MANUAL:
|
||
dst, why = MANUAL[src]
|
||
else:
|
||
why = ""
|
||
terms[src] = dict(dst=dst, rx=_rx(dst), manual=bool(why), why=why)
|
||
for src, (dst, why) in MANUAL.items():
|
||
terms.setdefault(src, dict(dst=dst, rx=_rx(dst), manual=True, why=why))
|
||
PR.BANK_FILE.write_text(json.dumps(
|
||
dict(book=M.BOOK["title"], signed=False,
|
||
note="ручной канон сессии, НЕ подпись владельца (образец D39.47)",
|
||
model=TERM_MODEL, dropped=sorted(DROP), terms=terms),
|
||
ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f"банк собран: {len(terms)} терминов · ручных правок {sum(1 for v in terms.values() if v['manual'])}"
|
||
f" · выброшено {len(DROP)}")
|
||
cmd_show()
|
||
|
||
|
||
def cmd_show() -> None:
|
||
b = PR.bank()
|
||
if not b:
|
||
print("банка нет")
|
||
return
|
||
src = source_text()
|
||
print(f"{'терм':8s}{'канон':34s}{'в срезе':>9s} регекс")
|
||
for t, v in sorted(b.items(), key=lambda kv: -src.count(kv[0])):
|
||
print(f"{t:8s}{v['dst']:34s}{src.count(t):9d} {v['rx']}"
|
||
+ (" ← ручная правка" if v.get("manual") else ""))
|
||
|
||
|
||
def coverage(source: str, out: str) -> tuple[int, int]:
|
||
"""(попаданий, знаменатель) канонной формы — метрика эксп-21 §16.1, счётная.
|
||
|
||
Знаменатель — упоминания термина В ИСХОДНИКЕ; попадания капятся знаменателем, иначе
|
||
многословный выход набирал бы покрытие повторами.
|
||
"""
|
||
k = s = 0
|
||
for t, v in PR.bank().items():
|
||
n = source.count(t)
|
||
if n:
|
||
s += n
|
||
k += min(len(re.findall(v["rx"], out, re.I)), n)
|
||
return k, s
|
||
|
||
|
||
def selfcheck() -> int:
|
||
bad = 0
|
||
|
||
def chk(n: str, ok: bool, d: str = "") -> None:
|
||
nonlocal bad
|
||
bad += not ok
|
||
print(f"[{'OK ' if ok else 'ПРОВАЛ'}] {n}" + (f" {d}" if d else ""))
|
||
|
||
chk("регекс усекает словоизменение", bool(re.search(_rx("истинная ци"), "истинной ци", re.I)))
|
||
chk("регекс не ловит чужое слово", not re.search(_rx("апертура"), "аперитив", re.I))
|
||
chk("регекс держит многословную форму",
|
||
bool(re.search(_rx("море истинной ци"), "Море истинной ци", re.I)))
|
||
# Регрессии на реальные формы банка: короткое склоняемое слово и прилагательное.
|
||
for canon, form in (("род Цинь", "роду Цинь"), ("род Цинь", "рода Циня"),
|
||
("левый канцлер", "левого канцлера"), ("глава рода", "главе рода"),
|
||
("Цинь Фэн", "Цинь Фэну"), ("Су Цинсюэ", "Су Цинсюэ"),
|
||
("Гу Хэ", "Гу Хэ"), ("Фэн-лао", "Фэн-лао")):
|
||
chk(f"«{canon}» находится в форме «{form}»", bool(re.search(_rx(canon), form, re.I)))
|
||
chk("канон не ловит другой род", not re.search(_rx("род Цинь"), "рода Су", re.I))
|
||
b = PR.bank()
|
||
if b:
|
||
t = next(iter(b))
|
||
k, s = coverage(source_text(), b[t]["dst"] * 3)
|
||
chk("покрытие не превышает знаменатель", k <= s, f"{k}/{s}")
|
||
chk("покрытие пустого выхода = 0", coverage(source_text(), "")[0] == 0)
|
||
print(f"\n{'БАНК-ОСНАСТКА ЧИСТА' if not bad else f'ПРОВАЛОВ: {bad}'}")
|
||
return bad
|
||
|
||
|
||
if __name__ == "__main__":
|
||
a = sys.argv[1:] or ["--mine"]
|
||
fn = {"--mine": cmd_mine, "--ask": cmd_ask, "--canon": cmd_canon, "--show": cmd_show,
|
||
"--selfcheck": lambda: sys.exit(1 if selfcheck() else 0)}.get(a[0])
|
||
fn() if fn else print(__doc__)
|