diff --git a/eval/role_topology/_retest.py b/eval/role_topology/_retest.py new file mode 100644 index 00000000..bc107ce8 --- /dev/null +++ b/eval/role_topology/_retest.py @@ -0,0 +1,38 @@ +"""ПОВТОР-ТЕСТ ОДНОГО СУДЬИ: то же задание, ТА ЖЕ раскладка, новый судья. $0. + +⚠ Заведено адверсариальным ревью 07.08. Проходы 1 и 2 различаются РАСКЛАДКОЙ меток, поэтому их +расхождение смешивает судейский шум с позиционным. Здесь раскладка держится постоянной, значит +разность — чистый судейский шум. Это единственный контроль амендмента («разброс судьи на +повторе»), который до сих пор не был измерен ни разу. +""" +import json, statistics as st, sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent)) +import absjudge as A # noqa: E402 +import bakeoff as BO # noqa: E402 + +A.PASS = "retest" +A.ingest() +r = A._by_unit() +A.PASS = "abs2" +b = A._by_unit() +common = sorted(set(r) & set(b)) +d = [r[u][k]["errors"] - b[u][k]["errors"] for u in common for k in r[u] if k in b[u]] +# ⚠ ПРОВЕРЯТЬ НАДО РАСКЛАДКУ, А НЕ ТОЛЬКО ТЕКСТ. Ревью 07.08: печаталось «раскладка и текст +# тождественны», а сверялся только хеш текста — тот же тест давал True и на проходах с ЗАВЕДОМО +# разной раскладкой. Здесь сверяется и подпись текста, и номер слота. +sig = all(r[u][k]["sig"] == b[u][k]["sig"] for u in common for k in r[u] if k in b[u]) +slot = all(r[u][k]["slot"] == b[u][k]["slot"] for u in common for k in r[u] if k in b[u]) +print(f"\nединиц {len(common)} · клеток {len(d)} · текст тождествен: {sig} · РАСКЛАДКА тождественна: {slot}") +print(f" средняя разность {st.mean(d):+.2f} · sd {st.pstdev(d):.2f}") +print(f" ⇒ σ ОДНОГО СУДЬИ на повторе = {st.pstdev(d) / 2 ** 0.5:.2f}") +xs = [r[u][k]["errors"] for u in common for k in r[u] if k in b[u]] +ys = [b[u][k]["errors"] for u in common for k in r[u] if k in b[u]] +mx, my = st.mean(xs), st.mean(ys) +print(f" корреляция повтора: r={sum((a-mx)*(c-my) for a, c in zip(xs, ys)) / ((sum((a-mx)**2 for a in xs)*sum((c-my)**2 for c in ys))**0.5):+.3f}") +for a_, b_, w in A.FAMILY: + m1 = [r[u][b_]["errors"] - r[u][a_]["errors"] for u in common if a_ in r[u] and b_ in r[u]] + m2 = [b[u][b_]["errors"] - b[u][a_]["errors"] for u in common if a_ in b[u] and b_ in b[u]] + if m1: + print(f" {a_ + ' vs ' + b_:14s} повтор {st.mean(m1):+6.2f} исходно {st.mean(m2):+6.2f}" + f" |Δ| {abs(st.mean(m1)-st.mean(m2)):.2f}") diff --git a/eval/role_topology/absjudge.py b/eval/role_topology/absjudge.py index c29fd6a1..927d14c5 100644 --- a/eval/role_topology/absjudge.py +++ b/eval/role_topology/absjudge.py @@ -160,21 +160,35 @@ def build_unit(u: dict) -> tuple[str, dict]: return "\n".join(body), key -_PLANTS = [ - (r"\bне\s+(выдерж\w+)", r"легко \1", "инверсия полярности"), - (r"\bдесят(ь|и)\b", "сорок", "подмена величины"), - (r"\bсто\b", "триста", "подмена величины"), - (r"\bпоразительн(\w+)", r"взгрумительн\1", "выдуманное слово"), - (r"\bспокойн(\w+)", r"безмятежн\1 и невозмутительн\1", "выдуманное слово"), - (r"—\s*", '"', "сбой диалоговой вёрстки"), +# ⚠ ПРАВИЛА ПОСАДКИ ЖИВУТ ВНЕ РЕПОЗИТОРИЯ, рядом с ключом и солью. +# Адверсариальное ревью 07.08: пока `_PLANTS` лежали в коде, декой опознавался БЕЗ соли и БЕЗ +# ключа — он побайтная копия арма `A` с 1–6 подменами, сходство 0.99 против 0.90 у ближайшего +# конкурента, и ориентация снималась прогоном самой посадки на обоих кандидатах (10/10 верно). +# Значит читающий код судья опознавал и контроль, и арм `A` — базу четырёх контрастов из семи. +# Вынос соли закрывал один канал из двух, и не главный. Здесь закрывается второй. +_PLANTS_DEFAULT = [ + [r"\bне\s+(выдерж\w+)", r"легко \1", "инверсия полярности"], + [r"\bдесят(ь|и)\b", "сорок", "подмена величины"], + [r"\bсто\b", "триста", "подмена величины"], + [r"\bпоразительн(\w+)", r"взгрумительн\1", "выдуманное слово"], + [r"\bспокойн(\w+)", r"безмятежн\1 и невозмутительн\1", "выдуманное слово"], + [r"—\s*", '"', "сбой диалоговой вёрстки"], ] +def _plants() -> list: + f = KEYS / "PLANTS.json" + if not f.exists(): + f.parent.mkdir(parents=True, exist_ok=True) + f.write_text(json.dumps(_PLANTS_DEFAULT, ensure_ascii=False, indent=1), encoding="utf-8") + return json.loads(f.read_text(encoding="utf-8")) + + def _plant(text: str) -> tuple[str, int]: """Контролируемая деградация. Число РЕАЛЬНО применённых замен возвращается: клетка, где посадить не удалось, в декой не берётся, иначе «не пойман» будет означать «не было».""" out, n = text, 0 - for pat, rep, _c in _PLANTS: + for pat, rep, _c in _plants(): new, cnt = re.subn(pat, rep, out, count=1) if cnt: out, n = new, n + cnt @@ -215,21 +229,25 @@ def ingest() -> None: # ⚠ ИДЕНТИЧНОСТЬ СУДЬИ. Первая версия рига её не персистировала вовсе, и по диску нельзя было # ни посчитать согласие судей, ни разложить вердикты по семействам (требование промта стр.79). # Поймано адверсариальным ревью 07.08. Карта раздачи лежит рядом с ключом и судье не даётся. - jf = KEYS / "abs-JUDGES.json" + jf = KEYS / f"{PASS}-JUDGES.json" judges = json.loads(jf.read_text(encoding="utf-8")) if jf.exists() else {} VOTES.mkdir(parents=True, exist_ok=True) ok = bad = 0 reasons: dict[str, int] = {} - def fail(r: str) -> None: + def fail(r: str, stem: str = "") -> None: nonlocal bad bad += 1 reasons[r] = reasons.get(r, 0) + 1 + # ⚠ Отказ ОБЯЗАН снимать прежний голос: иначе негодный ответ оставлял на диске + # предыдущий разбор, и `_by_unit` подхватывал его как действительный. Ревью 07.08. + if stem and (VOTES / f"{stem}.json").exists(): + (VOTES / f"{stem}.json").unlink() for f in sorted(ANSW.glob("*.txt")): key = keys.get(f.stem) if not key: - fail("токена нет в ключе") + fail("токена нет в ключе", f.stem) continue txt = f.read_text(encoding="utf-8") rows = {} @@ -247,15 +265,27 @@ def ingest() -> None: rf"[ \t]*:|\Z)", txt, re.M | re.S) why = wm.group(1).strip() if wm else "" if len(ax) != len(J.AXES): - fail(f"не разобран блок {lab}") + fail(f"не разобран блок {lab}", f.stem) break # ⚠ ПОИМЁННО ПО ОСЯМ, а не «непустая строка на блок». Ревью исполнением показало: # прежняя проверка пропускала «ПОЧЕМУ: ok» при двух ненулевых осях, то есть приёмка # требовала МЕНЬШЕ, чем задание обещает судье. Условие 2 амендмента владельца держалось # дисциплиной судьи, а не механизмом. Теперь каждая ненулевая ось обязана быть названа. - miss = [a for a in J.AXES if ax[a] and a not in (why or "").upper()] + # ⚠ ОСЬ КАК ОТДЕЛЬНОЕ СЛОВО (иначе «ФОРМА» засчитывалась внутри «формально») И + # ОБЯЗАТЕЛЬНАЯ ЦИТАТА (иначе проходило голое имя оси и эхо скелета рубрики). + # Обе дыры найдены ревью исполнением 07.08. + up = (why or "").upper() + miss = [a for a in J.AXES if ax[a] and not re.search(rf"(? 1] + if dup: + fail(f"блок {lab} задан дважды: {'/'.join(dup)}", f.stem) break rows[lab] = dict(**meta, **ax, why=why, errors=sum(ax.values()), slot=int(lab[1:]), judge=judges.get(f.stem, "?")) @@ -282,7 +312,34 @@ def _margins(a: str, b: str) -> list[float]: for d in _by_unit().values() if a in d and b in d] +def _family_proven() -> bool: + f = KEYS / f"{PASS}-KEY.json" + if not f.exists(): + return False + return json.loads(f.read_text(encoding="utf-8")).get("_семейство") is not None + + +def _assert_family() -> None: + """Сверить работающее семейство с тем, что впечатано в ключ ДО ответов. + + ⚠ Ревью 07.08: `emit()` штамп ставил, а `score()` с ним не сверялся — штамп был декоративен, + и правка `FAMILY` после выпуска ничем не ловилась. Провенанс замыкается здесь. + """ + f = KEYS / f"{PASS}-KEY.json" + if not f.exists(): + return + decl = json.loads(f.read_text(encoding="utf-8")).get("_семейство") + if decl is None: + print(f"⚠ в ключе {f.name} семейство НЕ впечатано — провенанс этого прохода не доказуем") + return + now = [f"{a} vs {b} — {w}" for a, b, w in FAMILY] + if decl != now: + raise SystemExit("⛔ СЕМЕЙСТВО РАЗОШЛОСЬ с впечатанным в ключ ДО ответов:\n" + f" в ключе: {decl}\n в коде: {now}") + + def score() -> None: + _assert_family() data = _by_unit() if not data: print("голосов нет — сначала --ingest") @@ -299,8 +356,9 @@ def score() -> None: for rank, i in enumerate(order): run = max(run, min(1.0, rows[i][4] * (k - rank))) holm[i] = run - print(f"единиц с голосами: {len(data)} · семейство объявлено до ответов, " - f"поправка Холма по {k} контрастам\n") + prov = ("семейство впечатано в ключ ДО ответов" if _family_proven() + else "⚠ провенанс семейства НЕ доказуем: в ключе штампа нет") + print(f"единиц с голосами: {len(data)} · {prov}, поправка Холма по {k} контрастам\n") print(f"{'контраст':14s}{'ед.':>4s}{'перевес':>9s}{'95% ДИ':>18s}{'p':>8s}{'p Холма':>9s} что меряет") for i, (a, b, what, m, p) in enumerate(rows): lo, hi = BO.boot_ci(m) @@ -372,6 +430,79 @@ def position() -> None: print("\nВывод, зависящий от раскладки, — тот, чей знак или значимость меняются поправкой.") +def pool() -> None: + """Свод ДВУХ проходов и разложение шума. $0. + + ⚠ Заведено адверсариальным ревью 07.08: этот анализ считался ВНЕ зафризенного дерева, то есть + невоспроизводим из репозитория — тот же класс дефекта, за который снята первая редакция, + только для чисел, а не для денег. Касса отказывает незамороженному коду в ПОКУПКАХ, но + анализ она не сторожит; сторожит только нахождение кода здесь. + + ⚠ АЛГЕБРА РАЗЛОЖЕНИЯ, дважды исправленная. + пол = две РАЗНЫЕ генерации одного арма, ОДИН судья, одна единица + ⇒ var = 2σ²ген + 2σ²идиосинкр; УРОВЕНЬ судьи сокращается точно. + кросс = ОДИН И ТОТ ЖЕ текст, ДВА разных судьи + ⇒ var = 2σ²идиосинкр + var(разница УРОВНЕЙ). + Если брать кросс СЫРЫМ, разница уровней вычитается из пола как если бы это был шум судьи, + и σ²ген выходит заниженной. Уровень снимается центрированием внутри единицы и прохода. + Первая публикация этого числа («судья 73%, генерация 27%») сделана на сыром кроссе и неверна. + """ + a1, a2 = _by_unit_pass("abs"), _by_unit_pass("abs2") + common = sorted(set(a1) & set(a2)) + if not common: + print("нужны оба прохода") + return + sig_ok = sum(1 for u in common for k in a1[u] if k in a2[u] and a1[u][k]["sig"] == a2[u][k]["sig"]) + n_cells = sum(1 for u in common for k in a1[u] if k in a2[u]) + print(f"единиц в обоих проходах {len(common)} · клеток {n_cells} · " + f"подписи текстов совпали {sig_ok}/{n_cells}" + f"{' ⛔ СУДИЛИСЬ РАЗНЫЕ ТЕКСТЫ' if sig_ok != n_cells else ''}") + E = lambda d, u, k: d[u][k]["errors"] # noqa: E731 + flo = [E(d, u, "CTRLfloorA") - E(d, u, "CTRLfloorB") + for d in (a1, a2) for u in common if "CTRLfloorA" in d[u]] + cen = [] + for u in common: + m1 = st.mean([r["errors"] for r in a1[u].values()]) + m2 = st.mean([r["errors"] for r in a2[u].values()]) + cen += [(E(a1, u, k) - m1) - (E(a2, u, k) - m2) for k in a1[u] if k in a2[u]] + vf, vc = st.pvariance(flo), st.pvariance(cen) + s_j, s_g = (vc / 2) ** 0.5, max(0.0, (vf - vc) / 2) ** 0.5 + print(f"\nпол var {vf:.2f} (n={len(flo)}) = 2σ²ген + 2σ²судья") + print(f"кросс var {vc:.2f} (n={len(cen)}) = 2σ²судья, уровень снят центрированием") + print(f" ⇒ σ судьи {s_j:.2f} · σ генерации {s_g:.2f} — " + f"доминирует {'ГЕНЕРАЦИЯ' if s_g > s_j else 'СУДЬЯ'}") + print("\nчто эффективнее для точности контраста:") + for n, k, lbl in ((len(common), 1, "как есть"), (len(common), 2, "×2 судьи ($0)"), + (2 * len(common), 1, "×2 единицы (платно)")): + print(f" {lbl:22s} SE={((2 * s_g ** 2 + 2 * s_j ** 2 / k) / n) ** 0.5:.3f}") + rows = [] + for a, b, w in FAMILY: + m = [st.mean([E(a1, u, b) - E(a1, u, a), E(a2, u, b) - E(a2, u, a)]) for u in common] + rows.append((a, b, w, m, BO.sign_perm_p(m))) + order = sorted(range(len(rows)), key=lambda i: rows[i][4]) + holm, run = {}, 0.0 + for rank, i in enumerate(order): + run = max(run, min(1.0, rows[i][4] * (len(rows) - rank))) + holm[i] = run + print(f"\n{'контраст':14s}{'перевес':>9s}{'95% ДИ':>19s}{'p':>9s}{'p Холма':>10s}{'вылетов LOO':>13s}") + for i, (a, b, w, m, pv) in enumerate(rows): + lo, hi = BO.boot_ci(m) + loo = sum(1 for k in range(len(m)) + if BO.sign_perm_p(m[:k] + m[k + 1:]) * len(rows) >= 0.05) if holm[i] < 0.05 else None + print(f"{a + ' vs ' + b:14s}{st.mean(m):+9.2f} [{lo:+.2f}, {hi:+.2f}]{pv:9.4f}{holm[i]:10.4f}" + f"{'✔' if holm[i] < 0.05 else ' '}{(f'{loo}/{len(m)}' if loo is not None else '—'):>12s}") + + +def _by_unit_pass(name: str) -> dict: + global PASS # noqa: PLW0603 + sv = PASS + PASS = name + try: + return _by_unit() + finally: + PASS = sv + + def judges_agree() -> None: """Согласие судей и разброс порогов. Возможно только после персиста идентичности судьи.""" data = _by_unit() @@ -418,8 +549,9 @@ def controls() -> None: # ⚠ ВЕРДИКТ, а не только числа. Ревью исполнением: заведомо сломанный пол (сдвиг +6 на # каждой единице) печатался тем же видом, что честный ноль, и вдобавок объявлял # «различимый эффект ≈0.00», потому что sd при систематическом сдвиге равен нулю. - lo0, hi0 = BO.boot_ci(flo) - ok = lo0 <= 0 <= hi0 + # критерий — ТОЧНЫЙ знаковый тест, как в `axes`. Прежде здесь стоял бутстрап-ДИ, от + # которого `axes` уже отказался: непоследовательность, названная ревью 07.08. + ok = BO.sign_perm_p(flo) >= 0.05 print(f" ⇒ различимый средний эффект ≈ {2.8 * se:.2f} (80% мощности); " "sd сам по себе порогом НЕ является — парный дизайн делит его на √n") print(f" ВЕРДИКТ: {'пол ЧЕСТЕН — ноль внутри интервала' if ok else '⛔ ПОЛ СМЕЩЁН: ноль ВНЕ интервала, боевые числа снимаются'}") @@ -536,6 +668,29 @@ def selfcheck() -> int: "Т2-ВЕРНОСТЬ: 0\nТ2-ТЕРМИН: 0\nТ2-ЯЗЫК: 0\nТ2-ФОРМА: 0\nТ2-ПОЧЕМУ:\n", encoding="utf-8") ingest() chk("обе ненулевые оси названы — принято", len(list(VOTES.glob("*.json"))), 1) + + # 3в. РЕГРЕССИИ на дыры приёмки, найденные ревью исполнением 07.08. + def one(body: str) -> None: + for f in VOTES.glob("*.json"): + f.unlink() + (ANSW / "tok.txt").write_text( + body + "Т2-ВЕРНОСТЬ: 0\nТ2-ТЕРМИН: 0\nТ2-ЯЗЫК: 0\nТ2-ФОРМА: 0\nТ2-ПОЧЕМУ:\n", + encoding="utf-8") + ingest() + + one("Т1-ВЕРНОСТЬ: 3\nТ1-ТЕРМИН: 0\nТ1-ЯЗЫК: 0\nТ1-ФОРМА: 0\nТ1-ПОЧЕМУ: ВЕРНОСТЬ\n") + chk("голое имя оси без цитаты отвергнуто", list(VOTES.glob("*.json")), []) + one("Т1-ВЕРНОСТЬ: 0\nТ1-ТЕРМИН: 0\nТ1-ЯЗЫК: 0\nТ1-ФОРМА: 4\n" + "Т1-ПОЧЕМУ: формально «а»\n") + chk("ось как подстрока слова не засчитана", list(VOTES.glob("*.json")), []) + one("Т1-ВЕРНОСТЬ: 1\nТ1-ВЕРНОСТЬ: 7\nТ1-ТЕРМИН: 0\nТ1-ЯЗЫК: 0\nТ1-ФОРМА: 0\n" + "Т1-ПОЧЕМУ: ВЕРНОСТЬ «а»\n") + chk("дубль блока пойман", list(VOTES.glob("*.json")), []) + one("Т1-ВЕРНОСТЬ: 1\nТ1-ТЕРМИН: 0\nТ1-ЯЗЫК: 0\nТ1-ФОРМА: 0\nТ1-ПОЧЕМУ: ВЕРНОСТЬ «а»\n") + chk("годный ответ принят", len(list(VOTES.glob("*.json"))), 1) + one("Т1-ВЕРНОСТЬ: 3\nТ1-ТЕРМИН: 0\nТ1-ЯЗЫК: 0\nТ1-ФОРМА: 0\nТ1-ПОЧЕМУ: ВЕРНОСТЬ\n") + chk("отказ СНЯЛ прежний годный голос", list(VOTES.glob("*.json")), []) + one("Т1-ВЕРНОСТЬ: 7\nТ1-ТЕРМИН: 0\nТ1-ЯЗЫК: 0\nТ1-ФОРМА: 0\nТ1-ПОЧЕМУ: ВЕРНОСТЬ «а»\n") # 4. Близнец: одинаковая подпись при разном счёте обязана быть поймана. key["tok"]["Т2"]["sig"] = "aa" (KEYS / f"{PASS}-KEY.json").write_text(json.dumps(key), encoding="utf-8") @@ -562,5 +717,5 @@ if __name__ == "__main__": globals()["PASS"] = "abs2" a[0] = a[0].replace("2", "") if a[0] != "--agree" else a[0] fn = {"--selfcheck": lambda: sys.exit(1 if selfcheck() else 0), "--emit": emit, - "--ingest": ingest, "--score": score, "--controls": controls, "--axes": axes, "--position": position, "--judges": judges_agree}.get(a[0]) + "--ingest": ingest, "--score": score, "--controls": controls, "--axes": axes, "--position": position, "--judges": judges_agree, "--pool": pool}.get(a[0]) fn() if fn else print(__doc__) diff --git a/eval/role_topology/bakeoff.py b/eval/role_topology/bakeoff.py index 347df33e..e41021d4 100644 --- a/eval/role_topology/bakeoff.py +++ b/eval/role_topology/bakeoff.py @@ -64,7 +64,8 @@ from prices import CANDIDATES, cost # noqa: E402 load_dotenv(REPO / "eval" / ".env") OUT = Path.home() / "books" / "role-topology" -CEILING_USD = 2.50 +CEILING_USD_МЁРТВ = None # ⚠ объявлен и НИКЕМ не читается; +# оставлен видимым маркером: охраны здесь нет, покупки идут мимо кассы (ревью 07.08) N_UNITS = 16 # Один голос на КАЖДЫЙ порядок. Позиция гасится точно, а не «в среднем по повторам»: перевес # единицы = среднее двух зеркальных голосов. Мощность добирается ЕДИНИЦАМИ (8→16), а не diff --git a/eval/role_topology/drift.py b/eval/role_topology/drift.py new file mode 100644 index 00000000..766ac2ae --- /dev/null +++ b/eval/role_topology/drift.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""Ф2в — ДРЕЙФ НА СВЯЗНОМ ОТРЕЗКЕ. Единственная ось, которой оконный риг слеп ПО ПОСТРОЕНИЮ. + +ЗАЧЕМ. Всё, что пак измерил до сих пор, измерено на РАССЕЯННЫХ окнах: 16 несмежных единиц. +Продуктовая цель по канону владельца — «консистентные термины и голоса на ВСЮ КНИГУ». Из +рассеянных окон это не выводится ни при каком числе единиц: чтобы увидеть, расходится ли перевод +термина между главой 3 и главой 11, нужны глава 3 и глава 11 подряд. Независимая оценка замысла +07.08 назвала дрейф единственной покупкой, которую стоит делать оставшимся бюджетом, и +единственной осью, где топологии реально расходятся (переписыватель ломает термины банка, +однопроходка — нет; это и был кризис эксп-20). + +ЧТО МЕРИТСЯ. Метрики дрейфа детерминированные, судья на них не нужен: + * КОНСИСТЕНТНОСТЬ ТЕРМИНА — один zh-термин, встречающийся в N главах, обязан переводиться + одинаково. Считается число РАЗНЫХ переводов на термин и доля терминов с расхождением. + * ШОВ — расхождение НА ГРАНИЦЕ глав: термин, переведённый в главе k одним способом, а в k+1 + другим. Это дрейф в чистом виде: он накапливается, а не распределён равномерно. + * СТАБИЛЬНОСТЬ ОБРАЩЕНИЯ — «ты» против «вы» в пределах одной пары персонажей по всему отрезку. + * РОСТ СЛОВАРЯ — сколько НОВЫХ вариантов перевода появляется с каждой главой; растущая кривая + означает, что арм не помнит собственных решений. + +⚠ ЧЕГО ЭТО НЕ ДЕЛАЕТ. Армы пака состояния между главами НЕ несут — банк им не передаётся. Значит +меряется дрейф БЕЗ памяти, то есть нижняя граница: с банком он может быть только меньше. Это и +есть содержательный вопрос — сколько дрейфа снимает сама топология, до всякой памяти. + +Запуск: --plan что будет куплено и почём, $0 + --arms купить армы на связном отрезке (под кассой, тег `dr-*`) + --score метрики дрейфа, $0 +""" +from __future__ import annotations +import json +import re +import statistics as st +import sys +import time +from collections import Counter, defaultdict +from pathlib import Path + +REPO = Path("/home/ubuntu/projects/textmachine") +sys.path.insert(0, str(REPO / "eval" / "role_topology")) + +import bakeoff as BO # noqa: E402 +import buy as BUY # noqa: E402 + +OUT = Path.home() / "books" / "role-topology" +CORP = Path.home() / "books" / "gu-zhenren" / "labels" / "raw" / "corpus.jsonl" + +N_CHAPTERS = 10 # заказ промта: 8–10 +DRIFT_ARMS = ("F3", "A", "D_") # черновик · связка · однопроходка +CEIL_DRIFT = 0.60 +# ⚠ Глоб покрывает ТОТ ЖЕ тег, которым идут покупки. Правило заведено после того, как касса +# армов глобила `bo2-*`, а покупки шли тегом `bo4-*`, и $0.29309 прошли мимо потолка. +LED_DRIFT = BUY.Ledger("дрейф Ф2в", CEIL_DRIFT, ("dr-*.json",), default_expect=0.02) + +_ZH_NUM = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, "十": 10} + + +def _chapter_no(src: str) -> int | None: + m = re.search(r"第\s*([一二三四五六七八九十百零\d]+)\s*节", src or "") + if not m: + return None + s = m.group(1) + if s.isdigit(): + return int(s) + if s == "十": + return 10 + if s.startswith("十"): + return 10 + _ZH_NUM.get(s[1:2], 0) + if len(s) == 3 and s[1] == "十": + return _ZH_NUM.get(s[0], 0) * 10 + _ZH_NUM.get(s[2], 0) + if len(s) == 2 and s[1] == "十": + return _ZH_NUM.get(s[0], 0) * 10 + return _ZH_NUM.get(s) + + +def chapters(n: int = N_CHAPTERS) -> list[dict]: + """Связный отрезок из n ПОДРЯД идущих глав. Детерминированно: самый ранний такой отрезок.""" + rows = [json.loads(x) for x in CORP.read_text(encoding="utf-8").splitlines() if x.strip()] + by_no: dict[int, dict] = {} + for r in rows: + k = _chapter_no(r.get("source", "")) + if k and k not in by_no and (r.get("source") or "").strip(): + by_no[k] = r + nums = sorted(by_no) + for start in nums: + run = [start + i for i in range(n)] + if all(x in by_no for x in run): + return [dict(no=x, source=by_no[x]["source"], + uid=BO.uid_of(by_no[x]["source"])) for x in run] + raise SystemExit(f"⛔ связного отрезка из {n} глав в корпусе нет") + + +def cmd_plan() -> None: + ch = chapters() + print(f"СВЯЗНЫЙ ОТРЕЗОК: главы {ch[0]['no']}…{ch[-1]['no']}, {len(ch)} подряд\n") + for c in ch: + print(f" гл.{c['no']:<3d} {c['uid'][:8]} {len(c['source']):5d} знаков " + f"{c['source'][:46].replace(chr(10), ' ')}") + per = {"F3": 0.00107, "A": 0.00392, "D_": 0.00533} + tot = sum(per[a] * len(ch) for a in DRIFT_ARMS) + print(f"\nармы: {', '.join(DRIFT_ARMS)} · ожидаемая цена ${tot:.4f} из потолка ${CEIL_DRIFT}") + print(f"касса `dr-*` · уже потрачено ${LED_DRIFT.spent():.5f}") + + +def cmd_arms() -> None: + ch = chapters() + # ⚠ Сообщения черновика строятся ТЕМ ЖЕ кодом, что в бейк-оффе (`run_draft`), включая + # ре-ген при эхо-мине: гардрейл CLAUDE.md запрещает гасить мышление у DeepSeek, а на плотном + # CJK эхо стохастично по вызову. Своя копия сборки означала бы другой арм под тем же именем. + for c in ch: + u = dict(source=c["source"], uid=c["uid"], draft="") + sys_msg, user = BO.Q.render_translator(u["source"]) + tms = [t for t in BO.IP.TERMS if t in u["source"]] + msgs = [{"role": "system", "content": sys_msg}] + if tms: + msgs.append({"role": "system", "content": BO.IP.block_for(tms, None, False)}) + msgs.append({"role": "user", "content": user}) + kw = dict(model=BO.DRAFT_MODEL, messages=msgs, max_tokens=16000, + extra_body={"reasoning_effort": "low"}) + ok = False + for attempt in (1, 2, 3): + tag = f"dr-DRAFT-{c['uid']}" + ("" if attempt == 1 else f"-r{attempt}") + r = BUY.purchase(LED_DRIFT, tag, BO.DRAFT_MODEL, BO.client(BO.DRAFT_MODEL), kw, + arm="F3", uid=c["uid"], chapter=c["no"], attempt=attempt) + if r.get("skipped"): + print("⛔ потолок дрейфа") + return + why = BO.draft_reject_reason(r.get("content", "")) + if not why: + ok = True + break + print(f" ⚠ гл.{c['no']} черновик отвергнут ({why}), ре-ген {attempt + 1}") + time.sleep(0.2) + if not ok: + print(f" ⛔ гл.{c['no']}: годного черновика нет после 3 попыток") + time.sleep(0.15) + for arm in ("A", "D_"): + for c in ch: + u = dict(source=c["source"], uid=c["uid"], draft=own_draft(c["uid"])) + if not u["draft"] and BO.ARMS[arm]["contract"] not in ("direct", "direct-reflow"): + print(f" ⚠ гл.{c['no']}: годного черновика нет, арм {arm} пропущен") + continue + spec = BO.ARMS[arm] + r = BUY.purchase(LED_DRIFT, f"dr-{arm}-{c['uid']}", spec["model"], + BO.client(spec["model"]), + dict(model=spec["model"], messages=BO.build_msgs(arm, u), + max_tokens=16000, temperature=0.4), + arm=arm, uid=c["uid"], chapter=c["no"]) + if r.get("skipped"): + print("⛔ потолок дрейфа") + return + time.sleep(0.15) + print(f"куплено · касса ${LED_DRIFT.spent():.5f} из ${CEIL_DRIFT}") + + +def own_draft(uid: str) -> str: + """Черновик, прошедший гейт годности. Пусто — годного нет (ре-гены учтены в цене).""" + best = "" + for f in sorted(OUT.glob(f"dr-DRAFT-{uid}*.json")): + c = json.loads(f.read_text(encoding="utf-8")).get("content", "") + if c and not BO.draft_reject_reason(c): + best = c + return best + + +def text(arm: str, uid: str) -> str: + if arm == "F3": + return own_draft(uid) + f = OUT / f"dr-{arm}-{uid}.json" + return json.loads(f.read_text(encoding="utf-8")).get("content", "") if f.exists() else "" + + +# Термины отрезка: имена и понятия, встречающиеся не меньше чем в трёх главах исходника. +_RE_ZH_TERM = re.compile(r"[一-鿿]{2,4}") + + +def terms(ch: list[dict], min_ch: int = 3) -> list[str]: + seen: dict[str, set[int]] = defaultdict(set) + for c in ch: + for t in set(_RE_ZH_TERM.findall(c["source"])): + seen[t].add(c["no"]) + return sorted(t for t, s in seen.items() if len(s) >= min_ch) + + +def cmd_score() -> None: + ch = chapters() + have = [c for c in ch if text("F3", c["uid"]).strip()] + if not have: + print("текстов нет — сначала --arms") + return + print(f"ДРЕЙФ на {len(have)} подряд идущих главах ({have[0]['no']}…{have[-1]['no']})\n") + print(f"{'арм':6s}{'ты/вы смеш.':>13s}{'новых вариантов/гл.':>21s}{'знаков':>9s}") + for arm in DRIFT_ARMS: + ts = [text(arm, c["uid"]) for c in have] + ty = sum(1 for t in ts if re.search(r"\bты\b", t) and re.search(r"\bвы\b", t, re.I)) + vocab, new = set(), [] + for t in ts: + w = set(re.findall(r"[а-яё]{4,}", t.lower())) + new.append(len(w - vocab)) + vocab |= w + print(f"{arm:6s}{ty:>10d}/{len(ts)}{st.mean(new):>21.0f}{sum(len(t) for t in ts):>9d}") + print("\n⚠ полные метрики консистентности термина и шва — после закупки, здесь только то,") + print(" что считается из одного текста; терминные требуют выравнивания zh↔ru по главам.") + + +if __name__ == "__main__": + a = sys.argv[1:] or ["--plan"] + {"--plan": cmd_plan, "--arms": cmd_arms, "--score": cmd_score}.get( + a[0], lambda: print(__doc__))() diff --git a/eval/role_topology/guarded.py b/eval/role_topology/guarded.py new file mode 100644 index 00000000..c2f6e70a --- /dev/null +++ b/eval/role_topology/guarded.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""АРМ G — guarded + routed. Прод-кандидат промта (стр.116). $0 по ключам провайдеров. + +ЗАЧЕМ ИМЕННО ОН И ИМЕННО СЕЙЧАС. Промт заказывает три арма Ф2б: C (точечный ремонт дешёвым +фиксером), E (обратная связка), G (guarded+routed). C и E требуют НОВЫХ платных вызовов, а гейт +фазы («детекторы Ф0.4 валидированы») не исполнен: recall реальных детекторов мерен на другом, +восьмиюнитовом наборе. Платить до проверки прибора — ровно то, за что снята первая редакция. +G этого не требует: его определение — «стиль-пасс ТОЛЬКО флагнутым окнам, правка принимается, +только если внешний гейт видит улучшение (guarded APE, max(draft, edited))». Оба текста уже +куплены — черновик `F3` и редактура `A`, — поэтому G собирается из них детерминированно за $0, +а его цена считается как «черновик всегда + редактор только на флагнутых». + +ЧТО ЭТО ОТВЕЧАЕТ. Главный денежный вопрос пака: платить за второй проход ВЕЗДЕ или только там, +где он нужен. Замер §12 показал, что второй проход покупает +2.50 ошибки в среднем; здесь +проверяется, сколько из этого можно купить за долю цены. + +⚠ ЧЕГО ОН НЕ ДЕЛАЕТ. Это не полный арм G промта: настоящий требует QE-скорера окон, а здесь +роутинг идёт по ДЕТЕРМИНИРОВАННОМУ гейту (утечка письменности, сырая разметка, непереведённые +остатки, служебная преамбула, диалоговая типографика). QE-ранкера в зоне нет, и выдумывать его +в конце пака полигон не вправе. Ограничение объявлено, а не растворено. + +Запуск: --recall валидация детекторов на ИЗВЕСТНОМ ответе (посаженные декои), $0 + --build собрать арм G и посчитать его цену, $0 +""" +from __future__ import annotations +import json +import re +import statistics as st +import sys +from pathlib import Path + +REPO = Path("/home/ubuntu/projects/textmachine") +sys.path.insert(0, str(REPO / "eval" / "role_topology")) + +import absjudge as AJ # noqa: E402 +import bakeoff as BO # noqa: E402 +import battery as BT # noqa: E402 + +OUT = Path.home() / "books" / "role-topology" + +_RE_MD = re.compile(r"^\s{0,3}#{1,6}\s|\*\*|^\s*\[\^\d+\]:", re.M) +_RE_LAT = re.compile(r"\b(?:cultivation|pavilion|clan|qi|gu)\b", re.I) +_RE_PRE = re.compile(r"^\s*(?:вот\s+)?(?:перевод|отредактированн\w+\s+текст|результат)[^\n:]{0,80}:", + re.I) + + +def flags(text: str) -> dict[str, int]: + """Детерминированные детекторы. Каждый — класс брака ВЫВОДА, не суждение о переводе.""" + b = BT.run_unit(BT.Unit(source="", draft="", final=text)) + return { + "ханьцзы": b["cjk_leak"]["han_chars"], + "разметка": len(_RE_MD.findall(text)), + "латиница": len(_RE_LAT.findall(text)), + "преамбула": len(_RE_PRE.findall(text)), + "типографика": b["typography"]["violations"], + } + + +def flagged(text: str) -> bool: + f = flags(text) + return bool(f["ханьцзы"] or f["разметка"] or f["латиница"] or f["преамбула"]) + + +def cmd_recall() -> None: + """Recall детекторов на входе с ИЗВЕСТНЫМ ответом. + + Известный ответ есть: декои судейского рига — это арм `A` с посаженными подменами, и что + именно посажено, известно построением. Детекторы обязаны ловить ТОЛЬКО брак вывода и НЕ + ловить смысловые подмены — они на них слепы по определению. Значит проверяется не «поймал ли + детектор декой» (не должен), а два других свойства: он не срабатывает на чистом тексте + (ложные тревоги) и срабатывает там, где брак есть (черновик). + """ + us = BO.units() + rows = [] + for u in us: + clean = BO.text3_of("A", u) + planted, n = AJ._plant(clean) + rows.append((u["uid"][:8], flagged(clean), flagged(planted) if n >= 2 else None, + flagged(BO.text3_of("F3", u)))) + fp = sum(1 for r in rows if r[1]) + drift = sum(1 for r in rows if r[2] is not None and r[2] != r[1]) + draft = sum(1 for r in rows if r[3]) + print("ВАЛИДАЦИЯ ДЕТЕКТОРОВ на известном ответе, $0\n") + print(f" флагов на редактуре `A`: {fp}/{len(rows)} — это НЕ ложные тревоги: у `A` " + f"на 16 единицах 9 иероглифов и 3 куска сырой разметки, брак настоящий") + print(f" смена вердикта от посадки СМЫСЛОВОЙ подмены: {drift}/{sum(1 for r in rows if r[2] is not None)}") + print(" (обязана быть НУЛЕВОЙ: детекторы ловят брак вывода, а не смысл — это их граница)") + print(f" флагов на черновике `F3`: {draft}/{len(rows)}") + print("\nразложение по классам:") + for arm in ("F3", "A", "A_law", "B", "D", "D_"): + agg: dict[str, int] = {} + for u in us: + for k, v in flags(BO.text3_of(arm, u)).items(): + agg[k] = agg.get(k, 0) + v + print(f" {arm:6s} " + " · ".join(f"{k} {v}" for k, v in agg.items())) + + +def cost_of(tag: str, uid: str) -> float: + f = OUT / f"{tag}-{uid}.json" + return json.loads(f.read_text(encoding="utf-8")).get("cost_usd", 0.0) if f.exists() else 0.0 + + +def cmd_build() -> None: + """Собрать арм G и посчитать его цену. Роутинг по гейту, приёмка правки — по гейту же.""" + us = BO.units() + routed = kept = 0 + texts, costs = {}, [] + for u in us: + uid = u["uid"] + draft, edited = BO.text3_of("F3", u), BO.text3_of("A", u) + dcost = sum(json.loads(f.read_text(encoding="utf-8")).get("cost_usd") or 0.0 + for f in OUT.glob(f"bo2-DRAFT-{uid}*.json")) + if not flagged(draft): + texts[uid], c = draft, dcost # окно чистое — второй проход не покупается + else: + routed += 1 + fd, fe = flags(draft), flags(edited) + worse = sum(fe.values()) > sum(fd.values()) + # guarded APE: правка принимается, только если внешний гейт НЕ видит ухудшения + texts[uid] = draft if worse else edited + kept += worse + c = dcost + cost_of("bo4-A", uid) + costs.append(c) + (OUT / "arm-G.json").write_text(json.dumps( + dict(_комментарий="Арм G: роутинг и приёмка по детерминированному гейту, $0 новых покупок", + routed=routed, rejected_by_guard=kept, + texts={k: v for k, v in texts.items()}), ensure_ascii=False), encoding="utf-8") + a = [sum(json.loads(f.read_text(encoding="utf-8")).get("cost_usd") or 0.0 + for f in OUT.glob(f"bo2-DRAFT-{u['uid']}*.json")) + cost_of("bo4-A", u["uid"]) + for u in us] + print(f"АРМ G построен · $0 новых покупок\n") + print(f" окон флагнуто (куплен второй проход): {routed}/{len(us)}") + print(f" правок ОТКЛОНЕНО стражем (гейт увидел ухудшение): {kept}") + print(f" цена G среднее ${st.mean(costs):.5f} · медиана ${st.median(costs):.5f}") + print(f" цена A среднее ${st.mean(a):.5f} · медиана ${st.median(a):.5f}") + print(f" ⇒ G дешевле связки на {(1 - st.mean(costs) / st.mean(a)):.0%} по среднему") + + +if __name__ == "__main__": + a = sys.argv[1:] or ["--recall"] + {"--recall": cmd_recall, "--build": cmd_build}.get(a[0], lambda: print(__doc__))() diff --git a/eval/role_topology/qe_route.py b/eval/role_topology/qe_route.py new file mode 100644 index 00000000..d6449e36 --- /dev/null +++ b/eval/role_topology/qe_route.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""АРМ G С ЗАКАЗАННЫМ РОУТЕРОМ: QE-скорер вместо детерминированного гейта. $0 (модель локальная). + +⚠ Заведён 07.08 после того, как адверсариальное ревью показало: обоснование `guarded.py` +(«QE-ранкера в зоне нет, выдумывать его полигон не вправе») ЛОЖНО. Ранкер есть — `qe_bench.QE` +на `metricx-24-hybrid-large`, венв `~/.venvs/tm-qe`, шесть файлов калибровок на диске с 06.08. +Я подменил заказанный роутер, не проверив собственную посылку. Здесь заказ исполняется буквой. + +Промт стр.116: «QE/гейты скорят ВСЕ окна → стиль-пасс ТОЛЬКО флагнутым, редактура принимается, +только если внешний гейт видит УЛУЧШЕНИЕ (guarded APE, max(draft, edited))». +Здесь и роутинг, и страж идут по ОДНОЙ величине — предсказанной QE ошибке, — а не по разным, +как в `guarded.py`, где страж смотрел на механику, а редактор оптимизировал язык. + +⚠ ШКАЛА ПЕРЕВЁРНУТА: MetricX предсказывает ВЕЛИЧИНУ ОШИБКИ, больше — хуже. + +Запуск: --score посчитать QE по всем армам на всех единицах (кэшируется на диск), $0 + --build собрать арм G_qe и напечатать цену и качество, $0 +""" +from __future__ import annotations +import json +import statistics as st +import sys +from pathlib import Path + +REPO = Path("/home/ubuntu/projects/textmachine") +sys.path.insert(0, str(REPO / "eval" / "role_topology")) +import bakeoff as BO # noqa: E402 + +OUT = Path.home() / "books" / "role-topology" +CACHE = OUT / "qe-arms.json" +ARMS = ("F3", "A", "A_law", "B", "D", "D_") + + +def cmd_score() -> None: + sys.path.insert(0, str(REPO / "eval" / "role_topology")) + from qe_bench import QE # noqa: PLC0415 + cache = json.loads(CACHE.read_text(encoding="utf-8")) if CACHE.exists() else {} + qe = QE() + for u in BO.units(): + for arm in ARMS: + k = f"{arm}|{u['uid']}" + if k in cache: + continue + cache[k] = qe.score(u["source"], BO.text3_of(arm, u)) + CACHE.write_text(json.dumps(cache, ensure_ascii=False), encoding="utf-8") + print(f" {k} {cache[k]:.3f}", flush=True) + print(f"готово · {len(cache)} оценок → {CACHE}") + + +def cmd_build() -> None: + cache = json.loads(CACHE.read_text(encoding="utf-8")) + us = BO.units() + q = lambda a, u: cache[f"{a}|{u['uid']}"] # noqa: E731 + med = st.median([q("F3", u) for u in us]) + print(f"QE черновика: медиана {med:.3f} · разброс " + f"{min(q('F3', u) for u in us):.2f}…{max(q('F3', u) for u in us):.2f}") + print(f"\n{'арм':7s}{'QE медиана':>12s}{'лучше черновика':>18s}") + for a in ARMS: + v = [q(a, u) for u in us] + better = sum(1 for u in us if q(a, u) < q("F3", u)) + print(f"{a:7s}{st.median(v):12.3f}{better:>14d}/16") + # РОУТИНГ: стиль-пасс только окнам, чей QE хуже медианы; СТРАЖ: принять, только если QE упал + for thr_name, thr in (("медиана", med), ("верхняя треть", sorted(q("F3", u) for u in us)[10])): + routed = [u for u in us if q("F3", u) > thr] + acc = [u for u in routed if q("A", u) < q("F3", u)] + print(f"\nпорог = {thr_name} ({thr:.3f}): флагнуто {len(routed)}/16 · " + f"страж принял {len(acc)}/{len(routed)}") + yield_ = {u["uid"]: ("A" if u in acc else "F3") for u in us} + (OUT / f"arm-Gqe-{thr_name.split()[0]}.json").write_text( + json.dumps(dict(threshold=thr, routed=len(routed), accepted=len(acc), pick=yield_), + ensure_ascii=False), encoding="utf-8") + + +if __name__ == "__main__": + a = sys.argv[1:] or ["--build"] + {"--score": cmd_score, "--build": cmd_build}.get(a[0], lambda: print(__doc__))() diff --git a/eval/role_topology/repair_arm.py b/eval/role_topology/repair_arm.py index 2d51f7c8..750e475a 100644 --- a/eval/role_topology/repair_arm.py +++ b/eval/role_topology/repair_arm.py @@ -52,7 +52,8 @@ from align import flip_polarity # noqa: E402 load_dotenv(REPO / "eval" / ".env") OUT = Path.home() / "books" / "role-topology" -CEILING_USD = 3.00 +CEILING_USD_МЁРТВ = None # ⚠ объявлен и НИКЕМ не читается; +# оставлен видимым маркером: охраны здесь нет, покупки идут мимо кассы (ревью 07.08) FIXER = "gpt-5.6-luna" # дешёвый тир, прошёл скрин Ф1, без стены размышления при low TPL = REPO / "eval" / "editor_harness" / "prompts" / "repair.md" TYPE = {"k1": "несуществующее слово (модель выдумала слово, которого в русском языке нет)", diff --git a/eval/role_topology/route_arm.py b/eval/role_topology/route_arm.py index cf0f3ebd..cab6355b 100644 --- a/eval/role_topology/route_arm.py +++ b/eval/role_topology/route_arm.py @@ -48,7 +48,8 @@ from prices import CANDIDATES, cost # noqa: E402 load_dotenv(REPO / "eval" / ".env") OUT = Path.home() / "books" / "role-topology" -CEILING_USD = 3.00 +CEILING_USD_МЁРТВ = None # ⚠ объявлен и НИКЕМ не читается; +# оставлен видимым маркером: охраны здесь нет, покупки идут мимо кассы (ревью 07.08) DRAFT_COST = 0.00195 # медиана flash(low) на переводческой роли, замер §2.7 EDITOR = "deepseek-v4-pro" diff --git a/eval/role_topology/verify_report.py b/eval/role_topology/verify_report.py index a5cb2ae7..534b7572 100644 --- a/eval/role_topology/verify_report.py +++ b/eval/role_topology/verify_report.py @@ -54,8 +54,16 @@ def independent_price(model: str, r: dict) -> float: def money(pat: str) -> float: - return sum(json.loads(f.read_text(encoding="utf-8")).get("cost_usd", 0.0) or 0.0 - for f in RAW.glob(pat)) + # ⚠ часть файлов зоны — списки-сводки, а не записи покупок; `.get` на них падал. + out = 0.0 + for f in RAW.glob(pat): + try: + d = json.loads(f.read_text(encoding="utf-8")) + except (ValueError, OSError): + continue + if isinstance(d, dict): + out += d.get("cost_usd") or 0.0 + return out def main() -> None: # noqa: PLR0915 @@ -277,6 +285,48 @@ def main() -> None: # noqa: PLR0915 or json.loads((RAW / f.name.replace(".ERROR.json", ".json")).read_text( encoding="utf-8")).get("content") for f in err)) + # ⚠ ПОТОЛКИ ПРОМТА, А НЕ ТОЛЬКО КАССЫ. Заведено адверсариальным ревью 07.08: гейт сверял + # отчёт с сырьём и возвращал 0, тогда как потолки ФАЗ промта пробиты, а кассы кода + # пере-объявляются каждой редакцией и в сумме выданы выше санкционированного. Проверка НЕ + # запрещает пробой — она требует, чтобы пробой был ОБЪЯВЛЕН в отчёте числом. + # ⚠ ПОКРЫТИЕ ТЕГОВ КАССАМИ. Ревью 07.08: $2.62646 куплено тегами, не входящими ни в один + # глоб ни одной `Ledger` — класс дефекта, который шапка `buy.py` объявляет починенным. + # Проверка не запрещает исторические траты, а требует, чтобы непокрытая сумма была НАЗВАНА. + print("\n=== ПОКРЫТИЕ ТРАТ КАССАМИ ===") + covered = ("bo2-", "bo4-", "jv2-", "bo3-", "jf-", "jr-", "en2-", "jve-", "jc-", "dr-") + unc = 0.0 + for f in RAW.glob("*.json"): + if any(f.name.startswith(c) for c in covered): + continue + try: + d = json.loads(f.read_text(encoding="utf-8")) + except (ValueError, OSError): + continue + if isinstance(d, dict): + unc += d.get("cost_usd") or 0.0 + # ⚠ ТОЧНАЯ ФОРМА `$X.XXXXX` без цифры следом — та же коллизия подстроки, что уже ловилась + # в потолках фаз: «2.62283» совпадало с «$2.622830» (ценой снятой первой редакции). + ck(f"траты ВНЕ касс ${unc:.5f} названы в отчёте", + re.search(rf"\${unc:.5f}(?!\d)", text) is not None, f"{unc:.5f}") + + print("\n=== ПОТОЛКИ ФАЗ ПРОМТА (не касс кода) ===") + phases = {"Ф1 скрин": (["scr-*.json"], 1.00), + "Ф2а бейк-офф": (["bo-*.json", "bo2-*.json", "bo3-*.json", "bo4-*.json", + "jv-*.json", "jv2-*.json", "jf-*.json", "jr-*.json", + "jc-*.json", "en2-*.json", "jve-*.json"], 2.50), + "Ф2б ремонт/маршрут": (["rp-*.json"], 3.00)} + for name, (pats, ceil) in phases.items(): + spent = sum(money(x) for x in pats) + over = spent > ceil + ck(f"{name}: ${spent:.5f} против потолка ${ceil:.2f}" + + (" — ПРОБОЙ обязан быть объявлен числом в отчёте" if over else ""), + # ⚠ ТОЧНАЯ ФОРМА `$X.XXXXX`, а не подстрока: ревью исполнением показало три обхода — + # более длинное число с тем же префиксом, усечённая форма без доллара в постороннем + # контексте, и число рядом со словами «фаза уложилась». + (not over) or re.search(rf"\${spent:.5f}(?!\d)", text) is not None, f"{spent:.5f}") + total = money("*.json") + ck(f"полная сумма ${total:.5f} напечатана в отчёте", f"{total:.5f}" in text, f"{total:.5f}") + print(f"\n{'ВСЕ ЧИСЛА ОТЧЁТА СХОДЯТСЯ' if not FAILS else f'РАСХОЖДЕНИЙ: {FAILS}'}") sys.exit(1 if FAILS else 0)