scripts/reconcile_fixture.py

"""Local deterministic data-quality fixtures. Not a live data pipeline validator."""
from __future__ import annotations
import json
import sys
from collections import Counter
from pathlib import Path

def key_violations(records: list[dict], key_fields: list[str]) -> dict:
    """Return missing/null and repeated composite keys; never silently dedup rows."""
    if not key_fields:
        raise ValueError("key_fields must not be empty")
    keys=[]; missing=[]
    for i,row in enumerate(records):
        if any(field not in row or row[field] is None for field in key_fields):
            missing.append(i); continue
        values=tuple(row[field] for field in key_fields)
        try: hash(values)
        except TypeError as exc: raise ValueError("keys must be scalar/hashable") from exc
        keys.append(values)
    counts=Counter(keys)
    duplicates=[{'key':list(key),'count':count} for key,count in counts.items() if count>1]
    duplicates.sort(key=lambda entry:repr(entry['key']))
    return {'rows':len(records),'missing_key_rows':missing,'duplicate_keys':duplicates,'unique_nonnull_keys':len(counts),'valid':not missing and not duplicates}

def compare_key_sets(source: list[dict], target: list[dict], key_fields: list[str]) -> dict:
    """Read-only distinct-key comparison, not a row/value equivalence proof."""
    def project(rows):
        return {tuple(row[f] for f in key_fields) for row in rows if all(f in row and row[f] is not None for f in key_fields)}
    source_keys, target_keys=project(source),project(target)
    return {'missing_target_keys':sorted(map(list, source_keys-target_keys),key=repr),
            'unexpected_target_keys':sorted(map(list, target_keys-source_keys),key=repr),
            'source_unique_keys':len(source_keys),'target_unique_keys':len(target_keys)}

def cdc_latest(events: list[dict], key: str='id', sequence: str='seq') -> dict:
    """Fixture-only deterministic latest by strictly ordered numeric source seq; errors on ties/conflicts."""
    state={}
    for event in events:
        entity=event[key];seq=event[sequence];op=event.get('op')
        if entity is None or not isinstance(seq,(int,float)) or isinstance(seq,bool) or op not in ('upsert','delete'):
            raise ValueError('invalid CDC event')
        if entity not in state or seq>state[entity][sequence]:
            state[entity]=dict(event)
        elif seq==state[entity][sequence] and event!=state[entity]:
            raise ValueError('conflicting CDC events with same source sequence')
    return {k:v for k,v in state.items() if v['op']=='upsert'}

def main(argv):
    if len(argv)!=2:
        raise SystemExit('Usage: python scripts/reconcile_fixture.py benchmarks/fixtures/reconciliation.json')
    data=json.loads(Path(argv[1]).read_text(encoding='utf-8'))
    key=data['key_fields']; source=data['source'];target=data['target']
    print(json.dumps({'source':key_violations(source,key),'target':key_violations(target,key),
                      'key_set_comparison':compare_key_sets(source,target,key)},indent=2))
if __name__=='__main__': main(sys.argv)