troupe-standup/wizard-work/ledger/verify_stackchain_catalog.py

#!/usr/bin/env python3
"""CHECK: re-derive the stackchain catalog from the raw source JSONs and diff
against the committed artifact (stackchain_catalog.json + stackchain_catalog.md).

Exit 0 only if the committed artifact is exactly what the sources produce, plus
the markdown summary table is sane. This is a real regeneration test, not a
tautology: it re-reads the source files and recomputes every row.
"""
import json
import re
import sys

SRC_RAW = '/root/stackchain_catalog_raw.json'
SRC_B2 = '/root/stackchain_catalog_batch2.json'
SRC_B3 = '/root/stackchain_catalog_batch3.json'
ACTUAL_JSON = '/root/troupe-standup/wizard-work/ledger/stackchain_catalog.json'
ACTUAL_MD = '/root/troupe-standup/wizard-work/ledger/stackchain_catalog.md'

TYPE_RE = re.compile(r'"type"\s*:\s*"(.*?)"')
TEXT_RE = re.compile(r'"text"\s*:\s*"([\s\S]*?)"')
ERROR_RE = re.compile(r'ERROR: (.*?)(?:, full=|$)')


def load(p):
    with open(p, encoding='utf-8') as f:
        return json.load(f)


def clean_text(v):
    return v.replace('\\n', '\n').replace('\\n', '\n')


def parse_entry_value(v):
    if not isinstance(v, str) or not v.strip():
        return None
    try:
        d = json.loads(v)
        if isinstance(d, dict):
            return d
    except Exception:
        pass
    m = TYPE_RE.search(v)
    text_m = TEXT_RE.search(v)
    return {
        'type': m.group(1) if m else '(unknown)',
        'text': clean_text(text_m.group(1)) if text_m else '',
    }


def reextract():
    raw = load(SRC_RAW)
    b2 = load(SRC_B2)
    b3 = load(SRC_B3)

    entries = []
    for idx, (fname, v) in enumerate(raw.items(), 1):
        d = parse_entry_value(v)
        if d and d.get('text'):
            entries.append({'no': None, 'filename': fname, 'source': 'raw',
                            'status': 'transcribed', 'type': d.get('type') or '(unknown)',
                            'text': d['text'], 'words': len(d['text'].split())})
        else:
            entries.append({'no': None, 'filename': fname, 'source': 'raw',
                            'status': 'empty', 'type': '(not transcribed)',
                            'text': '', 'words': 0})

    for idx, (fname, v) in enumerate(b2.items(), 1):
        d = parse_entry_value(v)
        if d and d.get('text'):
            entries.append({'no': None, 'filename': fname, 'source': 'batch2',
                            'status': 'transcribed', 'type': d.get('type') or '(unknown)',
                            'text': d['text'], 'words': len(d['text'].split())})
        else:
            entries.append({'no': None, 'filename': fname, 'source': 'batch2',
                            'status': 'error', 'type': '(retry failed)',
                            'text': '', 'words': 0})

    for idx, (fname, v) in enumerate(b3.items(), 1):
        d = parse_entry_value(v)
        if d and d.get('text'):
            entries.append({'no': None, 'filename': fname, 'source': 'batch3',
                            'status': 'transcribed', 'type': d.get('type') or '(unknown)',
                            'text': d['text'], 'words': len(d['text'].split())})
        else:
            entries.append({'no': None, 'filename': fname, 'source': 'batch3',
                            'status': 'error', 'type': '(retry failed)',
                            'text': '', 'words': 0})

    entries.sort(key=lambda e: e['filename'])
    for i, e in enumerate(entries, 1):
        e['no'] = i
    return entries


def load_actual():
    with open(ACTUAL_JSON, encoding='utf-8') as f:
        rows = json.load(f)
    md = open(ACTUAL_MD, encoding='utf-8').read()
    return rows, md


def md_summary_rows(md):
    rows = []
    for line in md.splitlines():
        if line.startswith('|') and '---' not in line and 'No' not in line:
            cells = [c.strip() for c in line.strip('|').split('|')]
            if cells:
                rows.append(cells)
    return rows


def main():
    errors = []
    expected = reextract()
    actual, md = load_actual()

    if not isinstance(actual, list):
        errors.append('artifact JSON is not a list')
        actual = []

    if len(actual) != len(expected):
        errors.append('row count: actual=%d expected=%d' % (len(actual), len(expected)))

    for i, (a, e) in enumerate(zip(actual, expected), 1):
        for key in ('no', 'filename', 'source', 'status', 'type', 'text', 'words'):
            if a.get(key) != e[key]:
                errors.append('row %d field %r: actual=%r expected=%r' % (i, key, a.get(key), e[key]))

    # markdown sanity: every entry must appear in the summary table with matching No/File/Type/Words.
    seen = set()
    for cells in md_summary_rows(md):
        if len(cells) == 6 and cells[1].startswith('`'):
            try:
                no = int(cells[0])
            except ValueError:
                continue
            seen.add(cells[1].strip('`'))
            if no < 1 or no > len(actual):
                errors.append('md table row No out of range: %s' % no)
                continue
            e = actual[no - 1]
            if cells[2] != e['source'] or cells[3] != e['type'] or cells[5] != str(e['words']):
                errors.append('md row %d: %r vs %r' % (no, cells[2:], (e['source'], e['type'], e['words'])))
    expected_files = {r['filename'] for r in expected}
    if seen != expected_files:
        errors.append('md table missing/extra files: %s' % sorted(seen ^ expected_files))

    if errors:
        print('CHECK FAILED:')
        for e in errors:
            print(' -', e)
        sys.exit(1)

    print('CHECK OK: %d rows re-derived from source JSONs and identical to the '
          'committed artifact.' % len(actual))
    sys.exit(0)


if __name__ == '__main__':
    main()