textmachine/eval/dovodka/podacha.py

722 lines
52 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""ПОДАЧА ВХОДА КРИТИКУ — что решает цену стадии: СЛОВА инструкции или ФОРМА подачи. Платный.
⚠ ОТКУДА ВОПРОС. Разбор токенов по ролям (журнал §14.2) дал разрыв, который никто не заказывал:
китайский смысловой критик думает 1 084 токена, английский — 9 576 при вдвое КОРОТКОМ входе.
Объяснение нашлось замером структуры (§14.3): китайская вебновелла набрана ПО ПРЕДЛОЖЕНИЮ НА
СТРОКУ, и черновик её разбивку скопировал (77 строк исходника → 78 абзацев перевода, 0.96:1);
английский приходит сплошным блоком (1 строка → 8 абзацев, 0.12:1). То есть дорого не «сличать
два текста», а сличать два НЕВЫРОВНЕННЫХ текста. У DeepSeek размышление тарифицируется по цене
ВЫХОДА, поэтому 85% цены дорогой клетки — то, что модель надумала про себя.
⚠⚠ ЭТО ОДНО НАБЛЮДЕНИЕ, И ОБЪЯСНИЛ ЕГО Я ЖЕ. Норма, заведённая разговором 17.08: прежде чем
строить схему поверх наблюдения, проверить НАБЛЮДЕНИЕ отдельным дешёвым замером. Здесь ничего не
выравнивается и не размечается — меняется только ПОДАЧА одного и того же материала.
⛔⛔ НАХОДКА, СДЕЛАННАЯ ПРИ ПОСТРОЙКЕ ЭТОГО ЗАМЕРА И МЕНЯЮЩАЯ ЕГО ЦЕЛЬ. Английский исходник
приходит критику одной строкой НЕ ПОТОМУ, что так устроен английский или книга, а потому, что так
устроен НАШ ЭКСТРАКТОР: `pair_en.raw_text()` сносит ВСЕ теги регексом `<[^>]+>` → пробел, вместе
с `</p>`. Замер по самому epub: в книге **11 447 абзацев, медиана 62 знака**; наш плоский текст
несёт 1 720 строк вместо 13 243. На единицу в 1642 знака приходится ~26 авторских абзацев, а мы
подаём их одним блоком. То есть невыровненность английской пары — АРТЕФАКТ НАШЕЙ НАРЕЗКИ, и
чинится он бесплатно, у нас же. Проверено: текст всех 16 английских единиц находится в абзацной
версии побайтно после нормализации пробелов, значит структуру можно вернуть, НЕ трогая
зафризенный экстрактор закрытой оси (менять его нельзя — uid единиц есть функция его текста).
ЧЕТЫРЕ ВАРИАНТА. Исходник и черновик во всех четырёх несут ОДИН И ТОТ ЖЕ текст; меняется ПОДАЧА:
P0 БАЗА нынешний промт критика, исходник как есть (одним блоком)
PF ФРАЗА + правдивое описание параллельности текстов (идея владельца 17.08)
PA АБЗАЦЫ база + исходнику ВОЗВРАЩЕНЫ авторские абзацы из epub (перевод не трогаем)
PB ОБА фраза + абзацы
⚠ ПОЧЕМУ НЕ «РАЗБИТЬ ПО ПРЕДЛОЖЕНИЮ НА СТРОКУ», КАК ПЛАНИРОВАЛОСЬ. Замечание владельца 20.08:
осмысленная разметка — это ГРУППИРОВКА по смыслу, а не более мелкое дробление; дробить мельче
значит увеличить число кусков, которые модель обязана сопоставить. Плюс замер: медиана авторского
абзаца у Кристоффа 62 знака, то есть примерно одно предложение, и синтетическая разбивка была бы
почти побайтным близнецом `PA` — второй оплаченной копией того же условия. Авторские абзацы и
честнее (это структура, которую написал автор), и продуктово осмысленнее (это то, что можно
починить в проде одной правкой экстрактора).
У КИТАЙСКОЙ ПАРЫ ВАРИАНТОВ ДВА, И ЭТО НЕ ЭКОНОМИЯ. Её исходник УЖЕ разбит по предложению на
строку — свойство материала, замеренное: 78 строк по 24 знака на главу. Возвращать абзацы нечему,
`PA`/`PB` там были бы побайтной копией `P0`/`PF`, то есть двумя оплаченными близнецами.
Селфтест это ПРОВЕРЯЕТ на живом сырье, а не верит комментарию.
⚠ Японская пара сюда не входит, но замерена тем же счётом и ложится к китайской: 19 строк по
43 знака, черновик 19 абзацев — отношение 1.00:1, самое выровненное из трёх книг. Значит
«дорогой режим» есть у ОДНОЙ книги из трёх, и именно у той, которую испортили мы.
⚠ ФОРМУЛИРОВКА ФРАЗЫ — ТОЛЬКО ПРАВДА, И ЭТО НЕ ЭТИКА, А МЕХАНИКА. «Предложение за предложением»
было бы ЛОЖЬЮ: наш же промт редактора обязывает сливать абзацы (замер: 77 строк исходника → 43
абзаца отредактированного). Соврём — модель будет искать соответствие, не найдёт и потратит
БОЛЬШЕ, а не меньше.
⚠⚠ РИСК, РАДИ КОТОРОГО ЗАВЕДЕНЫ ПОСАДКИ. Фраза про полное покрытие исходника может заставить
критика ПЕРЕСТАТЬ ИСКАТЬ ПРОПУСКИ — а потерянный кусок один из главных классов дефекта. Выигрыш
в токенах, купленный слепотой к целому классу, не выигрыш. Поэтому в каждый черновик садятся ДВА
дефекта с ИЗВЕСТНЫМ АДРЕСОМ (`plants.json`, пишется ДО первой покупки):
ПРОПУСК из черновика удаляется целый абзац — тот самый класс, что под угрозой
ИНВЕРСИЯ снимается отрицание — смысл переворачивается, текст остаётся грамотным
⇒ НИКТО НИЧЕГО НЕ РАЗМЕЧАЕТ РУКАМИ. Правильный ответ известен ДО прогона, потому что порчу внесли
мы сами; «нашёл или нет» считается детерминированно. Вопрос владельца 20.08 («а кто данные
разметит в таблицу? ты сам?») закрыт устройством замера, а не обещанием.
⚠ И У ДЕТЕКТОРА ЕСТЬ НУЛЕВАЯ МОДЕЛЬ — норма Д0.6 требует от всякого классификатора печатать
частоту срабатывания на СЛУЧАЙНОМ входе. Здесь она бесплатна: те же главы уже отсужены критиком
по НЕиспорченному черновику (клетки `A3`/`E3`), и детектор гоняется по ним. Сколько раз он
«находит» посадку там, где её не было, — и есть его шансовый пол.
⚠ ВАРИАНТЫ СТРЕЛЯЮТСЯ ВПЕРЕМЕШКУ ВНУТРИ ГЛАВЫ, И ЭТО НЕ АККУРАТНОСТЬ. Квирк-бюллетень (:110)
несёт замер: шесть дефолтных розыгрышей ОДНОГО запроса, разнесённых по времени, дали строго
растущий ряд размышления 1952 → 9104 → 10818 → 11157 → 13421 → 15720 знаков. Купи мы сначала все
`P0`, потом все `PF` — этот дрейф и стал бы «эффектом». Порядок вариантов внутри главы к тому же
перетасован детерминированно от соли, чтобы и позиция в очереди не доставалась одному варианту.
⚠ ДВЕ МОДЕЛИ — вопрос владельца 20.08 «это только дипсика касается?». `deepseek-v4-pro` (жилец) и
`glm-5` (настоящая альтернатива редакторского тира: другой вендор, цены не поднимал).
⚠ `glm-5` идёт с ВКЛЮЧЁННЫМ размышлением, и это ОБЪЯВЛЕННОЕ отступление от боевой конфигурации:
`roster.THINK` гасит ему thinking (`extra_body {"thinking": {"type": "disabled"}}`), а мы меряем
ровно размышление — с выключенным мерить было бы нечего. Цена отступления: у glm thinking-ON
раньше не влезал в таймаут (квирки :52), и это может выстрелить. Выстрелит — печатается как
результат, а не прячется.
⚠ ГРОК И GEMINI ОТВЕРГНУТЫ ДО ПОКУПКИ, с основанием: у grok размышление гуляет ×163 на побайтно
одном входе (квирки :69 — 98 · 8000 · 14014 · 16000 на одном куске), при четырёх главах этот шум
съест любой эффект; Gemini на нашем материале массово падает в контент-фильтр (эксп-21 §1/§8).
Запуск: podacha.py --dry | --selftest | --plants | --buy [zh|en] | --score
"""
from __future__ import annotations
import hashlib
import importlib.util
import json
import re
import statistics as st
import sys
from pathlib import Path
REPO = Path("/home/ubuntu/projects/textmachine")
ZONE = Path(__file__).resolve().parent
for d in ("dovodka", "editor_tier", "tenant_panel", "role_topology", "bank_arbitration"):
sys.path.insert(0, str(REPO / "eval" / d))
def _load(name: str, path: Path):
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
# ⚠ Контур берётся У ДРАЙВЕРА английской оси, а не грузится вторым экземпляром: `en_axis` держит
# СВОЙ `contour.py`, и `wire()` настраивает хуки именно в нём. Свой второй экземпляр остался бы с
# китайскими хуками, английские черновики читались бы китайским путём и возвращали пусто —
# английская ось молча стала бы пустой. Класс уже стоил фазе оплаченных клеток, поймано `--dry`.
EN = _load("en_axis_p", ZONE / "en_axis.py")
C = EN.C
MONEY, PR, ROSTER, ED22, PARA = C.MONEY, C.PR, C.ROSTER, C.ED22, C.PARA
OUT = MONEY.OUT
ZK = _load("zakhod_p", ZONE / "zakhod.py") # гейт перелома цены и пред-полётная проба
EPUB_EN = Path.home() / "books" / "Kristoff_Jay_-_Empire_of_the_Vampire_3_-_Empire_of_the_Dawn.epub"
SALT = "podacha-2026-08-20" # соль отбора глав и порядка вариантов; впечатана ДО покупок
N_UNITS = 4 # решение владельца 20.08: «восемь глав слишком, сократи до 4»
MODELS = ("deepseek-v4-pro", "glm-5")
MAX_OUT = 32000 # как в заходе: обрыв на потолке стоит денег и не даёт данных
TRUTH = OUT / "podacha-plants.json" # правильный ответ; пишется ДО первой покупки
# ⚠ ПАРА ЖИВЁТ В ДАННЫХ. Ни одна ветка ниже не спрашивает «а это английский?»: возврат абзацев —
# такой же параметр пары, как её буква тега или касса. Пара, которой в репозитории ещё нет,
# заводится строкой здесь и не требует правки ни одной функции.
PAIRS = {
"zh": {"letter": "k", "phase": "ФД-K", "crit": "A3", "vars": ("P0", "PF"),
"wire": lambda: C.configure(PARA.ZH), "units": lambda: C.units_ok(),
"reflow": None}, # исходник уже по предложению на строку — возвращать нечего
"en": {"letter": "l", "phase": "ФД-L", "crit": "E3", "vars": ("P0", "PF", "PA", "PB"),
"wire": EN.wire, "units": EN.units,
"reflow": lambda src: restore_paragraphs(src, EPUB_EN)},
}
ALL_VARS = ("P0", "PF", "PA", "PB")
WHAT = {"P0": "база — нынешний промт критика",
"PF": "+ ФРАЗА о параллельности текстов (идея владельца)",
"PA": "+ исходнику ВОЗВРАЩЕНЫ авторские абзацы",
"PB": "фраза И абзацы вместе"}
# ── ФРАЗА. Каждое утверждение проверяемо, ни одного обещания сверх правды ─────────────────────
# «должен покрывать», а не «покрывает»: второе было бы обещанием, которого мы не даём, и оно же
# снимало бы с критика обязанность искать пропуски. Первое называет ТРЕБОВАНИЕ — то самое, из-за
# которого пропуск и есть ошибка.
PARALLEL_NOTE = (
"\n\nКАК СООТНОСЯТСЯ ЭТИ ДВА ТЕКСТА. Они идут ПАРАЛЛЕЛЬНО: порядок изложения в переводе тот "
"же, что в исходнике, и перевод ДОЛЖЕН покрывать исходник целиком. При этом членение на "
"абзацы и предложения может отличаться — переводчику разрешено сливать и дробить, и это не "
"ошибка. Читай оба текста параллельно, сверяя ход за ходом, и НЕ ищи точных пар предложений.")
# ── ВОЗВРАТ АВТОРСКИХ АБЗАЦЕВ ────────────────────────────────────────────────────────────────
# ⚠ ЗАФРИЗЕННЫЙ ЭКСТРАКТОР ЗАКРЫТОЙ ОСИ НЕ ТРОГАЕТСЯ. `pair_en.raw_text()` определяет uid единиц
# (`pair_en.py:144` предупреждает прямо), и правка его пере-именовала бы всю английскую ось —
# то есть обнулила бы отсуженное. Поэтому абзацы восстанавливаются ЗДЕСЬ и только для подачи:
# книга читается второй раз, с сохранением границ `</p>`, `</div>`, `<br>`; текст единицы ищется
# в ней по нормализованным пробелам, и возвращается тот же самый текст — но с переводами строк.
# ⚠ Ни знака не добавляется и не теряется: селфтест сверяет обе версии побайтно после снятия
# пробельных различий. Если единица не нашлась — возвращается исходная подача, и число ненайденных
# печатается; молча подменять вариант нельзя, иначе смета оплатит близнецов.
_WS = re.compile(r"\s+")
_para_cache: dict[str, tuple[str, str, list[int]]] = {}
def _epub_paragraphed(path: Path) -> tuple[str, str, list[int]]:
"""(текст с абзацами, его нормализованная копия, карта позиций). Считается один раз."""
key = str(path)
if key in _para_cache:
return _para_cache[key]
import html as _html # noqa: PLC0415
import zipfile # noqa: PLC0415
z = zipfile.ZipFile(path)
parts = []
for n in sorted(x for x in z.namelist() if x.endswith((".xhtml", ".html", ".htm"))):
raw = z.read(n).decode("utf-8", "ignore")
raw = re.sub(r"(?i)</(p|div|h[1-6])\s*>", "\n", raw)
raw = re.sub(r"(?i)<br\s*/?>", "\n", raw)
parts.append(re.sub(r"[ \t]+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", raw))))
text = "\n".join(parts)
norm, idx = [], []
prev_ws = True
for i, ch in enumerate(text):
if ch.isspace():
if not prev_ws:
norm.append(" ")
idx.append(i)
prev_ws = True
else:
norm.append(ch)
idx.append(i)
prev_ws = False
_para_cache[key] = (text, "".join(norm).strip(), idx)
return _para_cache[key]
def restore_paragraphs(src: str, epub: Path) -> str:
"""Тот же текст единицы, но с авторскими границами абзацев. Пусто — если не нашёлся."""
text, norm, idx = _epub_paragraphed(epub)
want = _WS.sub(" ", src).strip()
i = norm.find(want)
if i < 0:
return ""
a, b = idx[i], idx[min(i + len(want) - 1, len(idx) - 1)] + 1
return "\n".join(x.strip() for x in text[a:b].split("\n") if x.strip())
# ── ПОСАДКИ: ДВА ДЕФЕКТА С ИЗВЕСТНЫМ АДРЕСОМ ─────────────────────────────────────────────────
# ⚠ Классы выбраны не воображением. «Снятое отрицание» замерено на этом же сырье как ложащееся на
# 31 единицу из 31 (норма маржевого декоя, `sud.MARGIN_PLANTS`): меняется одно слово, текст
# остаётся связным и грамотным, ошибка ТОЛЬКО смысловая. «Удалённый абзац» — тот самый класс,
# которому угрожает фраза про полное покрытие; он и есть предмет страховки.
_NEG = re.compile(r"\е\s+([а-яё]{3,}(?:л|ла|ло|ли|ет|ит|ют|ат|ят))\b")
MIN_PARA = 120 # абзац короче не удаляем: пропажу двух слов не заметит и человек
# ⚠ ВЕРХНЯЯ ГРАНИЦА ЗАВЕДЕНА ВЫЧИТКОЙ ПОСАДОК ПЕРЕД ПОКУПКОЙ, И ЭТО НЕ ПЕДАНТИЗМ. Первая редакция
# знала только нижний порог, и на английской единице `100b088f71` под удаление попал абзац в
# 1602 знака при черновике в 1642 — то есть посадка сносила ПОЧТИ ВЕСЬ текст. Критик увидел бы
# пустую клетку, «нашёл пропуск» стало бы тавтологией, а замер токенов на этой единице мерил бы
# другую задачу. Пропажа обязана быть НЕЗАМЕТНОЙ на глаз в объёме — иначе это не тонкая посадка,
# а второй грубый декой (ровно та ошибка, за которую в паке 23 контроль оказался пуст).
MAX_DROP_SHARE = 0.25 # удаляемый абзац — не больше четверти черновика
def plant(draft: str, uid: str) -> tuple[str, dict]:
"""Испорченный черновик и ПРАВИЛЬНЫЙ ОТВЕТ к нему. Детерминированно от соли и uid.
⚠ Возвращает не только адреса, но и УЛИКИ — слова из удалённого абзаца и оба варианта фразы с
отрицанием. По ним детектор потом решает «нашёл или нет», и решает без единой ручной разметки.
"""
paras = [p for p in draft.split("\n") if p.strip()]
cand = [i for i, p in enumerate(paras)
if MIN_PARA <= len(p) <= len(draft) * MAX_DROP_SHARE]
truth: dict = {"uid": uid}
if cand:
# середина списка кандидатов, сдвинутая хешем: ни первый абзац (там имена и зачин), ни
# последний (там развязка) — оба слишком заметны и сделали бы посадку грубой
h = int(hashlib.sha256(f"{SALT}|{uid}|drop".encode()).hexdigest(), 16)
i = cand[len(cand) // 4 + h % max(1, len(cand) // 2)]
truth["dropped"] = paras[i]
paras = paras[:i] + paras[i + 1:]
out = "\n\n".join(paras)
m = _NEG.search(out)
if m:
truth["negation_before"] = m.group(0)
truth["negation_after"] = m.group(1)
out = out[:m.start()] + m.group(1) + out[m.end():]
return out, truth
def truth_all() -> dict:
return json.loads(TRUTH.read_text(encoding="utf-8")) if TRUTH.exists() else {}
def write_truth() -> dict:
"""Правильный ответ на диск ДО первой покупки. Существующий НЕ переписывается.
⚠ Пере-запись после ответов означала бы, что «правильный ответ» стал функцией того, что модель
ответила. Ровно этот класс — раскладка меток, пере-эмитированная поверх отсуженной оси, — уже
стоил фазе аварии, и починка была в том, что ключ пишется один раз и до ответов.
"""
if TRUTH.exists():
return truth_all()
key: dict = {}
for p, cfg in PAIRS.items():
cfg["wire"]()
for u in chosen(p):
_t, tr = plant(C.base_draft(u["uid"]), u["uid"])
key[f"{p}|{u['uid']}"] = tr
TRUTH.write_text(json.dumps(key, ensure_ascii=False, indent=1), encoding="utf-8")
print(f"правильный ответ записан ДО покупок: {TRUTH} ({len(key)} глав)")
return key
# ── ЗАПРОС ───────────────────────────────────────────────────────────────────────────────────
def crit_msgs(p: str, src: str, draft: str, var: str) -> list[dict]:
"""Запрос критика в заданном варианте подачи.
⚠ Системный промт берётся у БОЕВОГО критика (`contour.crit_msgs`) и только дополняется: если
собрать его здесь заново, замер сравнивал бы два разных промта, а не две подачи одного.
"""
if var not in ALL_VARS:
raise SystemExit(f"⛔ вариант {var!r} неизвестен; знаю {' '.join(ALL_VARS)}")
cfg = PAIRS[p]
cfg["wire"]()
shown = src
if var in ("PA", "PB"):
if not cfg["reflow"]:
raise SystemExit(f"у пары {p} возврат абзацев не заведён — вариант {var} невозможен")
shown = cfg["reflow"](src)
if not shown.strip():
raise SystemExit(f"⛔ авторские абзацы для этой единицы не нашлись — вариант {var} "
"стал бы побайтным близнецом базы, покупка отказана")
m = C.crit_msgs(shown, draft)
if var in ("PF", "PB"):
m[0]["content"] = m[0]["content"].rstrip() + PARALLEL_NOTE
return m
def tag(p: str, var: str, model: str, uid: str) -> str:
short = {"deepseek-v4-pro": "ds", "glm-5": "gl"}[model]
return f"dv-{PAIRS[p]['letter']}-{var.lower()}{short}-{uid}"
def cell(tg: str) -> dict | None:
f = OUT / f"{tg}.json"
if not f.exists():
return None
d = json.loads(f.read_text(encoding="utf-8"))
return d if d.get("finish") == "stop" and (d.get("content") or "").strip() else None
def call_kwargs(model: str, msgs: list[dict]) -> dict:
"""Боевая конфигурация модели — КРОМЕ явно объявленного отступления по размышлению glm-5."""
kw = ROSTER.call_kwargs(model, msgs, max_out=MAX_OUT)
if model == "glm-5":
kw.pop("extra_body", None) # вендор-дефолт = thinking ON; см. шапку файла
return kw
# ── ОТБОР ГЛАВ ───────────────────────────────────────────────────────────────────────────────
def pool(p: str) -> list[dict]:
"""Главы, где есть черновик, боевой перевод И ответ критика по НЕиспорченному входу.
Последнее обязательно: эта клетка — нулевая модель детектора, и без неё замер не может
отличить «детектор сработал на посадке» от «детектор срабатывает всегда».
"""
cfg = PAIRS[p]
cfg["wire"]()
out = []
for u in cfg["units"]():
d = C.base_draft(u["uid"])
if not d.strip():
continue
base = OUT / f"{C.tag(cfg['crit'], u['uid'], 'crit')}.json"
if not (base.exists()
and (json.loads(base.read_text(encoding="utf-8")).get("content") or "").strip()):
continue
# ⚠ ОБЕ ПОСАДКИ ОБЯЗАНЫ ЛОЖИТЬСЯ, И ЭТО УСЛОВИЕ ОТБОРА, А НЕ ПОСТ-ФИЛЬТР. Глава, куда
# тонкая пропажа не помещается (все абзацы либо короче порога, либо больше четверти
# текста), даёт клетку без правильного ответа — и «не нашёл» на ней означает «нечего было
# находить». Такие главы в замер не идут вовсе, а не молча портят долю.
after, tr = plant(d, u["uid"])
if not tr.get("dropped") or not tr.get("negation_before"):
continue
# авторские абзацы обязаны находиться — иначе вариант станет близнецом базы
if cfg["reflow"] and not cfg["reflow"](u["source"]).strip():
continue
out.append(u)
return out
def chosen(p: str) -> list[dict]:
"""N_UNITS глав, отобранных ДЕТЕРМИНИРОВАННО от объявленной соли, а не по длине.
Длина коррелирует с трудностью, трудность — с тем, сколько модель думает; отбор по длине
покупал бы часть ответа заранее.
"""
us = pool(p)
return sorted(us, key=lambda u: hashlib.sha256(
f"{SALT}|{p}|{u['uid']}".encode()).hexdigest())[:N_UNITS]
def var_order(p: str, uid: str, model: str) -> list[str]:
"""Порядок вариантов ВНУТРИ главы — перетасован от соли. См. шапку про дрейф размышления."""
vs = list(PAIRS[p]["vars"])
return sorted(vs, key=lambda v: hashlib.sha256(
f"{SALT}|{p}|{uid}|{model}|{v}".encode()).hexdigest())
# ── ПОКУПКА ──────────────────────────────────────────────────────────────────────────────────
def buy(p: str) -> None:
ZK.price_gate()
cfg = PAIRS[p]
cfg["wire"]()
write_truth()
ZK.preflight(list(MODELS), cfg["phase"], cfg["letter"])
us = chosen(p)
key = truth_all()
led = MONEY.Guarded(cfg["phase"], default_expect=0.045)
print(f"\nпара {p} · глав {len(us)} · варианты {' '.join(cfg['vars'])} · "
f"модели {' '.join(MODELS)} · касса {cfg['phase']}")
for u in us:
uid, src = u["uid"], u["source"]
if f"{p}|{uid}" not in key:
raise SystemExit(f"у главы {uid} нет записанного правильного ответа — "
"покупка остановлена, посадки должны быть зафиксированы ДО неё")
draft, _tr = plant(C.base_draft(uid), uid)
for model in MODELS:
for var in var_order(p, uid, model):
tg = tag(p, var, model, uid)
if cell(tg):
continue
# ⚠ Гейт перелома цены — перед КАЖДОЙ клеткой, а не один раз на старте: прогон
# длится часами, и пересёкший 16:00 UTC цикл писал бы пост-переломные вызовы по
# июльскому прайсу.
ZK.price_gate(quiet=True)
rec = MONEY.purchase(led, tg, model, ED22.client(model),
call_kwargs(model, crit_msgs(p, src, draft, var)),
role="critic", uid=uid, arm=var, pair=p)
if rec.get("skipped"):
raise SystemExit(
f"⛔ ГАРД ОТКАЗАЛ на клетке {tg} — прогон ОСТАНОВЛЕН.\n"
" Резать панель или поднимать потолок решением сессии ЗАПРЕЩЕНО.\n"
" Это СТОП и вопрос владельцу: назвать недостающую сумму.")
if rec.get("finish") != "stop":
print(f"{tg}: finish={rec.get('finish')} "
f"{str(rec.get('error'))[:120]} — клетка объявляется несобранной")
MONEY.report()
# ── РАЗБОР ───────────────────────────────────────────────────────────────────────────────────
def findings(txt: str) -> list[str]:
"""Строки-находки критика. Тот же разбор, что у боевого контура, плюс фильтр согласий."""
lines = [ln.strip("-•* \t") for ln in txt.splitlines() if ln.strip().startswith(("-", "", "*"))]
return [ln for ln in lines if not ZK._is_confirmation(ln)] # noqa: SLF001
_WORD = re.compile(r"[А-Яа-яЁёA-Za-z]{4,}")
def _caught_drop(txt: str, tr: dict) -> bool:
"""Заметил ли критик УДАЛЁННЫЙ абзац. Детерминированно, без ручной разметки.
Правило: критик обязан сослаться на содержимое пропавшего куска — назвать хотя бы два редких
слова из него. Редкость меряется по самому тексту главы: слова, встречающиеся в удалённом
абзаце и НЕ встречающиеся в остальном черновике, — единственная улика, которая не могла
приехать из соседних абзацев.
"""
dropped = tr.get("dropped") or ""
uniq = tr.get("_uniq") or []
if not dropped or not uniq:
return False
return sum(1 for w in uniq if w in txt.lower()) >= 2
def _caught_neg(txt: str, tr: dict) -> bool:
"""Заметил ли критик снятое отрицание: процитировал испорченную форму или назвал отрицание."""
after = (tr.get("negation_after") or "").lower()
if not after:
return False
low = txt.lower()
return (after in low and ("отриц" in low or "не " + after in low)) or f"«{after}" in low
def _uniq_words(tr: dict, draft_after: str) -> list[str]:
"""Редкие слова удалённого абзаца — считаются ЗДЕСЬ, а не хранятся: правильный ответ не должен
зависеть от того, как мы сегодня решили считать редкость."""
d = (tr.get("dropped") or "").lower()
rest = draft_after.lower()
return sorted({w for w in _WORD.findall(d) if len(w) >= 5 and w not in rest})[:12]
def score() -> None:
key = truth_all()
if not key:
raise SystemExit("⛔ правильного ответа на диске нет — замер не разбирается")
for p, cfg in PAIRS.items():
cfg["wire"]()
us = chosen(p)
print("\n" + "=" * 78)
print(f"ПАРА {p.upper()} · глав {len(us)} · варианты {' '.join(cfg['vars'])}")
print("=" * 78)
for model in MODELS:
rows: dict = {}
for u in us:
uid = u["uid"]
tr = dict(key.get(f"{p}|{uid}") or {})
after, _ = plant(C.base_draft(uid), uid)
tr["_uniq"] = _uniq_words(tr, after)
for var in cfg["vars"]:
d = cell(tag(p, var, model, uid))
if not d:
continue
txt = d.get("content") or ""
fs = findings(txt)
rows.setdefault(var, {})[uid] = dict(
think=d.get("reasoning_tokens") or 0,
billed=ROSTER.billed_output(model, d.get("completion_tokens") or 0,
d.get("reasoning_tokens") or 0,
d.get("total_tokens") or 0,
d.get("prompt_tokens") or 0),
inp=d.get("prompt_tokens") or 0, cost=d.get("cost_usd") or 0,
n=len(fs), drop=_caught_drop(txt, tr), neg=_caught_neg(txt, tr))
if not rows:
print(f"\n {model}: клеток нет")
continue
print(f"\n === {model} ===")
print(f" {'вар':4s}{'кл':>3s}{'вход':>7s}{'РАЗМЫШЛ':>9s}{'оплач.вых':>10s}"
f"{'$/кл':>9s}{'находок':>9s}{'пропуск':>9s}{'отриц':>7s} что это")
for var in cfg["vars"]:
r = rows.get(var) or {}
if not r:
print(f" {var:4s} — клеток нет")
continue
med = lambda k: st.median([v[k] for v in r.values()]) # noqa: E731
print(f" {var:4s}{len(r):3d}{med('inp'):7.0f}{med('think'):9.0f}"
f"{med('billed'):10.0f}{med('cost'):9.5f}{med('n'):9.1f}"
f"{sum(v['drop'] for v in r.values()):5d}/{len(r):<3d}"
f"{sum(v['neg'] for v in r.values()):4d}/{len(r):<2d} {WHAT[var]}")
base = rows.get("P0") or {}
if base:
print("\n ПАРНАЯ РАЗНИЦА ПРОТИВ БАЗЫ (одна и та же глава, одна и та же модель):")
for var in cfg["vars"]:
if var == "P0" or var not in rows:
continue
ds = [rows[var][u]["billed"] - base[u]["billed"]
for u in rows[var] if u in base]
dn = [rows[var][u]["n"] - base[u]["n"] for u in rows[var] if u in base]
if not ds:
continue
print(f" {var:4s} оплаченный выход {st.mean(ds):+8.0f} токенов "
f"({st.mean(ds) / max(1, st.mean([base[u]['billed'] for u in base])):+.0%}) "
f"· находок {st.mean(dn):+.1f} · единиц {len(ds)}")
# НУЛЕВАЯ МОДЕЛЬ ДЕТЕКТОРА — по клеткам критика, где посадки НЕ БЫЛО
print("\n НУЛЕВАЯ МОДЕЛЬ ДЕТЕКТОРА (клетки критика по НЕиспорченному черновику):")
fp_d = fp_n = tot = 0
for u in us:
uid = u["uid"]
f = OUT / f"{C.tag(cfg['crit'], uid, 'crit')}.json"
if not f.exists():
continue
txt = json.loads(f.read_text(encoding="utf-8")).get("content") or ""
tr = dict(key.get(f"{p}|{uid}") or {})
after, _ = plant(C.base_draft(uid), uid)
tr["_uniq"] = _uniq_words(tr, after)
tot += 1
fp_d += _caught_drop(txt, tr)
fp_n += _caught_neg(txt, tr)
print(f" ложных «нашёл пропуск» {fp_d}/{tot} · ложных «нашёл отрицание» {fp_n}/{tot}")
print(" ⇒ выше этого пола детектор ничего не доказывает")
# ── СМЕТА И ПРОВЕРКИ ─────────────────────────────────────────────────────────────────────────
def price_of(p: str, model: str) -> float:
"""Ожидаемая цена клетки — из ЗАМЕРЕННЫХ токенов уже купленных клеток критика ЭТОЙ пары,
пере-оценённых ТЕКУЩИМ пином. Зашитых цен в файле нет: они врали вчетверо после пере-пина."""
toks = []
for f in OUT.glob(f"{C.tag(PAIRS[p]['crit'], '', 'crit')}*.json"):
if any(x in f.name for x in ("ERROR", "LENGTH", "ЭХО")):
continue
r = json.loads(f.read_text(encoding="utf-8"))
if r.get("finish") == "stop" and r.get("prompt_tokens"):
toks.append((r["prompt_tokens"], r.get("cached_tokens") or 0,
r.get("completion_tokens") or 0, r.get("reasoning_tokens") or 0,
r.get("total_tokens") or 0))
if not toks:
return 0.045
med = [st.median([x[i] for x in toks]) for i in range(5)]
return ROSTER.cost(model, int(med[0]), int(med[1]), int(med[2]), int(med[3]), int(med[4]))
def cmd_dry() -> None:
print(f"ПОДАЧА ВХОДА КРИТИКУ · соль {SALT} · глав на пару {N_UNITS}\n")
ZK.price_gate()
tot = 0.0
for p, cfg in PAIRS.items():
cfg["wire"]()
us, pl = chosen(p), pool(p)
print(f"\n=== пара {p} · пул {len(pl)} · взято {len(us)} · касса {cfg['phase']} ===")
sub = 0.0
for model in MODELS:
pr = price_of(p, model)
need = sum(1 for u in us for v in cfg["vars"]
if not cell(tag(p, v, model, u["uid"])))
sub += need * pr
print(f" {model:18s} надо {need:2d} клеток × ${pr:.5f} = ${need * pr:.4f}")
sp = MONEY.Guarded(cfg["phase"]).spent()
ceil = MONEY.CEILINGS[cfg["phase"]]
mark = "ВЛЕЗАЕТ" if sub <= ceil - sp else f"НЕ ВЛЕЗАЕТ (не хватает ${sub - (ceil - sp):.4f})"
print(f" ИТОГО ${sub:.4f} · потолок {cfg['phase']} ${ceil:.2f} · "
f"потрачено ${sp:.4f}{mark}")
tot += sub
print(f"\nВЕСЬ ЗАМЕР: ${tot:.4f} по текущему (пиковому) пину ростера.")
def cmd_plants() -> None:
"""Показать посадки глазами — обязательная вычитка ПЕРЕД покупкой."""
key = write_truth()
for p, cfg in PAIRS.items():
cfg["wire"]()
for u in chosen(p):
tr = key[f"{p}|{u['uid']}"]
after, _ = plant(C.base_draft(u["uid"]), u["uid"])
print(f"\n--- {p}/{u['uid']} ---")
print(f" УДАЛЁН абзац ({len(tr.get('dropped') or '')} знаков): "
f"{(tr.get('dropped') or '')[:160]}")
print(f" редкие слова-улики: {' '.join(_uniq_words(tr, after)) or 'НЕТ'}")
print(f" ОТРИЦАНИЕ: «{tr.get('negation_before', '')}» → "
f"«{tr.get('negation_after', '')}»")
def cmd_selftest() -> int:
bad = 0
def ck(n: str, ok: bool, d: str = "") -> None:
nonlocal bad
bad += not ok
print(f"[{'OK ' if ok else 'ПРОВАЛ'}] {n}" + (f" {d}" if d else ""))
for p, cfg in PAIRS.items():
cfg["wire"]()
us = chosen(p)
ck(f"{p}: глав ровно {N_UNITS}", len(us) == N_UNITS, f"их {len(us)}")
ck(f"{p}: отбор ДЕТЕРМИНИРОВАН", [u["uid"] for u in chosen(p)] == [u["uid"] for u in us])
u = us[0]
src, draft = u["source"], C.base_draft(u["uid"])
# ⚠ ГЛАВНОЕ ОБОСНОВАНИЕ ПАНЕЛИ — ПРОВЕРЯЕТСЯ НА ЖИВОМ СЫРЬЕ, А НЕ ВЕРИТСЯ КОММЕНТАРИЮ.
# «У китайской пары исходник уже разбит» — если это перестанет быть правдой, панель из
# двух вариантов молча станет неполной.
per_line = st.median([len(x) for x in src.split("\n") if x.strip()])
already = per_line < 60
ck(f"{p}: исходник {'УЖЕ разбит' if already else 'сплошной'}"
f"возврат абзацев {'НЕ нужен' if already else 'нужен'}, вариантов {2 if already else 4}",
len(cfg["vars"]) == (2 if already else 4) and bool(cfg["reflow"]) == (not already),
f"медиана строки {per_line:.0f} знаков")
if cfg["reflow"]:
# ⚠ ВОЗВРАТ АБЗАЦЕВ ПРОВЕРЯЕТСЯ НА ВСЕХ ВЗЯТЫХ ГЛАВАХ, а не на первой: единица,
# которая не нашлась в книге, сделала бы вариант близнецом базы — оплаченным дважды.
got = [cfg["reflow"](x["source"]) for x in us]
ck(f"{p}: авторские абзацы нашлись для ВСЕХ глав", all(g.strip() for g in got),
f"нашлось {sum(1 for g in got if g.strip())} из {len(got)}")
sp = got[0]
ck(f"{p}: возврат абзацев реально дробит исходник",
sp.count("\n") >= 5, f"строк {src.count(chr(10)) + 1}{sp.count(chr(10)) + 1}")
ck(f"{p}: возврат абзацев НЕ теряет и НЕ добавляет знаков",
re.sub(r"\s+", "", sp) == re.sub(r"\s+", "", src),
f"{len(re.sub(r'[^А-Яа-яA-Za-z]', '', sp))} против "
f"{len(re.sub(r'[^А-Яа-яA-Za-z]', '', src))} букв")
# четыре варианта обязаны быть ПОПАРНО РАЗНЫМИ запросами, иначе часть сметы — близнецы
msgs = {v: json.dumps(crit_msgs(p, src, draft, v), ensure_ascii=False)
for v in cfg["vars"]}
ck(f"{p}: варианты попарно различны", len(set(msgs.values())) == len(msgs))
# база обязана быть ПОБАЙТНО боевым критиком — иначе меряем два разных промта
ck(f"{p}: P0 — побайтно боевой критик",
msgs["P0"] == json.dumps(C.crit_msgs(src, draft), ensure_ascii=False))
ck(f"{p}: ФРАЗА уходит в системный промт и только туда",
PARALLEL_NOTE.strip()[:40] in json.loads(msgs["PF"])[0]["content"]
and PARALLEL_NOTE.strip()[:40] not in json.loads(msgs["PF"])[1]["content"])
ck(f"{p}: ФРАЗА не обещает построчного соответствия",
"предложение за предложением" not in PARALLEL_NOTE.lower()
and "НЕ ищи точных пар" in PARALLEL_NOTE)
ck(f"{p}: черновик во всех вариантах ОДИН И ТОТ ЖЕ",
len({json.loads(m)[1]["content"].split("ЧЕРНОВОЙ ПЕРЕВОД:")[-1]
for m in msgs.values()}) == 1)
# посадки: обе садятся, обе детерминированы, обе оставляют текст читаемым
after, tr = plant(draft, u["uid"])
ck(f"{p}: посадка ДЕТЕРМИНИРОВАНА", plant(draft, u["uid"])[0] == after)
# ⚠ ПО ВСЕМ ВЗЯТЫМ ГЛАВАМ, А НЕ ПО ПЕРВОЙ. Вычитка перед покупкой поймала посадку в 1602
# знака на четвёртой английской главе, тогда как на первой всё было в порядке: гейт,
# смотрящий на `us[0]`, сертифицировал бы негодный замер.
alld = [plant(C.base_draft(x["uid"]), x["uid"]) for x in us]
ck(f"{p}: абзац удалён на КАЖДОЙ главе",
all(t.get("dropped") and t["dropped"] not in a for a, t in alld),
" · ".join(str(len(t.get("dropped") or "")) for _a, t in alld) + " знаков")
ck(f"{p}: отрицание снято на КАЖДОЙ главе",
all(t.get("negation_before") and t["negation_before"] not in a for a, t in alld),
" · ".join(f"«{t.get('negation_before')}»" for _a, t in alld))
ck(f"{p}: у пропажи есть редкие слова-улики на КАЖДОЙ главе",
all(len(_uniq_words(t, a)) >= 2 for a, t in alld),
" · ".join(str(len(_uniq_words(t, a))) for a, t in alld))
ck(f"{p}: пропажа ТОНКАЯ — не больше четверти черновика НИ НА ОДНОЙ главе",
all(0 < len(C.base_draft(x["uid"])) - len(a)
<= len(C.base_draft(x["uid"])) * MAX_DROP_SHARE
for x, (a, _t) in zip(us, alld, strict=True)),
" · ".join(f"{(len(C.base_draft(x['uid'])) - len(a)) / max(1, len(C.base_draft(x['uid']))):.0%}"
for x, (a, _t) in zip(us, alld, strict=True)))
# ⚠ ДЕТЕКТОР ОБЯЗАН УМЕТЬ ПАДАТЬ. Гейт, который не может сказать «не нашёл», не сторожит
# ничего — норма, заведённая пустым контролем пака 23.
tr2 = dict(tr, _uniq=_uniq_words(tr, after))
ck(f"{p}: детектор пропажи МОЛЧИТ на пустом ответе", not _caught_drop("", tr2))
ck(f"{p}: детектор пропажи СРАБАТЫВАЕТ на тексте самого абзаца",
_caught_drop((tr.get("dropped") or "").lower(), tr2))
ck(f"{p}: детектор отрицания МОЛЧИТ на пустом ответе", not _caught_neg("", tr))
ck(f"{p}: детектор отрицания СРАБАТЫВАЕТ на прямой формулировке",
_caught_neg(f"«{tr.get('negation_after')} …» → снято отрицание", tr))
# теги: покрыты глобом своей кассы, различны по варианту и по модели
tgs = {tag(p, v, m, u["uid"]) for v in cfg["vars"] for m in MODELS}
ck(f"{p}: теги различны по варианту и модели",
len(tgs) == len(cfg["vars"]) * len(MODELS))
ck(f"{p}: теги покрыты глобом кассы {cfg['phase']}",
all(t.startswith(f"dv-{cfg['letter']}-") for t in tgs))
ck(f"{p}: фаза заведена в кассе", cfg["phase"] in MONEY.CEILINGS)
ck(f"{p}: нулевая модель детектора обеспечена (клетки критика по чистому входу есть)",
all((OUT / f"{C.tag(cfg['crit'], x['uid'], 'crit')}.json").exists() for x in us))
ck(f"{p}: порядок вариантов внутри главы перетасован, а не алфавитный",
len(cfg["vars"]) < 3 or any(var_order(p, x["uid"], MODELS[0]) != list(cfg["vars"])
for x in us))
ck("пары не делят буквы тегов", len({v["letter"] for v in PAIRS.values()}) == len(PAIRS))
ck("glm-5 идёт с ВКЛЮЧЁННЫМ размышлением (объявленное отступление)",
"extra_body" not in call_kwargs("glm-5", [{"role": "user", "content": "x"}]))
ck("deepseek идёт в БОЕВОЙ конфигурации, без отступлений",
call_kwargs("deepseek-v4-pro", [{"role": "user", "content": "x"}])
== ROSTER.call_kwargs("deepseek-v4-pro", [{"role": "user", "content": "x"}],
max_out=MAX_OUT))
ck("правильный ответ пишется ДО покупок и не перезаписывается",
"if TRUTH.exists()" in Path(__file__).read_text(encoding="utf-8"))
print(f"\n{'ЗАМЕР ГОДЕН К ПОКУПКЕ' if not bad else f'ПРОВАЛОВ: {bad}'}")
return bad
if __name__ == "__main__":
a = sys.argv[1:] or ["--dry"]
if a[0] == "--selftest":
sys.exit(1 if cmd_selftest() else 0)
elif a[0] == "--dry":
cmd_dry()
elif a[0] == "--plants":
cmd_plants()
elif a[0] == "--score":
score()
elif a[0] == "--buy":
for x in ([y for y in a[1:] if y in PAIRS] or list(PAIRS)):
buy(x)
else:
raise SystemExit(f"⛔ неизвестный режим {a[0]!r}; "
"знаю --dry --selftest --plants --buy --score")