"""Independent existing-evidence audit. Never connects to a database or rewrites raw."""
import copy
import hashlib
import json
from pathlib import Path
import re

ROOT = Path(__file__).resolve().parent.parents[2]
PLAN = ROOT / 'work/2026-09-28-smart-canteen-dm8-poc/prepay-onupdate-schema-20261011/full27-shadow-20261011/formal-2ddl-root-plan/plan.json'
RAW = Path('/Users/wiscourper/Documents/workspace/local/dm8-prepay-formal-root-evidence-20261011.6Sibqq/events.jsonl')
BASE = Path('/Users/wiscourper/Documents/workspace/local/dm8-prepay-fresh-evidence-20261011.812KmG/events.jsonl')
PACKAGE = Path('/Users/wiscourper/Documents/workspace/local/dm8-prepay-formal-root-evidence-20261011.6Sibqq/baseline.json')
checks = 0

def strict(a, b):
    if type(a) is not type(b):
        return False
    if isinstance(a, dict):
        return set(a) == set(b) and all(strict(a[k], b[k]) for k in a)
    if isinstance(a, list):
        return len(a) == len(b) and all(strict(x, y) for x, y in zip(a, b))
    return a == b

def eq(a, b, label):
    global checks
    assert strict(a, b), label
    checks += 1

def load_lines(path):
    return [json.loads(line) for line in path.read_text().splitlines()]

def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()

eq(digest(PLAN), '9a3df53e798ad01720a47bb1a1695ed0fd82df986c6228f9606826327123a86c', 'Plan SHA')
eq(digest(RAW), '7235505367496cf281c30ea7c2fb570b850a31ee747d2d8c0dcd16b645337aca', 'Complete Root raw SHA')
eq(digest(BASE), '477dfdfa69b3fa380629661a6e7521d8e9dff3a59003ec684270b2deb0f9efca', 'Original raw SHA')
eq(digest(PACKAGE), '0f8f8dcca7142dd9cd6fad874f8fcbbace61c55fd8232e5d421e7e093d4d6476', 'Approved private package SHA')
plan = json.loads(PLAN.read_text())
events = load_lines(RAW)
old = {e['label']: e for e in load_lines(BASE) if e['event'] == 'read-original-result'}
package = json.loads(PACKAGE.read_text())
sent = [e['statement'] for e in events if e['event'] == 'append-before']
results = [e for e in events if e['event'] in ['original-read-result', 'original-DDL-result']]
eq(sent, plan['steps'], 'Exactly all54 original statements, binding types and ordinal order')
eq([e['statement'] for e in results], sent, 'One original result per sent statement')
eq([e['errorInfo'][0] for e in results], ['00000'] * 54, 'Every native SQLSTATE success')
eq(events[-1]['event'], 'complete', 'Original complete journal')
eq(events[-1]['state'], {'cursor': 54, 'read': 52, 'DDL': 2, 'pre_proven': True, 'alter_proven': True, 'final_proven': True}, 'Exact completed state')
eq(len([e for e in events if e['event'] == 'protected-phase']), 3, 'All original protected phases')

phases = {}
for e in results:
    if e['event'] == 'original-read-result':
        phases.setdefault(e['statement']['phase'], {})[e['statement']['label']] = e
dates = ['EXPIRE_TIME', 'PAY_TIME', 'SHIP_TIME', 'CREATE_TIME', 'UPDATE_TIME']
unchanged = ['identity', 'version', 'indexes', 'index_columns', 'constraints', 'constraint_columns', 'foreign_keys', 'row_count_before', 'row_count_after']
for phase in ['pre_2ddl_fresh_stable', 'after_alter_stable', 'after_trigger_stable']:
    actual = phases[phase]
    eq(len(actual), 17, phase + ' all17 reads')
    calibrated = phase != 'pre_2ddl_fresh_stable'
    for label in unchanged:
        for key in ['rows', 'native_types', 'native_streams', 'native_float_bits', 'native_keys']:
            eq(actual[label][key], old[label][key], phase + ' protected metadata ' + label + '/' + key)
    columns = copy.deepcopy(old['columns']['rows'])
    ddl = old['table_ddl']['rows'][0]['ddl']
    if calibrated:
        for c in columns:
            if c['column_name'] in dates:
                c['data_scale'] = type(c['data_scale'])(0)
                if c['column_name'] in ['CREATE_TIME', 'UPDATE_TIME']:
                    c['data_default'] = 'CURRENT_TIMESTAMP(0)'
                    c['default_length'] = type(c['default_length'])(len(c['data_default']))
        for name in dates:
            before = '"' + name + '" DATETIME(6)'
            eq(ddl.count(before), 1, 'Exact date declaration ' + name)
            ddl = ddl.replace(before, '"' + name + '" DATETIME(0)')
            if name in ['CREATE_TIME', 'UPDATE_TIME']:
                ddl = ddl.replace('"' + name + '" DATETIME(0) DEFAULT CURRENT_TIMESTAMP NOT NULL',
                    '"' + name + '" DATETIME(0) DEFAULT CURRENT_TIMESTAMP(0) NOT NULL')
    eq(actual['columns']['rows'], columns, phase + ' only five precision/default/defaultLength changes with native types')
    eq(actual['columns']['native_types'], old['columns']['native_types'], phase + ' metadata native types unchanged')
    eq(actual['table_ddl']['rows'][0]['ddl'], ddl, phase + ' complete table DDL exact allowed changes')
    for label in ['rows_before', 'rows_after']:
        expected = copy.deepcopy(old[label]['rows'])
        if calibrated:
            for row in expected:
                for name in dates:
                    want = plan['baseline']['date_map'][row['id_exact']][name]['source_fsp0']
                    row[name.lower()] = want
                    row['f_' + name.lower()] = want + '.000000'
        eq(actual[label]['rows'], expected, phase + ' all27 original values except approved five-date map, complete LOB bytes')
        for key in ['native_types', 'native_streams', 'native_float_bits', 'native_keys']:
            eq(actual[label][key], old[label][key], phase + ' unchanged physical original evidence ' + key)
    eq(actual['rows_before']['rows'], actual['rows_after']['rows'], phase + ' whole old-row read-window stable')
    if phase != 'after_trigger_stable':
        for label in ['triggers', 'new_trigger_collision', 'objects', 'dependencies']:
            eq(actual[label]['rows'], old[label]['rows'], phase + ' unchanged object closure ' + label)

final = phases['after_trigger_stable']
trigger = final['triggers']['rows']
eq(len(trigger), 1, 'Exactly one new trigger')
eq(trigger, final['new_trigger_collision']['rows'], 'Both actual trigger queries agree')
t = trigger[0]
eq([t[k] for k in ['trigger_name', 'table_name', 'table_owner', 'status', 'triggering_type', 'triggering_event']],
    ['TRG_YDY_PREPAY_ORDER_UT', 'YDY_PREPAY_ORDER', 'ZHCT_APP', 'Y', 'BEFORE ROW', 'UPDATE'], 'Actual owner/event/enabled trigger')
expected_objects = sorted(old['objects']['rows'] + [{'object_name': 'TRG_YDY_PREPAY_ORDER_UT', 'object_type': 'TRIGGER', 'status': 'VALID'}], key=lambda r: r['object_name'])
eq(final['objects']['rows'], expected_objects, 'Actual TABLE and new trigger VALID')
eq(final['dependencies']['rows'], package['expected_owned_dependencies'], 'Only new owned dependency')

def tokens(sql):
    ts = re.findall(r"'(?:''|[^'])*'|\"(?:\"\"|[^\"])*\"|--[^\r\n]*|[A-Za-z_][A-Za-z0-9_$]*|(?:<>|:=|<=|>=)|[^\s]", sql)
    return [s if s.startswith(("'", '"')) else s.upper() for s in ts if not s.startswith('--')]

def trigger_parts(sql):
    m = re.search(r'FOR\s+EACH\s+ROW\b', sql, re.I)
    assert m
    head = [s.upper()[1:-1] if s.startswith('"') else s for s in tokens(sql[:m.end()]) if s not in ['OR', 'REPLACE']]
    return head, tokens(sql[m.end():])

approved = trigger_parts(plan['steps'][35]['SQL'])
eq(trigger_parts(t['trigger_body']), approved, 'Whole original USER_TRIGGERS header/body exact token contract')
definition = phases['after_trigger_definition']['trigger_get_ddl']['rows']
eq(len(definition), 1, 'One actual final trigger GET_DDL')
eq(trigger_parts(definition[0]['ddl']), approved, 'Whole actual GET_DDL header/body exact token contract')
print(json.dumps({'status': 'ROOT_FORMAL_PREPAY_INDEPENDENT_RAW_PASS', 'checks': checks,
    'actual_read': 52, 'actual_DDL': 2, 'DML_HTTP_Auth': 0, 'old_rows': 2,
    'other22_full_LOB_native_conserved': True, 'five_date_map_exact': True,
    'indexes_PK_FK_conserved': True, 'new_trigger_Y_VALID_full_definition': True,
    'private_raw_sha256': digest(RAW), 'new_connections_SQL': 0}, ensure_ascii=False))