#!/usr/bin/env python3 """Independent, read-only MySQL HEX -> DM string/null reconciliation. Exit 0 means complete verified equality; exit 2 means only a declared subset matched; exit 1 means mismatch or invalid evidence. Never prints customer rows. """ import argparse from collections import Counter from decimal import Decimal, InvalidOperation import gzip import hashlib import json import math from pathlib import Path import re import struct INT_TYPES = {'tinyint', 'smallint', 'mediumint', 'int', 'integer', 'bigint', 'year'} TEXT_TYPES = {'char', 'varchar', 'tinytext', 'text', 'mediumtext', 'longtext', 'enum', 'set'} TEMPORAL_TYPES = {'date', 'datetime', 'timestamp', 'time'} NUMERIC = re.compile(r'^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$') NAME = re.compile(r'^[A-Za-z0-9_]+$') class ContractError(ValueError): """Messages must describe structure only, never include raw field values.""" def canonical(value, column): if value is None: return ('null',) if not isinstance(value, str): raise ContractError('Expected string/null cell') kind = column['data_type'].lower() if kind in INT_TYPES or kind in {'decimal', 'numeric', 'float', 'double', 'real'}: if not NUMERIC.fullmatch(value): raise ContractError('Invalid numeric encoding') try: number = Decimal(value) except InvalidOperation: raise ContractError('Invalid decimal') if not number.is_finite(): raise ContractError('Nonfinite decimal') if kind in INT_TYPES and number != number.to_integral_value(): raise ContractError('Nonintegral integer') if kind in {'float', 'real', 'double'}: native = float(value) if not math.isfinite(native): raise ContractError('Nonfinite floating point') fmt = '>d' if kind == 'double' else '>f' try: packed = struct.pack(fmt, native) except (OverflowError, struct.error): raise ContractError('Floating point overflow') if not math.isfinite(struct.unpack(fmt, packed)[0]): raise ContractError('Floating point overflow') return ('ieee64' if kind == 'double' else 'ieee32', packed.hex()) # Decimal.normalize uses context precision and can silently round large # coefficients. Format without arithmetic, then trim fractional zeros. exact = format(number, 'f') if '.' in exact: exact = exact.rstrip('0').rstrip('.') if number == 0: exact = '0' return ('number', exact) if kind == 'char': return ('text', value.rstrip(' ')) if kind in TEXT_TYPES: return ('text', value) if kind in TEMPORAL_TYPES: # No zero-date, timezone or sentinel normalization is implicit here. return ('temporal', value) raise ContractError('Unsupported source data type: ' + kind) def read_rows(file, width, source_hex=False): with gzip.open(file, 'rt', encoding='utf-8') as stream: for line in stream: row = json.loads(line) if not isinstance(row, list) or len(row) != width: raise ContractError('Row shape differs from source metadata') if source_hex: decoded = [] for cell in row: if cell is None: decoded.append(None) elif isinstance(cell, str) and re.fullmatch(r'(?:[0-9A-Fa-f]{2})*', cell): decoded.append(bytes.fromhex(cell).decode('utf-8')) else: raise ContractError('Malformed source HEX encoding') row = decoded yield row def table_counter(file, table, columns, float_source=None): source = float_source is not None float_indexes = [i for i, c in enumerate(columns) if c['data_type'].lower() in {'float', 'double', 'real'}] id_indexes = [i for i, c in enumerate(columns) if c['name'] == 'id'] counts = Counter() rows = 0 nonnull_float_cells = 0 for row in read_rows(file, len(columns), source_hex=source): if source and float_indexes: if len(id_indexes) != 1 or row[id_indexes[0]] is None: raise ContractError('Floating source requires unique id column mapping') key = row[id_indexes[0]] values = float_source.get(table, {}).get(key, {}) for index in float_indexes: name = columns[index]['name'] if row[index] is not None: if name not in values or not isinstance(values[name], str): raise ContractError('Missing precise MySQL floating source cell') row[index] = values[name] nonnull_float_cells += 1 elif name in values and values[name] is not None: raise ContractError('Precise floating source contradicts NULL baseline') normalized = [canonical(cell, col) for cell, col in zip(row, columns)] digest = hashlib.sha256(json.dumps(normalized, ensure_ascii=False, separators=(',', ':')).encode('utf-8')).digest() counts[digest] += 1 rows += 1 return counts, rows, nonnull_float_cells def compare(baseline_file, float_file, target_dir, manifest_file): baseline_file, target_dir = Path(baseline_file), Path(target_dir) baseline = json.loads(baseline_file.read_text(encoding='utf-8')) floats = json.loads(Path(float_file).read_text(encoding='utf-8')) manifest = json.loads(Path(manifest_file).read_text(encoding='utf-8')) metadata = baseline['metadata'] declared_counts = manifest['counts'] if not isinstance(metadata, dict) or not isinstance(declared_counts, dict): raise ContractError('Missing table metadata/counts') if manifest.get('errors'): raise ContractError('Target export reports errors') if manifest.get('completed') is False: raise ContractError('Target export incomplete') deferred_entries = manifest.get('deferred_tables', []) if not isinstance(deferred_entries, list): raise ContractError('Invalid deferred table list') deferred = [] for entry in deferred_entries: if isinstance(entry, str): table = entry elif isinstance(entry, dict) and isinstance(entry.get('table'), str): table = entry['table'] if (type(entry.get('rows')) is not int or entry['rows'] != baseline['counts'].get(table)): raise ContractError('Deferred row count differs from baseline') if ('zero_date_cells' in entry and (type(entry['zero_date_cells']) is not int or entry['zero_date_cells'] < 0)): raise ContractError('Invalid deferred zero-date cell count') else: raise ContractError('Invalid deferred table entry') deferred.append(table) if len(set(deferred)) != len(deferred): raise ContractError('Duplicate deferred table') if set(deferred) - set(metadata): raise ContractError('Deferred tables unknown to source') expected = set(metadata) - set(deferred) if set(declared_counts) != expected: raise ContractError('Exported table set differs from source minus explicit deferred tables') if any(not NAME.fullmatch(name) for name in metadata): raise ContractError('Unsafe table filename') if baseline.get('tables') != len(metadata): raise ContractError('Source table count manifest mismatch') if set(baseline['counts']) != set(metadata): raise ContractError('Source count manifest table set mismatch') if sum(baseline['counts'].values()) != baseline['rows']: raise ContractError('Source total count manifest mismatch') reports = [] for table in sorted(expected): columns = metadata[table] if len({c['name'] for c in columns}) != len(columns): raise ContractError('Duplicate source column names') if 'columns' in manifest and manifest['columns'].get(table) != [c['name'] for c in columns]: raise ContractError('Target column order mismatch: ' + table) source_file = baseline_file.parent / (table + '.hex.jsonl.gz') target_file = target_dir / (table + '.jsonl.gz') try: source_counts, source_rows, float_cells = table_counter(source_file, table, columns, floats) target_counts, target_rows, _ = table_counter(target_file, table, columns) except (ContractError, UnicodeError, json.JSONDecodeError, OSError, ValueError) as error: # Underlying decoder messages may contain private data; don't emit them. raise ContractError('Invalid evidence for table ' + table + ' (' + type(error).__name__ + ')') count_ok = (source_rows == baseline['counts'][table] and target_rows == declared_counts[table]) missing = sum((source_counts - target_counts).values()) extra = sum((target_counts - source_counts).values()) reports.append({'table': table, 'source_rows': source_rows, 'target_rows': target_rows, 'manifest_counts_valid': count_ok, 'precise_float_cells': float_cells, 'missing_or_changed_rows': missing, 'extra_or_changed_rows': extra, 'equal': count_ok and missing == 0 and extra == 0}) subset_equal = all(item['equal'] for item in reports) partial = bool(deferred) or manifest.get('partial_rehearsal_only') is True equal = subset_equal and not partial return {'equal': equal, 'verified_subset_equal': subset_equal, 'status': 'mismatch' if not subset_equal else 'partial' if partial else 'complete', 'source_database': baseline['db'], 'target_schema': manifest.get('schema'), 'target_build': manifest.get('build'), 'source_tables': len(metadata), 'verified_tables': len(reports), 'source_total_rows': baseline['rows'], 'verified_source_rows': sum(x['source_rows'] for x in reports), 'verified_target_rows': sum(x['target_rows'] for x in reports), 'deferred_tables': sorted(deferred), 'different_tables': [x['table'] for x in reports if not x['equal']], 'tables': reports} def main(): parser = argparse.ArgumentParser(description=__doc__) for name in ('baseline', 'float-source', 'target-dir', 'manifest', 'output'): parser.add_argument('--' + name, required=True) args = parser.parse_args() try: report = compare(args.baseline, args.float_source, args.target_dir, args.manifest) except Exception as error: report = {'equal': False, 'verified_subset_equal': False, 'status': 'invalid_evidence', 'error': str(error) if isinstance(error, ContractError) else type(error).__name__} Path(args.output).write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') print(json.dumps({k: v for k, v in report.items() if k != 'tables'}, ensure_ascii=False)) return 0 if report['equal'] else 2 if report['status'] == 'partial' else 1 if __name__ == '__main__': raise SystemExit(main())