netgescon-day0/scripts/ai/governance_sync.py

111 lines
4.0 KiB
Python
Executable File

#!/usr/bin/env python3
import glob
import os
import re
from datetime import datetime
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
TOPICS_INDEX = os.path.join(ROOT, 'docs', 'ai', 'TOPICS_INDEX.md')
CONTROL_DIR = os.path.join(ROOT, 'docs', 'ai', 'control-tower')
COMPACT_DIR = os.path.join(ROOT, 'docs', 'ai', 'restart', 'compact-summaries')
OUT_SNAPSHOT = os.path.join(CONTROL_DIR, 'DIRECTION_SNAPSHOT.md')
OUT_BUGS = os.path.join(CONTROL_DIR, 'OPEN_BUGS_MASTER.md')
def read(path):
try:
with open(path, 'r', encoding='utf-8', errors='replace') as f:
return f.read()
except FileNotFoundError:
return ''
def extract_topics():
text = read(TOPICS_INDEX)
links = re.findall(r'\- \[(.*?)\]\((topics/[^)]+)\)', text)
rows = []
for title, rel in links:
abs_path = os.path.join(ROOT, 'docs', 'ai', rel)
if os.path.exists(abs_path):
mtime = datetime.fromtimestamp(os.path.getmtime(abs_path)).strftime('%Y-%m-%d %H:%M:%S')
content = read(abs_path)
next_step = ''
m = re.findall(r'#### Prossimo passo\n\-\s*(.*)', content)
if m:
next_step = m[-1].strip()
if not next_step:
next_step = '(da definire)'
rows.append((title, f'docs/ai/{rel}', mtime, next_step))
return rows
def extract_open_bugs():
bugs = []
for path in sorted(glob.glob(os.path.join(COMPACT_DIR, '*.compact.md'))):
text = read(path)
in_table = False
for line in text.splitlines():
if line.strip().startswith('| Keyword | Codice | Pagina | File | Riga |'):
in_table = True
continue
if in_table and (not line.strip() or line.startswith('## ')):
in_table = False
continue
if in_table and line.startswith('|') and '---' not in line:
cols = [c.strip() for c in line.strip('|').split('|')]
if len(cols) >= 5:
keyword, code, page, filep, row = cols[:5]
if keyword and keyword != 'n/a':
bugs.append((keyword, code, page, filep, row, os.path.relpath(path, ROOT)))
# dedupe by key fields
uniq = {}
for b in bugs:
key = (b[0], b[1], b[2], b[3], b[4])
uniq[key] = b
return list(uniq.values())[:50]
def write_snapshot(topics, bugs):
os.makedirs(CONTROL_DIR, exist_ok=True)
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
with open(OUT_SNAPSHOT, 'w', encoding='utf-8') as out:
out.write('# Direction Snapshot\n\n')
out.write(f'- Updated: {now}\n')
out.write(f'- Topics tracked: {len(topics)}\n')
out.write(f'- Open bugs tracked: {len(bugs)}\n\n')
out.write('## Topics status\n')
out.write('| Topic | File | Last Update | Next Step |\n')
out.write('|---|---|---|---|\n')
for t in topics:
out.write(f'| {t[0]} | {t[1]} | {t[2]} | {t[3]} |\n')
out.write('\n## Manuali da tenere aggiornati\n')
out.write('- docs/ai/SESSION_HANDOFF.md\n')
out.write('- docs/ai/TOPICS_INDEX.md\n')
out.write('- docs/ai/control-tower/STREAMS_BOARD.md\n')
out.write('- docs/ai/control-tower/OPEN_BUGS_MASTER.md\n')
def write_open_bugs(bugs):
with open(OUT_BUGS, 'w', encoding='utf-8') as out:
out.write('# Open Bugs Master\n\n')
out.write('Solo info essenziali per ripartenza: keyword, codice, pagina, file, riga.\n\n')
out.write('| Keyword | Codice | Pagina | File | Riga | Fonte |\n')
out.write('|---|---|---|---|---|---|\n')
for b in bugs:
out.write(f'| {b[0]} | {b[1]} | {b[2]} | {b[3]} | {b[4]} | {b[5]} |\n')
def main():
topics = extract_topics()
bugs = extract_open_bugs()
write_snapshot(topics, bugs)
write_open_bugs(bugs)
print(f'Updated: {os.path.relpath(OUT_SNAPSHOT, ROOT)}')
print(f'Updated: {os.path.relpath(OUT_BUGS, ROOT)}')
if __name__ == '__main__':
main()