MeanField/reports/check_report_conversion.py

#!/usr/bin/env python3
"""Check the handoff's LaTeX-to-Markdown conversion without modifying inputs.

Pandoc's parsed tables preserve cell boundaries, which catches both lost values
and accidental columns caused by vertical bars in mathematical expressions.
Figure captions need a separate comparison because GFM stores them as raw HTML.
Only the explicitly requested JSON record is written; no report is rebuilt.
"""
from pathlib import Path
from collections import Counter
from html.parser import HTMLParser
import argparse
import copy
import datetime
import hashlib
import json
import re
import shutil
import subprocess
import sys


STEM = 'MeanField_research_handoff_2026-09-20'
MARKDOWN_FORMAT = 'gfm-tex_math_gfm+tex_math_dollars'


def nodes(value, skip_figures=False):
    if isinstance(value, dict):
        if skip_figures and value.get('t') == 'Figure':
            return
        yield value
        for child in value.values():
            yield from nodes(child, skip_figures)
    elif isinstance(value, list):
        for child in value:
            yield from nodes(child, skip_figures)


def semantic_tokens(value):
    """Keep values and math distinct; disregard layout-only whitespace."""
    if isinstance(value, dict):
        kind = value.get('t')
        if kind == 'Str':
            return [('text', value['c'])]
        if kind == 'Code':
            return [('code', value['c'][1])]
        if kind == 'Math':
            return [('math', value['c'][0]['t'],
                     re.sub(r'\s+', ' ', value['c'][1]).strip())]
        return [token for child in value.values()
                for token in semantic_tokens(child)]
    if isinstance(value, list):
        return [token for child in value for token in semantic_tokens(child)]
    return []


def row_values(row):
    # Retaining spans and cell boundaries rejects values shifted into a neighbor.
    return [(cell[2], cell[3], semantic_tokens(cell[4])) for cell in row[1]]


def table_values(table, omit_empty_head=False):
    content = table['c']
    head = [row_values(row) for row in content[3][1]]
    if omit_empty_head and all(not cell[2] for row in head for cell in row):
        head = []
    body = [row_values(row) for group in content[4]
            for row in group[2] + group[3]]
    foot = [row_values(row) for row in content[5][1]]
    return {'columns': len(content[2]), 'caption': semantic_tokens(content[1]),
            'head': head, 'body': body, 'foot': foot}


def expected_filtered_table(table):
    result = copy.deepcopy(table)
    head, bodies = result['c'][3][1], result['c'][4]
    removed = False
    if len(head) == 1 and bodies and bodies[0][3]:
        head_contents = [cell[4] for cell in head[0][1]]
        first_contents = [cell[4] for cell in bodies[0][3][0][1]]
        if head_contents == first_contents:
            bodies[0][3].pop(0)
            removed = True
    return result, removed


class HTMLInventory(HTMLParser):
    """Compare rendered captions, including their inline mathematical markup."""
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.ids = []
        self.figures = []
        self.figure = None
        self.in_caption = False

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if 'id' in attrs:
            self.ids.append(attrs['id'])
        if tag == 'figure':
            self.figure = {'id': attrs.get('id', ''), 'images': [],
                           'caption_count': 0, 'caption': []}
        if self.figure is None:
            return
        if tag == 'img':
            self.figure['images'].append(
                {'src': attrs.get('src', ''), 'alt': attrs.get('alt', '')})
        if tag == 'figcaption':
            self.in_caption = True
            self.figure['caption_count'] += 1
        elif self.in_caption:
            self.figure['caption'].append(
                ('start', tag, sorted(attrs.items())))

    def handle_startendtag(self, tag, attrs):
        self.handle_starttag(tag, attrs)
        if tag not in {'img', 'br', 'hr', 'input', 'meta', 'link'}:
            self.handle_endtag(tag)

    def handle_endtag(self, tag):
        if self.figure is None:
            return
        if tag == 'figcaption':
            self.in_caption = False
        elif tag == 'figure':
            self.figures.append(self.figure)
            self.figure = None
        elif self.in_caption:
            self.figure['caption'].append(('end', tag))

    def handle_data(self, data):
        if self.in_caption:
            self.figure['caption'].extend(('text', word) for word in data.split())


def math_counts(blocks):
    return Counter((node['c'][0]['t'], re.sub(r'\s+', ' ', node['c'][1]).strip())
                   for node in nodes(blocks, skip_figures=True)
                   if node.get('t') == 'Math')


def counter_entries(counter):
    return [{'value': value, 'count': count} for value, count in sorted(counter.items())]


def file_record(path):
    data = path.read_bytes()
    return {'path': str(path), 'sha256': hashlib.sha256(data).hexdigest(),
            'bytes': len(data)}


def check(args, record):
    pandoc = shutil.which('pandoc')
    if pandoc is None:
        raise RuntimeError('Pandoc is required but was not found on PATH.')
    record['pandoc_version'] = subprocess.check_output(
        [pandoc, '--version'], text=True).splitlines()[0]

    def convert(text, source_format, target_format, filtered=False):
        command = [pandoc, '--from=' + source_format, '--to=' + target_format]
        if filtered:
            command.append('--lua-filter=' + str(args.lua_filter))
        return subprocess.run(command, input=text, text=True, capture_output=True,
                              check=True, cwd=args.tex.parent).stdout

    inputs = [args.tex, args.markdown, args.lua_filter]
    record['inputs'] = [file_record(path) for path in inputs]
    tex_text, markdown_text = args.tex.read_text(), args.markdown.read_text()
    raw = json.loads(convert(tex_text, 'latex', 'json'))
    source = json.loads(convert(tex_text, 'latex', 'json', filtered=True))
    markdown = json.loads(convert(markdown_text, MARKDOWN_FORMAT, 'json'))
    errors = record['errors']

    # The generated TOC is not document content. Verify its membership before
    # excluding it from the independent body-math comparison.
    expected_toc = [node['c'][1][0] for node in nodes(source['blocks'])
                    if node.get('t') == 'Header' and node['c'][0] <= args.toc_depth]
    toc_blocks = []
    if markdown['blocks'] and markdown['blocks'][0].get('t') == 'BulletList':
        toc_blocks = [markdown['blocks'].pop(0)]
    toc_targets = [node['c'][2][0][1:] for node in nodes(toc_blocks)
                   if node.get('t') == 'Link' and node['c'][2][0].startswith('#')]
    if toc_targets != expected_toc:
        errors.append('Generated TOC entries differ from the source headings.')

    tables = lambda ast: [node for node in nodes(ast['blocks'])
                          if node.get('t') == 'Table']
    raw_tables, source_tables, markdown_tables = map(tables, (raw, source, markdown))
    record['table_counts'] = {'source': len(raw_tables), 'filtered': len(source_tables),
                              'markdown': len(markdown_tables)}
    if len({len(raw_tables), len(source_tables), len(markdown_tables)}) != 1:
        errors.append('Table counts differ between source, filtered source and Markdown.')
    record['tables'] = []
    for index, (original, filtered, actual) in enumerate(
            zip(raw_tables, source_tables, markdown_tables), start=1):
        expected, removed = expected_filtered_table(original)
        # GFM requires a header separator even when LaTeX has no header. Its
        # resulting empty row carries no data and is the sole allowed addition.
        expected_values = table_values(filtered)
        actual_values = table_values(actual, omit_empty_head=not expected_values['head'])
        unchanged = expected == filtered
        equivalent = expected_values == actual_values
        record['tables'].append({
            'index': index, 'removed_duplicate_header': removed,
            'filter_changes_exactly_expected': unchanged,
            'all_cell_values_and_math_preserved': equivalent,
            'body_rows': len(expected_values['body']),
            'numeric_tokens': sum(any(char.isdigit() for char in token[-1])
                                  for token in semantic_tokens(filtered)),
        })
        if not unchanged:
            errors.append(f'Table {index}: unexpected Lua-filter mutation.')
        if not equivalent:
            errors.append(f'Table {index}: cell values, boundaries, math or caption changed.')

    source_math, markdown_math = map(math_counts,
                                    (source['blocks'], markdown['blocks']))
    record['body_math'] = {
        'source_count': sum(source_math.values()),
        'markdown_count': sum(markdown_math.values()),
        'source_only': counter_entries(source_math - markdown_math),
        'markdown_only': counter_entries(markdown_math - source_math),
    }
    if source_math != markdown_math:
        errors.append('Body mathematics differs; figure-caption math is checked separately.')

    paths = re.findall(r'\\(?:nolinkurl|path)\{([^}]*)\}', tex_text)
    code_counts = lambda ast: Counter(node['c'][1] for node in nodes(ast['blocks'])
                                      if node.get('t') == 'Code')
    source_code, markdown_code = map(code_counts, (source, markdown))
    path_differences = [{'path': path, 'source_count': source_code[path],
                         'markdown_count': markdown_code[path]}
                        for path in sorted(set(paths))
                        if path not in markdown_text or source_code[path] != markdown_code[path]]
    record['paths'] = {'source_occurrences': len(paths), 'distinct': len(set(paths)),
                       'differences': path_differences}
    if path_differences:
        errors.append('Repository path text or occurrence counts changed.')

    inventories = []
    for ast in (source, markdown):
        inventory = HTMLInventory()
        inventory.feed(convert(json.dumps(ast), 'json', 'html5'))
        inventory.close()
        inventories.append(inventory)
    expected_html, actual_html = inventories
    record['figures'] = {
        'source_count': len(expected_html.figures),
        'markdown_count': len(actual_html.figures),
        'images_ids_and_captions_preserved': expected_html.figures == actual_html.figures,
        'source': expected_html.figures,
        'markdown': actual_html.figures,
    }
    if expected_html.figures != actual_html.figures:
        errors.append('Figure identifiers, image paths or rendered captions differ.')

    missing_targets = [target for target in toc_targets if target not in actual_html.ids]
    record['toc'] = {'expected_entries': len(expected_toc), 'entries': len(toc_targets),
                     'matches_source_headings': toc_targets == expected_toc,
                     'missing_targets': missing_targets}
    if missing_targets:
        errors.append('Some generated TOC links have no Markdown destination.')

    record['inputs_unchanged'] = record['inputs'] == [file_record(path) for path in inputs]
    if not record['inputs_unchanged']:
        errors.append('An input changed during the audit; rerun against a stable report.')


def main():
    reports = Path(__file__).resolve().parent
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--tex', type=Path, default=reports / (STEM + '.tex'))
    parser.add_argument('--markdown', type=Path, default=reports / (STEM + '.md'))
    parser.add_argument('--lua-filter', type=Path, default=reports / 'pandoc_tables.lua')
    parser.add_argument('--toc-depth', type=int, default=2)
    parser.add_argument('--output', type=Path, required=True,
                        help='Explicit destination for the JSON audit record.')
    args = parser.parse_args()
    for name in ['tex', 'markdown', 'lua_filter', 'output']:
        setattr(args, name, getattr(args, name).resolve())
    if args.output.suffix.lower() != '.json':
        parser.error('--output must name a JSON file; source documents are read-only.')
    if args.output in {args.tex, args.markdown, args.lua_filter, Path(__file__).resolve()}:
        parser.error('--output cannot overwrite an input or the checker.')
    record = {'checked_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
              'checker': file_record(Path(__file__).resolve()), 'errors': []}
    try:
        check(args, record)
    except (OSError, RuntimeError, ValueError, KeyError, IndexError,
            subprocess.CalledProcessError) as error:
        record['errors'].append(str(error))
        if isinstance(error, subprocess.CalledProcessError):
            record['pandoc_error'] = error.stderr
    record['passed'] = not record['errors']
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(record, indent=2) + '\n')
    status = 'PASS' if record['passed'] else 'FAIL'
    print(f'{status}: report conversion audit saved to {args.output}')
    for error in record['errors']:
        print(error, file=sys.stderr)
    return 0 if record['passed'] else 1


if __name__ == '__main__':
    sys.exit(main())