143 lines
5.8 KiB
Python
143 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Толерантный детерминированный apply anchored search/replace — Python-порт Go-стиля,
|
||
по спеке `docs/research/19-chunking-cohesion.md` §C2 (риг арма Q4b, `exp15:108`).
|
||
|
||
Спека дословно: якорь ищется с нормализацией пробелов/пунктуационных вариантов; неоднозначный
|
||
якорь (≥2 совпадений) → REJECT операции (не тихое применение первого); якорь не найден → REJECT;
|
||
reject-rate — телеметрия. Малформед → 1 регенерация → далее операция ОТБРАСЫВАЕТСЯ (черновик
|
||
остаётся) — blast-radius одной правки, а не чанка.
|
||
"""
|
||
from __future__ import annotations
|
||
import re
|
||
import unicodedata
|
||
from dataclasses import dataclass, field
|
||
|
||
OP_RE = re.compile(
|
||
r"<{5,9}\s*SEARCH\s*\n(?P<search>.*?)\n={5,9}\s*\n(?P<replace>.*?)\n>{5,9}\s*REPLACE",
|
||
re.S)
|
||
NO_CHANGE = "NO_CHANGE"
|
||
|
||
# Пунктуационные варианты, которые модель законно путает при копировании якоря.
|
||
PUNCT_CLASSES = [
|
||
("«»\"“”„‟", '"'),
|
||
("—–‒−-", "-"),
|
||
("…", "..."),
|
||
("’‘'`", "'"),
|
||
]
|
||
|
||
|
||
def fold(s: str) -> str:
|
||
"""Нормализация ДЛЯ ПОИСКА (не для вывода): NFKC · схлоп пробелов · классы пунктуации."""
|
||
s = unicodedata.normalize("NFKC", s)
|
||
for chars, to in PUNCT_CLASSES:
|
||
for ch in chars:
|
||
s = s.replace(ch, to)
|
||
return re.sub(r"\s+", " ", s).strip()
|
||
|
||
|
||
def _fold_map(text: str) -> tuple[str, list[int]]:
|
||
"""Свёрнутый текст + карта «индекс в свёрнутом → индекс в оригинале»."""
|
||
out, idx = [], []
|
||
prev_space = True # схлоп ведущих пробелов
|
||
for i, ch in enumerate(unicodedata.normalize("NFKC", text)):
|
||
if ch.isspace():
|
||
if prev_space:
|
||
continue
|
||
out.append(" ")
|
||
idx.append(i)
|
||
prev_space = True
|
||
continue
|
||
prev_space = False
|
||
for chars, to in PUNCT_CLASSES:
|
||
if ch in chars:
|
||
ch = to
|
||
break
|
||
if len(ch) == 1:
|
||
out.append(ch)
|
||
idx.append(i)
|
||
else: # … → ...
|
||
for c in ch:
|
||
out.append(c)
|
||
idx.append(i)
|
||
return "".join(out), idx
|
||
|
||
|
||
@dataclass
|
||
class ApplyResult:
|
||
text: str
|
||
ops_total: int = 0
|
||
applied: int = 0
|
||
rejected_notfound: int = 0
|
||
rejected_ambiguous: int = 0
|
||
rejected_overlap: int = 0
|
||
rejected_empty: int = 0
|
||
malformed: bool = False
|
||
no_change: bool = False
|
||
details: list[str] = field(default_factory=list)
|
||
|
||
@property
|
||
def rejected(self) -> int:
|
||
return (self.rejected_notfound + self.rejected_ambiguous
|
||
+ self.rejected_overlap + self.rejected_empty)
|
||
|
||
@property
|
||
def compliance(self) -> float:
|
||
"""Доля операций, доехавших до применения. Стоп-гейт Q4b: ≥0.90."""
|
||
return self.applied / self.ops_total if self.ops_total else 1.0
|
||
|
||
|
||
def parse_ops(reply: str) -> tuple[list[tuple[str, str]], bool]:
|
||
"""→ ([(search, replace)], malformed). NO_CHANGE = пустой список, не малформед."""
|
||
body = reply.strip()
|
||
ops = [(m.group("search"), m.group("replace")) for m in OP_RE.finditer(body)]
|
||
if ops:
|
||
# мусор ВНЕ операций (преамбулы/комментарии) — нарушение правила 6, но операции валидны:
|
||
# считаем малформедом только если разметка вообще не распозналась
|
||
return ops, False
|
||
if NO_CHANGE in body and len(body) < 200:
|
||
return [], False
|
||
return [], True
|
||
|
||
|
||
def apply_ops(draft: str, reply: str) -> ApplyResult:
|
||
ops, malformed = parse_ops(reply)
|
||
res = ApplyResult(text=draft, malformed=malformed, no_change=(not ops and not malformed))
|
||
if malformed or not ops:
|
||
return res
|
||
res.ops_total = len(ops)
|
||
folded, idx = _fold_map(draft)
|
||
spans: list[tuple[int, int, str]] = []
|
||
for search, replace in ops:
|
||
key = fold(search)
|
||
if not key:
|
||
res.rejected_empty += 1
|
||
res.details.append("REJECT пустой якорь")
|
||
continue
|
||
hits = [m.start() for m in re.finditer(re.escape(key), folded)]
|
||
if not hits:
|
||
res.rejected_notfound += 1
|
||
res.details.append(f"REJECT якорь не найден: {search[:60]!r}")
|
||
continue
|
||
if len(hits) > 1:
|
||
res.rejected_ambiguous += 1
|
||
res.details.append(f"REJECT якорь неоднозначен ({len(hits)}): {search[:60]!r}")
|
||
continue
|
||
s = idx[hits[0]]
|
||
e_fold = hits[0] + len(key) - 1
|
||
e = idx[e_fold] + 1
|
||
spans.append((s, e, replace))
|
||
# пересечения: операция, перекрывающая уже принятую, отбрасывается (спека: не пересекаются)
|
||
spans.sort()
|
||
accepted: list[tuple[int, int, str]] = []
|
||
for s, e, r in spans:
|
||
if accepted and s < accepted[-1][1]:
|
||
res.rejected_overlap += 1
|
||
res.details.append(f"REJECT пересечение спанов на {s}")
|
||
continue
|
||
accepted.append((s, e, r))
|
||
out = draft
|
||
for s, e, r in sorted(accepted, reverse=True): # с хвоста — офсеты не едут
|
||
out = out[:s] + r + out[e:]
|
||
res.text = out
|
||
res.applied = len(accepted)
|
||
return res
|