from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Iterable from ..sync_system.strategy_ops.compare_ops import normalized_data_for_compare @dataclass class SchemaDiffSample: sample_id: str local: Any remote: Any @dataclass class SchemaFieldDiffStat: field_name: str compared_count: int = 0 mismatch_count: int = 0 schema_refs: set[str] = field(default_factory=set) samples: list[SchemaDiffSample] = field(default_factory=list) class SchemaDiffValidator: def __init__( self, *, schema_name: str, ignore_fields: Iterable[str] | None = None, sample_limit: int = 3, ) -> None: self.schema_name = schema_name self.ignore_fields = {str(field) for field in (ignore_fields or []) if str(field)} self.sample_limit = max(1, sample_limit) self.reset() def reset(self) -> None: self.pairs_compared = 0 self.mismatched_pairs = 0 self._field_stats: dict[str, SchemaFieldDiffStat] = {} def has_records(self) -> bool: return self.pairs_compared > 0 def compare_payloads( self, local_payload: Mapping[str, Any] | None, remote_payload: Mapping[str, Any] | None, *, sample_id: str | None = None, schema_refs: Iterable[str] | None = None, ) -> dict[str, dict[str, Any]]: local_normalized = normalized_data_for_compare( dict(local_payload or {}), ignore_fields=self.ignore_fields, ) remote_normalized = normalized_data_for_compare( dict(remote_payload or {}), ignore_fields=self.ignore_fields, ) field_names = sorted(set(local_normalized) | set(remote_normalized)) diff_map: dict[str, dict[str, Any]] = {} refs = {self.schema_name, *(str(item) for item in (schema_refs or []) if str(item))} self.pairs_compared += 1 current_sample_id = sample_id or f"pair-{self.pairs_compared}" for field_name in field_names: stat = self._field_stats.setdefault(field_name, SchemaFieldDiffStat(field_name=field_name)) stat.compared_count += 1 stat.schema_refs.update(refs) local_value = local_normalized.get(field_name) remote_value = remote_normalized.get(field_name) if local_value == remote_value: continue stat.mismatch_count += 1 if len(stat.samples) < self.sample_limit: stat.samples.append( SchemaDiffSample( sample_id=current_sample_id, local=local_value, remote=remote_value, ) ) diff_map[field_name] = {"local": local_value, "remote": remote_value} if diff_map: self.mismatched_pairs += 1 return diff_map def build_report(self) -> dict[str, Any]: fields = [] for stat in sorted( self._field_stats.values(), key=lambda item: (-item.mismatch_count, item.field_name), ): if stat.compared_count <= 0: continue fields.append( { "field_name": stat.field_name, "mismatch_count": stat.mismatch_count, "total_count": stat.compared_count, "mismatch_rate": stat.mismatch_count / stat.compared_count, "schema_refs": sorted(stat.schema_refs), "samples": [ { "sample_id": sample.sample_id, "local": sample.local, "remote": sample.remote, } for sample in stat.samples ], } ) return { "schema_name": self.schema_name, "pairs_compared": self.pairs_compared, "mismatched_pairs": self.mismatched_pairs, "ignore_fields": sorted(self.ignore_fields), "fields": fields, } def format_summary_lines(self, *, max_fields: int = 20, max_samples: int = 1) -> list[str]: report = self.build_report() lines = [ ( f"schema={report['schema_name']} pairs={report['pairs_compared']} " f"mismatched_pairs={report['mismatched_pairs']} ignored={report['ignore_fields']}" ) ] for field_report in report["fields"][:max_fields]: line = ( f"field={field_report['field_name']} mismatch={field_report['mismatch_count']}/" f"{field_report['total_count']} refs={field_report['schema_refs']}" ) samples = field_report["samples"][:max_samples] if samples: sample = samples[0] line += f" sample[{sample['sample_id']}]={sample['local']!r}/{sample['remote']!r}" lines.append(line) return lines