#!/usr/bin/env python3 """Ф1 — ПРОФИЛЬ-СКРИН ЖИЛЬЦОВ. Оси = гейты цены и дисциплины, НЕ прокси качества. Что делает и чего НЕ делает. Скрин отсеивает кандидатов, неработоспособных в роли: рвут формат, эхают исходник, отказываются, отдают не тот язык, дороги сверх объявленного порога роли или не управляют размышлением. Он НЕ ранжирует прозу — урок эксп-20 прямой: механический показатель (число абзацев) оказался не качеством, а слепой суд его опроверг. Спорные по СТИЛЮ доводятся до Ф2/Ф3 обоими. Пороги записаны ЗДЕСЬ, в коде, ДО первого замера, и продублированы в пре-реге §0.5. Скрин ОБЯЗАН убивать: у эксп-21 он убил 2 из 12 и клейм «Ф1 отобрала жильцов» был снят. Но пороги выводятся из требований пайплайна, а не подгоняются под желаемый состав. Запуск: --dry план и смета, $0 --draft черновик для редакторской роли на тех же 2 единицах (боевой flash) --run сам скрин --score свод по персистированному сырью, $0 """ from __future__ import annotations import json import os import re import statistics as st import sys import time from pathlib import Path REPO = Path("/home/ubuntu/projects/textmachine") ZONE = Path(__file__).resolve().parent sys.path.insert(0, str(ZONE)) from dotenv import load_dotenv # noqa: E402 from openai import OpenAI # noqa: E402 sys.path.insert(0, str(REPO / "eval" / "role_topology")) import bakeoff as BO # noqa: E402 гейт годности черновика import material as M # noqa: E402 import money as MONEY # noqa: E402 import prompts as PR # noqa: E402 from roster import CANDIDATES, billed_output, by_out_price, call_kwargs # noqa: E402 load_dotenv(REPO / "eval" / ".env") OUT = MONEY.OUT DRAFT_MODEL = "deepseek-v4-flash" ROLES = ("translator", "editor") # Пороги цены роли, объявленные ДО замера. Основание — экономика пака: черновик на книгу в # 1500 единиц при $0.02/ед стоит $30, что уже вне продуктовой рамки «низкий COGS»; редактор # оправдан втрое дороже, потому что покупает измеренное качество (эксп-21 §12). PRICE_CAP = {"translator": 0.02, "editor": 0.06} CYR_MIN = 0.5 # Г4: доля кириллицы среди букв ECHO_MAX = 0.15 # Г2: боевой порог disposition.go _RE_HAN = re.compile(r"[㐀-鿿]") # ⚠ ДЕТЕКТОР MARKDOWN ИСПРАВЛЕН ПО ВСКРЫТИЮ КЛЕТОК (правило D39.61: перед сравнительным выводом # открыть единицы). Первая версия (унаследована из скрина эксп-21) ловила ЛЮБОЕ `**` и потому # помечала как «протёкшую разметку» строку `***` — стандартный русский типографский разделитель # сцены, которым модель верно передала смену сцены в исходнике. Из-за этого ПРОВАЛ получали # `glm-5` (ратифицированный резервный редактор) и `gemini-3.1-flash-lite` — то есть дефект # детектора менял состав панели. # Здесь детектор приведён к БОЕВОМУ (`checks/sanitizer.go:320` + D33.2): ATX-заголовок = решётки, # пробел и БУКВА/ЦИФРА (декоративный разделитель решётками боевой гейт тоже не считает), плюс # настоящий жирный спан `**текст**`, который боевой санитайзер не ловит, но в прозе он мусор. # Порог «нарушений более чем в одной клетке» НЕ менялся — менялось только то, что считается # нарушением. _RE_MD = re.compile(r"^\s{0,3}#{1,6}\s+[^\W\d_]|\*\*(?=\S)(?:(?!\*\*).)+\*\*", re.M) _RE_PREAMBLE = re.compile(r"^\s*(вот|ниже|here is|here's)\b[^\n:]{0,80}:", re.I) _RE_REFUSAL = re.compile( r"не могу|не буду|отказ|прости(те)?, но|cannot (assist|help|comply)|i'm sorry", re.I) # Маркеры насилия — только для ВЫБОРА чувствительной единицы, не для оценки. VIO = ("杀", "血", "尸", "斩", "惨叫", "骨", "折磨", "残忍", "痛苦") def units() -> list[dict]: """Две боевые единицы: «обычная» (медиана длины) и ЧУВСТВИТЕЛЬНАЯ (максимум насилия). Вторая нужна гейту Г3: вебновелла гарантированно упирается в насилие, и тихий отказ на главе N рвёт конвейер. Выбор детерминирован и объявлен до замера. """ us = M.units_zh() by_len = sorted(us, key=lambda u: len(u["source"])) normal = by_len[len(by_len) // 2] dens = sorted(us, key=lambda u: (-sum(u["source"].count(v) for v in VIO), u["uid"])) sens = next(u for u in dens if u["uid"] != normal["uid"]) normal, sens = dict(normal, kind="обычная"), dict(sens, kind="чувствительная") return [normal, sens] def client(model: str) -> OpenAI: base, env, *_ = CANDIDATES[model] return OpenAI(api_key=os.environ[env], base_url=base, timeout=900) def draft_of(uid: str) -> str: """ГОДНЫЙ черновик единицы: последняя попытка, прошедшая гейт. Пусто — годного нет.""" best = "" for f in sorted(OUT.glob(f"tp1-scr-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 cmd_draft() -> None: """Черновик под редакторскую роль скрина. Один и тот же для ВСЕХ кандидатов — парный дизайн. ⚠ РЕ-ГЕН ПРИ ЭХЕ добавлен ПОСЛЕ первой же покупки: единица `474f822806` вернулась ПОЛНЫМ китайским исходником (`finish=stop`, ханьцзы 1785, кириллица 0.00) — эхо-мина на плотном CJK, описанная в quirks 00. Гейт годности и N ре-генов — боевое лечение (ре-ген на flash в 7.6 раза дешевле эскалации на pro), и цена ре-генов ЛОЖИТСЯ В ЛЕДЖЕР, а не прячется. Функция гейта берётся из `bakeoff` — та самая, что проверена `checks21.py` на историческом артефакте; переписывать её здесь значило бы завести вторую, расходящуюся. """ led = MONEY.Guarded("Ф1", default_expect=0.004) for u in units(): blk = PR.glossary_block(PR.terms_in(u["source"])) msgs = PR.translator_msgs(u["source"], blk) for attempt in (1, 2, 3): tag = f"tp1-scr-draft-{u['uid']}" + ("" if attempt == 1 else f"-r{attempt}") rec = MONEY.purchase(led, tag, DRAFT_MODEL, client(DRAFT_MODEL), call_kwargs(DRAFT_MODEL, msgs), role="draft-for-screen", uid=u["uid"], attempt=attempt) if rec.get("skipped"): print("⛔ потолок Ф1 — стоп") return bad = BO.draft_reject_reason(rec.get("content", "")) print(f" черновик {u['uid'][:6]} ({u['kind']}) попытка {attempt} " f"finish={rec.get('finish')} {len(rec.get('content', '')):5d} зн " f"${rec.get('cost_usd', 0):.5f}" + (f" ⚠ {bad} — ре-ген" if bad else "")) if not bad: break print(f"потрачено Ф1 ${led.spent():.6f}") def msgs_for(role: str, u: dict) -> list[dict]: src = u["source"] if role == "translator": return PR.translator_msgs(src, PR.glossary_block(PR.terms_in(src))) return PR.editor_msgs(src, draft_of(u["uid"]), PR.law_block(PR.terms_in(src))) def gates(rec: dict, source: str) -> dict: t = rec.get("content", "") or "" letters = [c for c in t if c.isalpha()] han = len(_RE_HAN.findall(t)) cyr = sum(1 for c in letters if "а" <= c.lower() <= "я" or c.lower() == "ё") return dict( empty=not t.strip(), truncated=rec.get("finish") == "length", md=bool(_RE_MD.search(t)), preamble=bool(_RE_PREAMBLE.search(t)), han_chars=han, han_share=round(han / max(1, len(letters)), 5), cyr_share=round(cyr / max(1, len(letters)), 5), refusal=bool(_RE_REFUSAL.search(t[:400])) or bool(rec.get("no_message")), no_message=bool(rec.get("no_message")), reasoning_share=round(rec.get("reasoning_tokens", 0) / max(1, billed_output( rec["model"], rec.get("completion_tokens", 0), rec.get("reasoning_tokens", 0), rec.get("total_tokens", 0), rec.get("prompt_tokens", 0))), 3), cost=rec.get("cost_usd", 0.0), latency=rec.get("latency_s", 0.0), role=rec.get("role", ""), ) def verdict(rows: list[dict]) -> tuple[str, list[str]]: """Пороги применяются как объявлены. Пустой список — НЕ провал, а отсутствие замера.""" if not rows: return "НЕ ЗАМЕРЕН", ["потолок кончился до этой модели"] why = [] fmt_bad = sum(1 for r in rows if r["md"] or r["preamble"] or r["empty"] or r["truncated"]) if fmt_bad > 1: why.append(f"Г1 формат: нарушений в {fmt_bad} клетках") if any(r["han_share"] > ECHO_MAX for r in rows): why.append("Г2 эхо: клетка выше боевого порога 0.15") elif all(r["han_chars"] for r in rows): why.append("Г2 утечка: ханьцзы во ВСЕХ клетках") if any(r["refusal"] for r in rows): why.append("Г3 отказ на чувствительном фрагменте") if any(r["cyr_share"] < CYR_MIN and not r["empty"] for r in rows): why.append("Г4 выход не на целевом языке") for role in ROLES: rr = [r["cost"] for r in rows if r["role"] == role] if rr and st.median(rr) > PRICE_CAP[role]: why.append(f"Г5 цена {role}: ${st.median(rr):.4f} > ${PRICE_CAP[role]}") if any(r["truncated"] and r["empty"] for r in rows): why.append("Г6 размышление съело бюджет, выход пуст") return ("ПРОВАЛ" if why else "прошёл"), why def cmd_run(dry: bool = False) -> None: us = units() models = by_out_price() print(f"кандидатов {len(models)} · единиц {len(us)} · ролей {len(ROLES)} · " f"вызовов {len(models) * len(us) * len(ROLES)} · потолок Ф1 ${MONEY.CEILINGS['Ф1']}") for u in us: print(f" единица {u['uid']} ({u['kind']}, гл.{u['chapter']}) — {len(u['source'])} зн, " f"черновик {len(draft_of(u['uid']))} зн") print("порядок — по ВОЗРАСТАНИЮ цены выхода: кончится потолок — непокрытыми останутся дорогие,\n" "и это будет объявлено числом, а не скрыто") if dry: for m in models: print(f" {m}") return if not all(draft_of(u["uid"]) for u in us): print("⛔ нет черновика для редакторской роли — сначала --draft") return led = MONEY.Guarded("Ф1", default_expect=0.02) for m in models: for role in ROLES: for u in us: rec = MONEY.purchase(led, f"tp1-scr-{m}-{role}-{u['uid'][:6]}", m, client(m), call_kwargs(m, msgs_for(role, u)), role=role, uid=u["uid"], kind=u["kind"]) if rec.get("skipped"): print(f" ⛔ потолок — {m} и дальше НЕ ЗАМЕРЕНЫ") cmd_score() return g = gates(rec, u["source"]) print(f" {m:24s} {role:11s} {u['kind']:14s} finish={str(rec.get('finish')):9s}" f" {len(rec.get('content', '')):5d} зн ханьцзы={g['han_chars']:3d}" f" кир={g['cyr_share']:.2f} ${g['cost']:.5f} {g['latency']:.0f}с") time.sleep(0.2) cmd_score() def cmd_score() -> None: us = {u["uid"]: u for u in units()} per: dict[str, list[dict]] = {} for f in sorted(OUT.glob("tp1-scr-*.json")): r = json.loads(f.read_text(encoding="utf-8")) if r.get("skipped") or r.get("role") == "draft-for-screen" or r.get("finish") == "error": continue u = us.get(r.get("uid")) if u: per.setdefault(r["model"], []).append(gates(r, u["source"])) print(f"\n{'модель':26s}{'вердикт':>11s}{'$/ед transl':>13s}{'$/ед edit':>11s}" f"{'эхо':>6s}{'размышл':>9s} причины") print("-" * 116) res = {} for m in by_out_price(): rows = per.get(m, []) v, why = verdict(rows) pt = [r["cost"] for r in rows if r["role"] == "translator"] pe = [r["cost"] for r in rows if r["role"] == "editor"] han = sum(r["han_chars"] for r in rows) rs = st.median([r["reasoning_share"] for r in rows]) if rows else 0 print(f"{m:26s}{v:>11s}{(st.median(pt) if pt else 0):13.5f}" f"{(st.median(pe) if pe else 0):11.5f}{han:6d}{rs:9.1%} {'; '.join(why)}") res[m] = dict(verdict=v, why=why, n=len(rows), price_translator=st.median(pt) if pt else None, price_editor=st.median(pe) if pe else None, han=han, reasoning_share=rs) (OUT / "screen.json").write_text(json.dumps(res, ensure_ascii=False, indent=1), encoding="utf-8") led = MONEY.Guarded("Ф1") print(f"\nпотрачено Ф1 ${led.spent():.6f} из ${led.ceiling:.2f} · " f"пак ${MONEY.Pack().spent():.6f} из ${MONEY.PACK_CEILING:.2f}") print("⚠ числа печатаются по КАЖДОМУ кандидату — и вошедшему, и нет (симметрия промта)") def cmd_probe_unlisted() -> None: """$0-проба существования слагов, которых нет в /models (урок «нет в списке ≠ не работает»).""" import requests # noqa: PLC0415 key = os.environ.get("ZAI_API_KEY", "") for slug in ("glm-4.7-flash", "glm-4.7-flashx"): r = requests.get(f"https://api.z.ai/api/paas/v4/models/{slug}", headers={"Authorization": f"Bearer {key}"}, timeout=30) print(f" {slug:18s} HTTP {r.status_code} {r.text[:120]}") if __name__ == "__main__": a = sys.argv[1:] or ["--dry"] fn = {"--dry": lambda: cmd_run(dry=True), "--draft": cmd_draft, "--run": cmd_run, "--score": cmd_score, "--probe-unlisted": cmd_probe_unlisted}.get(a[0]) fn() if fn else print(__doc__)