292 lines
16 KiB
Python
Executable file
292 lines
16 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""Блоки целедостижения (ТЗ tasks/dispatch/2026-10-16-goals-ledger.md; формат state/goals/README.md).
|
||
Принадлежность: МОЁ (менеджмент). Zero-LLM, без зависимостей.
|
||
Один блок = state/goals/<slug>.md с разделами ## ЦЕЛЬ (с «Замер:»), ## ЛЕСТНИЦА (строки «N. ✓|→|· текст»; ровно одна →),
|
||
## СЛЕДУЮЩИЙ ХОД (одна строка хода, «ТЗ: …», «Приёмка: …»), ## БЛОКЕРЫ («- Q<n>: …» или «нет»), ## ЖУРНАЛ («- дата: ход → исход»).
|
||
Поле шапки `tree: <узел>[ + <узел>]` — узел дерева state/goals/TREE.md (жирные имена), в ЦЕЛИ строка «Конечная цель: …» (16.10, goals-tree).
|
||
Темп — тактами: такт = пробуждение координатора (автоход или тычок оператора), журнал state/goals/ticks.jsonl пишет `goals.py tick`.
|
||
«В работе» — задача в state/dispatch/registry.jsonl с goal=<slug> без итогового вердикта (ok|stale) за 72 ч.
|
||
usage:
|
||
goals.py list блок · ступень k/n · следующий ход · в работе/блокер
|
||
goals.py next <slug> текущая ступень и следующий ход целиком
|
||
goals.py advance <slug> "<исход>" [--step-done] [--next "<ход>" --accept "<приёмка>" [--tz <путь>]]
|
||
строка в ЖУРНАЛ; --step-done двигает → на следующую ступень; без --next ход
|
||
помечается «(переписать)» — lint не пропустит, пока его не перепишут
|
||
goals.py lint блок без следующего хода, ход без приёмки, блокер без Q, лестница без одной →; rc = число нарушений
|
||
goals.py counts [--ticks 5|--hours 24] ходов за последние N тактов: к целям N / ops M / сироты K (без goal) по задачам реестра диспетчера
|
||
goals.py tick [--src auto|operator] такт в ticks.jsonl (первым шагом автохода; тычок оператора — operator)
|
||
goals.py slugs имена блоков (для --goal)
|
||
goals.py selftest"""
|
||
import datetime as dt, json, pathlib, re, sys
|
||
|
||
import sys as _sys; _sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
||
from kit import HOME # корень дома из kit.json (HOME_KIT_ROOT)
|
||
GD = HOME / 'state/goals'
|
||
REG = HOME / 'state/dispatch/registry.jsonl'
|
||
JOURNAL = GD / 'journal.jsonl'
|
||
TICKS = GD / 'ticks.jsonl'
|
||
TREE = GD / 'TREE.md'
|
||
NOT_BLOCKS = ('README.md', 'TREE.md', 'WEEK-43.md')
|
||
TICKS_N = 5
|
||
STEP_RE = re.compile(r'^\s*(\d+)\.\s*([✓→·])\s*(.+)$')
|
||
REWRITE = '(переписать)'
|
||
FINAL = ('ok', 'stale')
|
||
|
||
|
||
def sections(text):
|
||
out, cur = {}, None
|
||
for ln in text.splitlines():
|
||
if ln.startswith('## '): cur = ln[3:].strip().upper(); out[cur] = []
|
||
elif cur: out[cur].append(ln)
|
||
return out
|
||
|
||
|
||
def parse(path):
|
||
text = path.read_text(encoding='utf-8')
|
||
s = sections(text)
|
||
steps = [(int(m[1]), m[2], m[3].strip()) for m in (STEP_RE.match(l) for l in s.get('ЛЕСТНИЦА', [])) if m]
|
||
nxt = [l.strip() for l in s.get('СЛЕДУЮЩИЙ ХОД', []) if l.strip()]
|
||
move = next((l for l in nxt if not re.match(r'(ТЗ|Приёмка)\s*:', l)), '')
|
||
acc = next((l.split(':', 1)[1].strip() for l in nxt if l.startswith('Приёмка:')), '')
|
||
tz = next((l.split(':', 1)[1].strip() for l in nxt if l.startswith('ТЗ:')), '')
|
||
bl = [l.strip()[2:].strip() for l in s.get('БЛОКЕРЫ', []) if l.strip().startswith('- ')]
|
||
title = next((l[2:].strip() for l in text.splitlines() if l.startswith('# ')), path.stem)
|
||
tree = next((m[1].strip() for m in (re.match(r'tree:\s*(.+)$', l) for l in text.split('\n---', 1)[0].splitlines()) if m), '')
|
||
final = next((l.split(':', 1)[1].strip() for l in s.get('ЦЕЛЬ', []) if re.match(r'Конечная цель( \([^)]*\))?:', l)), '')
|
||
return {'tree': tree, 'final': final, 'slug': path.stem, 'title': title, 'text': text, 'goal': '\n'.join(s.get('ЦЕЛЬ', [])).strip(), 'steps': steps,
|
||
'move': move, 'accept': acc, 'tz': tz, 'blockers': bl}
|
||
|
||
|
||
def blocks():
|
||
return [parse(p) for p in sorted(GD.glob('*.md')) if p.name not in NOT_BLOCKS]
|
||
|
||
|
||
def slugs():
|
||
return [p.stem for p in sorted(GD.glob('*.md')) if p.name not in NOT_BLOCKS]
|
||
|
||
|
||
def tree_nodes():
|
||
"""Узлы дерева — жирные имена в TREE.md до строки «Карта блоков»."""
|
||
try: t = TREE.read_text(encoding='utf-8').split('Карта блоков', 1)[0]
|
||
except FileNotFoundError: return set()
|
||
return set(re.findall(r'\*\*([^*]+)\*\*', t))
|
||
|
||
|
||
def ticks():
|
||
try: return [json.loads(l)['at'] for l in TICKS.open(encoding='utf-8') if l.strip()]
|
||
except FileNotFoundError: return []
|
||
|
||
|
||
def tick(src='auto'):
|
||
at = dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
||
with TICKS.open('a', encoding='utf-8') as f: f.write(json.dumps({'at': at, 'src': src}, ensure_ascii=False) + '\n')
|
||
return at, len(ticks())
|
||
|
||
|
||
def since_ticks(n=TICKS_N):
|
||
"""Начало окна последних n тактов (время n-го с конца такта) или None, если тактов меньше n."""
|
||
t = ticks()
|
||
return t[-n][:16] if len(t) >= n else None
|
||
|
||
|
||
def reg_rows():
|
||
"""Записи реестра; ev=tgoal (привязка задним числом, 17.10) подставляет goal задаче без него."""
|
||
rows, tg = [], {}
|
||
try:
|
||
for l in REG.open(encoding='utf-8'):
|
||
try: r = json.loads(l)
|
||
except Exception: continue
|
||
if r.get('ev') == 'tgoal': tg[r.get('label')] = r.get('goal')
|
||
else: rows.append(r)
|
||
except FileNotFoundError: return
|
||
for r in rows:
|
||
if r.get('ev', 'task') == 'task' and not r.get('goal') and tg.get(r.get('label')): r = dict(r, goal=tg[r['label']])
|
||
yield r
|
||
|
||
|
||
def in_work(hours=72):
|
||
"""goal → [метки задач без итогового вердикта за hours]."""
|
||
since = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=hours)).strftime('%Y-%m-%dT%H:%M')
|
||
tasks, final = {}, set()
|
||
for r in reg_rows():
|
||
ev = r.get('ev', 'task')
|
||
if ev == 'task' and r.get('goal') and r.get('at', '') >= since: tasks[r['label']] = r['goal']
|
||
elif ev == 'accept' and r.get('verdict') in FINAL: final.add(r.get('label'))
|
||
elif ev == 'st' and r.get('st') == 'closed': final.add(r.get('id'))
|
||
out = {}
|
||
for lab, g in tasks.items():
|
||
if lab not in final: out.setdefault(g, []).append(lab)
|
||
return out
|
||
|
||
|
||
def orphan_signals(since):
|
||
"""Сигналы реестра с момента since без goal (ev=signal без поля и без последующего ev=goal): та же единица сироты, что в goals-sweep."""
|
||
sig, got = {}, set()
|
||
try:
|
||
for l in REG.open(encoding='utf-8'):
|
||
try: r = json.loads(l)
|
||
except Exception: continue
|
||
if r.get('ev') == 'signal' and r.get('at', '') >= since and not r.get('goal'): sig[r['id']] = r
|
||
elif r.get('ev') == 'goal': got.add(r.get('id'))
|
||
except FileNotFoundError: return 0
|
||
return len([i for i in sig if i not in got])
|
||
|
||
|
||
def counts(hours=24, since=None):
|
||
"""Три корзины: (к целям, ops, сироты). Сирота — задача реестра без goal ИЛИ сигнал без goal (одно определение с goals-sweep)."""
|
||
since = since or (dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=hours)).strftime('%Y-%m-%dT%H:%M')
|
||
g = o = s = 0
|
||
for r in reg_rows():
|
||
if r.get('ev', 'task') == 'task' and r.get('at', '') >= since:
|
||
if not r.get('goal'): s += 1
|
||
elif r['goal'] == 'ops': o += 1
|
||
else: g += 1
|
||
return g, o, s + orphan_signals(since)
|
||
|
||
|
||
def suggest(text, n=2):
|
||
"""Ближайшие блоки целей по пересечению слов (5+ букв) с текстом блока; для подсказки при отказе."""
|
||
w = set(re.findall(r'\w{5,}', text.lower()))
|
||
sc = sorted(((len(w & set(re.findall(r'\w{5,}', b['text'].lower()))), b['slug']) for b in blocks()), reverse=True)
|
||
return [sl for k, sl in sc[:n] if k]
|
||
|
||
|
||
def counts_line(hours=None, n=TICKS_N):
|
||
"""По тактам; часы — только явным --hours или пока тактов в журнале меньше n."""
|
||
since = None if hours else since_ticks(n)
|
||
if since: g, o, s = counts(since=since); head = f'ходов за последние {n} тактов'
|
||
else:
|
||
g, o, s = counts(hours or 24); head = f'ходов за {hours or 24} ч' + ('' if hours else f' (тактов {len(ticks())} < {n})')
|
||
return f'{head}: к целям {g} / ops {o} / сироты {s}' + (f' ({round(100 * g / (g + o + s))} % к целям)' if g + o + s else '')
|
||
|
||
|
||
def stage(b):
|
||
n = len(b['steps'])
|
||
cur = next((i for i, (_, m, _) in enumerate(b['steps']) if m == '→'), None)
|
||
k = cur + 1 if cur is not None else sum(1 for _, m, _ in b['steps'] if m == '✓')
|
||
return k, n, (b['steps'][cur][2] if cur is not None else '')
|
||
|
||
|
||
def lint_block(b):
|
||
bad = []
|
||
if 'Замер:' not in b['goal']: bad.append('ЦЕЛЬ без «Замер:»')
|
||
if not b['final']: bad.append('ЦЕЛЬ без «Конечная цель:»')
|
||
nodes = tree_nodes()
|
||
if not b['tree']: bad.append('нет поля tree:')
|
||
elif nodes and not all(x.strip() in nodes for x in b['tree'].split('+')): bad.append(f"tree: {b['tree']} — нет в TREE.md")
|
||
if not b['steps']: bad.append('лестница пуста')
|
||
elif sum(1 for _, m, _ in b['steps'] if m == '→') != 1: bad.append('в лестнице не одна → (текущая ступень)')
|
||
if not b['move'] or REWRITE in b['move']: bad.append('нет следующего хода')
|
||
if not b['accept']: bad.append('ход без «Приёмка:»')
|
||
for x in b['blockers']:
|
||
if x.lower() != 'нет' and not re.match(r'Q\d+\b', x): bad.append(f'блокер без Q: {x[:60]}')
|
||
return bad
|
||
|
||
|
||
def cmd_list():
|
||
w = in_work()
|
||
print('| узел дерева | блок | ступень | следующий ход | в работе / блокер |\n|---|---|---|---|---|')
|
||
for b in sorted(blocks(), key=lambda b: (b['tree'], b['slug'])):
|
||
k, n, _ = stage(b)
|
||
bl = [x.split(':', 1)[0] for x in b['blockers'] if x.lower() != 'нет']
|
||
state = ', '.join(w.get(b['slug'], [])) or '—'
|
||
if bl: state += ' · блокер ' + ', '.join(bl)
|
||
print(f"| {b['tree'] or '?'} | {b['slug']} | {k}/{n} | {b['move'][:110]} | {state} |")
|
||
print(counts_line())
|
||
|
||
|
||
def find(slug):
|
||
p = GD / f'{slug}.md'
|
||
if not p.exists(): sys.exit(f'нет блока {slug}; есть: {", ".join(slugs())}')
|
||
return p
|
||
|
||
|
||
def cmd_next(slug):
|
||
b = parse(find(slug)); k, n, cur = stage(b)
|
||
print(f"{b['title']}\nступень {k}/{n}: {cur}\nследующий ход: {b['move']}\nТЗ: {b['tz'] or '—'}\nПриёмка: {b['accept'] or '—'}")
|
||
if b['blockers']: print('блокеры: ' + '; '.join(b['blockers']))
|
||
w = in_work().get(slug)
|
||
if w: print('в работе: ' + ', '.join(w))
|
||
|
||
|
||
def cmd_advance(slug, a):
|
||
p = find(slug)
|
||
o = {}
|
||
for k in ('--next', '--accept', '--tz'):
|
||
if k in a: i = a.index(k); o[k] = a[i + 1]; a = a[:i] + a[i + 2:]
|
||
step_done = '--step-done' in a; a = [x for x in a if x != '--step-done']
|
||
outcome = ' '.join(a).strip()
|
||
if not outcome: sys.exit('advance: нужен «<исход>»')
|
||
text = p.read_text(encoding='utf-8'); b = parse(p)
|
||
lines = text.splitlines(); sec = None; moved = False; out = []
|
||
today = dt.datetime.now(dt.timezone(dt.timedelta(hours=3))).strftime('%Y-%m-%d')
|
||
for ln in lines:
|
||
if ln.startswith('## '): sec = ln[3:].strip().upper()
|
||
m = STEP_RE.match(ln) if sec == 'ЛЕСТНИЦА' else None
|
||
if step_done and m:
|
||
if m[2] == '→' and not moved: ln = ln.replace('→', '✓', 1); moved = 'next'
|
||
elif m[2] == '·' and moved == 'next': ln = ln.replace('·', '→', 1); moved = True
|
||
out.append(ln)
|
||
text = '\n'.join(out) + '\n'
|
||
move = o.get('--next') or f"{REWRITE} {b['move']}".strip()
|
||
new_next = [move, f"ТЗ: {o.get('--tz', b['tz'] if not o.get('--next') else '—') or '—'}", f"Приёмка: {o.get('--accept', '' if o.get('--next') else b['accept'])}"]
|
||
text = re.sub(r'(## СЛЕДУЮЩИЙ ХОД\n)(.*?)(\n## )', lambda m: m[1] + '\n'.join(new_next) + '\n' + m[3], text, count=1, flags=re.S)
|
||
text = re.sub(r'(## ЖУРНАЛ\n)', lambda m: m[1] + f'- {today}: {b["move"][:120]} → {outcome}\n', text, count=1)
|
||
p.write_text(text, encoding='utf-8')
|
||
with JOURNAL.open('a', encoding='utf-8') as f:
|
||
f.write(json.dumps({'at': dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'goal': slug, 'move': b['move'],
|
||
'outcome': outcome, 'step_done': step_done, 'next': o.get('--next', '')}, ensure_ascii=False) + '\n')
|
||
bad = lint_block(parse(p))
|
||
print(f'{slug}: журнал +1' + (', ступень сдвинута' if step_done else '') + ('' if not bad else ' · lint: ' + '; '.join(bad)))
|
||
|
||
|
||
def cmd_lint():
|
||
n = 0
|
||
for b in blocks():
|
||
for x in lint_block(b): print(f"{b['slug']}: {x}"); n += 1
|
||
print(f'lint: нарушений {n} в {len(slugs())} блоках')
|
||
return n
|
||
|
||
|
||
def selftest():
|
||
import tempfile
|
||
global GD, JOURNAL, TICKS, TREE
|
||
real = (GD, JOURNAL, TICKS, TREE)
|
||
with tempfile.TemporaryDirectory() as d:
|
||
GD = pathlib.Path(d); JOURNAL = GD / 'journal.jsonl'; TICKS = GD / 'ticks.jsonl'; TREE = GD / 'TREE.md'
|
||
TREE.write_text('**Уния** — x\nКарта блоков: **Чужое**\n', encoding='utf-8')
|
||
assert tree_nodes() == {'Уния'} and since_ticks(2) is None
|
||
tick(); tick('operator'); assert since_ticks(2) == ticks()[0][:16] and 'тактов' in counts_line(n=2)
|
||
(GD / 't.md').write_text('---\ntree: Уния\n---\n# Блок t\n## ЦЕЛЬ\nКонечная цель: z\nx. Замер: y\n## ЛЕСТНИЦА\n1. ✓ a\n2. → b\n3. · c\n## СЛЕДУЮЩИЙ ХОД\nсделать b\nТЗ: —\n'
|
||
'Приёмка: файл\n## БЛОКЕРЫ\n- нет\n## ЖУРНАЛ\n', encoding='utf-8')
|
||
b = parse(GD / 't.md'); assert stage(b)[:2] == (2, 3) and not lint_block(b), lint_block(b)
|
||
cmd_advance('t', ['b', 'сделано', '--step-done'])
|
||
b = parse(GD / 't.md'); assert stage(b)[:2] == (3, 3) and 'нет следующего хода' in lint_block(b), lint_block(b)
|
||
cmd_advance('t', ['c', 'начато', '--next', 'сделать c', '--accept', 'проба rc=0'])
|
||
b = parse(GD / 't.md'); assert not lint_block(b) and b['move'] == 'сделать c', (b['move'], lint_block(b))
|
||
(GD / 'u.md').write_text('# u\n## ЦЕЛЬ\nz\n## ЛЕСТНИЦА\n1. ✓ a\n## СЛЕДУЮЩИЙ ХОД\nход\n## БЛОКЕРЫ\n- ждём Руслана\n## ЖУРНАЛ\n', encoding='utf-8')
|
||
assert len(lint_block(parse(GD / 'u.md'))) == 6, lint_block(parse(GD / 'u.md'))
|
||
assert slugs() == ['t', 'u']
|
||
GD, JOURNAL, TICKS, TREE = real
|
||
print('SELFTEST OK: разбор, ступень k/n, advance со сдвигом и переписью хода, 6 классов lint, узлы TREE, такты')
|
||
return 0
|
||
|
||
|
||
def main(a):
|
||
c = a[0] if a else 'list'
|
||
if c == 'list': cmd_list()
|
||
elif c == 'next' and len(a) > 1: cmd_next(a[1])
|
||
elif c == 'advance' and len(a) > 2: cmd_advance(a[1], a[2:])
|
||
elif c == 'lint': return min(cmd_lint(), 100)
|
||
elif c == 'counts': print(counts_line(int(a[a.index('--hours') + 1]) if '--hours' in a else None,
|
||
int(a[a.index('--ticks') + 1]) if '--ticks' in a else TICKS_N))
|
||
elif c == 'tick':
|
||
at, n = tick(a[a.index('--src') + 1] if '--src' in a else 'auto'); print(f'такт {n}: {at}')
|
||
elif c == 'slugs': print(' '.join(slugs()))
|
||
elif c == 'selftest': return selftest()
|
||
else: print(__doc__); return 2
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main(sys.argv[1:]))
|