207 lines
9.9 KiB
Python
Executable file
207 lines
9.9 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""Приёмка сдач по блоку «Приёмка:» ТЗ — zero-LLM ядро (agent-home-kit; исходник — bin/accept.py дома Дистилла).
|
||
|
||
Пункт блока проверяется по виду: URL отдаёт код, файл есть и не пуст, коммит есть в git дома,
|
||
число в сдаче удовлетворяет порогу, письмо координатору не длиннее N строк. Пункт без
|
||
проверяемого условия уходит судье: в наборе судья — человек или ваша модель (вердикт judge).
|
||
|
||
accept.py gate <ТЗ> есть ли в ТЗ блок приёмки с проверяемым пунктом (гейт dispatch.py add)
|
||
accept.py check <метка> проверить сдачу витка <метка>: ev=accept ok|return|judge в реестр
|
||
accept.py selftest проба на временном доме
|
||
|
||
Сдача витка: строки witness-log --done этой метки (state/witness/turns.jsonl) + письма шины от неё.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import pathlib
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
import datetime as dt
|
||
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
||
from kit import HOME as H # noqa: E402
|
||
|
||
REG = H / 'state/dispatch/registry.jsonl'
|
||
TURNS = H / 'state/witness/turns.jsonl'
|
||
BUS = H / 'state/sessions/bus.jsonl'
|
||
T_URL, T_GIT = 15, 10
|
||
|
||
HEAD_RE = re.compile(r'^\s*(?:#+\s*)?(?:\*\*)?(Приёмка|Приемка|Критерии приёмки|Критерии приемки|Условия приёмки|Acceptance(?: criteria)?|Готово,? когда)(?:\*\*)?\s*:?\s*(?:\*\*)?', re.I | re.M)
|
||
URL_RE = re.compile(r'https?://[^\s)»"\'<>,;]+')
|
||
SHA_RE = re.compile(r'\b[0-9a-f]{7,40}\b')
|
||
PATH_RE = re.compile(r'(?<![\w/.-])((?:/(?:[\w.-]+/)*|(?:[\w.-]+/)+)[\w.-]+\.[A-Za-z0-9]{1,6})(?![\w/])')
|
||
NUM_RE = re.compile(r'(=|==|≤|<=|≥|>=|<|>)\s*(\d+(?:[.,]\d+)?)')
|
||
CMD_RE = re.compile(r'\b[\w.-]+\.(?:py|sh)\b')
|
||
IP_RE = re.compile(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b')
|
||
LINES_RE = re.compile(r'(?:≤|<=|не более)\s*(\d+)\s*строк', re.I)
|
||
VAL_RE = r'(?<![\w.:/-])(\d+(?:[.,]\d+)?)(?![.:/]\d|[\w-])'
|
||
|
||
|
||
def iso() -> str:
|
||
return dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
||
|
||
|
||
def tz_text(raw: str) -> str:
|
||
"""tz — путь к файлу ТЗ (от корня дома) или сам текст задачи."""
|
||
if raw and '\n' not in raw and len(raw) < 300:
|
||
f = H / raw
|
||
if f.is_file():
|
||
return f.read_text(encoding='utf-8', errors='replace')
|
||
return raw or ''
|
||
|
||
|
||
def block(text: str) -> tuple[str | None, str]:
|
||
"""Блок приёмки: строка-заголовок (синонимы) и всё до пустой строки/следующего заголовка."""
|
||
m = HEAD_RE.search(text or '')
|
||
if not m:
|
||
return None, ''
|
||
out = []
|
||
for i, l in enumerate(text[m.end():].splitlines()):
|
||
if i and (not l.strip() or re.match(r'^\s*#', l)):
|
||
break
|
||
out.append(l)
|
||
return m.group(1), '\n'.join(out).strip()
|
||
|
||
|
||
def items(blk: str) -> list[str]:
|
||
"""Пункты: строки-списки, иначе — по «;» и концу предложения."""
|
||
lines = [re.sub(r'^\s*(?:[-*•]|\d+[.)])\s*', '', l).strip() for l in blk.splitlines() if l.strip()]
|
||
if len(lines) == 1:
|
||
lines = [x.strip() for x in re.split(r';\s+|(?<=[.!])\s+(?=[А-ЯA-Z«])', lines[0]) if x.strip()]
|
||
return [l.rstrip('.;') for l in lines if len(l) > 3]
|
||
|
||
|
||
def kind(it: str) -> str:
|
||
if URL_RE.search(it): return 'url'
|
||
if re.search(r'коммит|\bcommit\b', it): return 'commit'
|
||
if LINES_RE.search(it) and re.search(r'сдач|письм', it, re.I): return 'lines'
|
||
if NUM_RE.search(it) and not IP_RE.search(it): return 'number'
|
||
if PATH_RE.search(it) and not CMD_RE.search(it.split()[0] if it.split() else ''): return 'file'
|
||
return 'judge'
|
||
|
||
|
||
def gate(raw: str) -> tuple[bool, str]:
|
||
head, blk = block(tz_text(raw))
|
||
if head is None:
|
||
return False, 'ТЗ без блока «Приёмка:» — раздачи нет. Допиши «Приёмка: <URL отдаёт 200 | файл <путь> есть | коммит | число с порогом>»'
|
||
ks = [kind(i) for i in items(blk)]
|
||
if not any(k != 'judge' for k in ks):
|
||
return False, f'блок «{head}» есть, но в нём нет проверяемого условия (URL, путь файла, коммит, число с порогом)'
|
||
return True, f'приёмка: {len(ks)} пунктов, проверяемых без LLM {sum(k != "judge" for k in ks)}'
|
||
|
||
|
||
def http_code(url: str):
|
||
try:
|
||
return urllib.request.urlopen(urllib.request.Request(url, headers={'User-Agent': 'home-kit-accept/1'}), timeout=T_URL).status
|
||
except urllib.error.HTTPError as e:
|
||
return e.code
|
||
except Exception as e:
|
||
return type(e).__name__
|
||
|
||
|
||
def _jsonl(p: pathlib.Path) -> list[dict]:
|
||
out = []
|
||
if p.exists():
|
||
for l in p.read_text(encoding='utf-8').splitlines():
|
||
try:
|
||
out.append(json.loads(l))
|
||
except Exception:
|
||
pass
|
||
return out
|
||
|
||
|
||
def evidence(label: str) -> dict:
|
||
done = [r.get('text', '') for r in _jsonl(TURNS) if r.get('session') == label and r.get('kind') == 'done']
|
||
letters = [r.get('text', '') for r in _jsonl(BUS) if r.get('kind') == 'msg' and r.get('session') == label]
|
||
return {'done': '\n'.join(done), 'letter': '\n---\n'.join(letters) if letters else 'письма на шине нет'}
|
||
|
||
|
||
def check(it: str, ev: dict):
|
||
"""→ (True/False | None — судье, доказательство)."""
|
||
k = kind(it)
|
||
blob = ev['done'] + '\n' + ev['letter']
|
||
if k == 'url':
|
||
url = URL_RE.search(it).group(0)
|
||
m = re.search(r'\b([1-5]\d\d)\b', it.replace(url, ''))
|
||
want = int(m.group(1)) if m else 200
|
||
c = http_code(url)
|
||
return c == want, f'{url} → {c} (ждали {want})'
|
||
if k == 'file':
|
||
bad, ps = [], PATH_RE.findall(it)
|
||
for p in ps[:4]:
|
||
f = pathlib.Path(p) if p.startswith('/') else H / p
|
||
sz = f.stat().st_size if f.exists() else -1
|
||
if sz <= 0:
|
||
bad.append(f'{p}: ' + ('пуст' if sz == 0 else 'нет'))
|
||
return (not bad), ('; '.join(bad) if bad else ', '.join(f'{p} есть' for p in ps[:4]))
|
||
if k == 'commit':
|
||
for s in SHA_RE.findall(it + ' ' + blob)[:6]:
|
||
r = subprocess.run(['git', '-C', str(H), 'cat-file', '-t', s], capture_output=True, text=True, timeout=T_GIT)
|
||
if r.stdout.strip() == 'commit':
|
||
return True, f'коммит {s} есть'
|
||
return False, 'хеша коммита в сдаче нет'
|
||
if k == 'lines':
|
||
n = int(LINES_RE.search(it).group(1))
|
||
if ev['letter'].startswith('письма на шине нет'):
|
||
return False, 'письма на шине нет'
|
||
ls = [sum(1 for l in m.splitlines() if l.strip()) for m in ev['letter'].split('\n---\n')]
|
||
return max(ls) <= n, f'письмо координатору: {max(ls)} непустых строк (условие ≤ {n})'
|
||
if k == 'number':
|
||
op, val = NUM_RE.search(it).groups()
|
||
val = float(val.replace(',', '.'))
|
||
head = re.sub(r'\S*[/\d]\S*|\S+-\S+', ' ', it[:NUM_RE.search(it).start()])
|
||
words = re.findall(r'[\wа-яё]{3,}', head.lower())[-2:]
|
||
cmp = lambda x: {'=': x == val, '==': x == val, '≤': x <= val, '<=': x <= val, '≥': x >= val, '>=': x >= val, '<': x < val, '>': x > val}[op]
|
||
low = blob.lower()
|
||
for w in reversed(words):
|
||
vals = [m.group(1) for m in re.finditer(re.escape(w[:5]) + r'[\wа-яё«»"\s:.—=≤≥<>→-]{0,40}?' + VAL_RE, low)]
|
||
if vals:
|
||
v = float(vals[0].replace(',', '.'))
|
||
return cmp(v), f'в сдаче «{w[:5]}…» = {vals[0]} (условие {op} {val:g})'
|
||
return None, 'числа в сдаче не нашёл'
|
||
return None, 'судье'
|
||
|
||
|
||
def cmd_check(label: str) -> int:
|
||
add = [r for r in _jsonl(REG) if r.get('label') == label and 'ev' not in r]
|
||
if not add:
|
||
print(f'{label}: в реестре нет'); return 2
|
||
head, blk = block(tz_text(add[-1].get('tz', '')))
|
||
if head is None:
|
||
print(f'{label}: в ТЗ нет блока приёмки'); return 2
|
||
ev = evidence(label)
|
||
res = []
|
||
for it in items(blk):
|
||
ok, why = check(it, ev)
|
||
res.append({'p': it, 'ok': ok, 'evidence': why, 'by': 'zero-llm' if ok is not None else 'judge'})
|
||
verdict = 'return' if any(r['ok'] is False for r in res) else ('judge' if any(r['ok'] is None for r in res) else 'ok')
|
||
with REG.open('a', encoding='utf-8') as f:
|
||
f.write(json.dumps({'ev': 'accept', 'label': label, 'verdict': verdict, 'items': res, 'at': iso()}, ensure_ascii=False) + '\n')
|
||
for r in res:
|
||
print(f" {'✓' if r['ok'] else ('✗' if r['ok'] is False else '?')} {r['p'][:70]} — {r['evidence']}")
|
||
print(f'{label}: {verdict}')
|
||
return 0 if verdict == 'ok' else 1
|
||
|
||
|
||
def selftest() -> int:
|
||
t = 'Сделать X.\nПриёмка: файл out/x.md есть; коммит; число строк ≤ 5.\n'
|
||
h, b = block(t)
|
||
assert h == 'Приёмка' and len(items(b)) == 3, (h, items(b))
|
||
assert [kind(i) for i in items(b)] == ['file', 'commit', 'number'], [kind(i) for i in items(b)]
|
||
assert gate(t)[0] and not gate('без блока')[0]
|
||
print('selftest ok'); return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
a = sys.argv[1:]
|
||
if a[:1] == ['gate'] and len(a) > 1:
|
||
ok, msg = gate(a[1]); print(msg); sys.exit(0 if ok else 1)
|
||
if a[:1] == ['check'] and len(a) > 1:
|
||
sys.exit(cmd_check(a[1]))
|
||
if a[:1] == ['selftest']:
|
||
sys.exit(selftest())
|
||
print(__doc__)
|