"""Immutable duplicate-candidate contracts.""" from __future__ import annotations import hashlib import json import re from dataclasses import dataclass, field from enum import Enum from ..domain.models import RegionScopedIdentity from ..errors import ContractViolation _SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") def _identity_key(identity: RegionScopedIdentity) -> tuple[str, str, int]: return (identity.region_id, identity.osm_type.value, identity.osm_id) def _text_tuple(values: tuple[str, ...], field_name: str) -> tuple[str, ...]: if isinstance(values, str): raise ContractViolation(f"{field_name} must be an iterable, not a string") if any(not isinstance(value, str) or not value.strip() for value in values): raise ContractViolation(f"{field_name} must contain nonblank strings") return tuple(sorted(set(value.strip() for value in values))) class DuplicateCandidateStrength(str, Enum): """Review strength only; no value authorizes an automatic merge.""" EXACT_EVIDENCE = "exact_evidence" HIGH_CONFIDENCE = "high_confidence" NEEDS_REVIEW = "needs_review" COLLISION = "collision" @dataclass(frozen=True, slots=True) class DuplicateCandidateLimits: """Hard bounds for one duplicate-candidate generation pass.""" max_records: int = 100_000 max_bucket_size: int = 200 max_candidates: int = 1_000_000 def __post_init__(self) -> None: for field_name, hard_limit in ( ("max_records", 1_000_000), ("max_bucket_size", 1_000), ("max_candidates", 2_000_000), ): value = getattr(self, field_name) if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise ContractViolation(f"{field_name} must be a positive integer") if value > hard_limit: raise ContractViolation(f"{field_name} exceeds the Phase 2A.4 hard limit") @dataclass(frozen=True, slots=True) class DuplicateCandidate: """A deterministic pair requiring later review or adoption logic.""" left: RegionScopedIdentity right: RegionScopedIdentity strength: DuplicateCandidateStrength score: int reason_codes: tuple[str, ...] evidence: tuple[str, ...] candidate_key: str = field(init=False) auto_merge_allowed: bool = False def __post_init__(self) -> None: if not isinstance(self.left, RegionScopedIdentity) or not isinstance( self.right, RegionScopedIdentity ): raise ContractViolation("candidate identities must be RegionScopedIdentity values") left_key = _identity_key(self.left) right_key = _identity_key(self.right) if self.left.region_id != self.right.region_id: raise ContractViolation("duplicate candidate identities must share one source region") if left_key == right_key: raise ContractViolation("duplicate candidate identities must differ") if right_key < left_key: original_left = self.left object.__setattr__(self, "left", self.right) object.__setattr__(self, "right", original_left) left_key, right_key = right_key, left_key try: strength = ( self.strength if isinstance(self.strength, DuplicateCandidateStrength) else DuplicateCandidateStrength(self.strength) ) except (TypeError, ValueError) as exc: raise ContractViolation("candidate strength is not supported") from exc object.__setattr__(self, "strength", strength) if ( isinstance(self.score, bool) or not isinstance(self.score, int) or not 0 <= self.score <= 100 ): raise ContractViolation("candidate score must be an integer from 0 to 100") reasons = _text_tuple(self.reason_codes, "reason_codes") evidence = _text_tuple(self.evidence, "evidence") if not reasons or not evidence: raise ContractViolation("candidate reasons and evidence are required") object.__setattr__(self, "reason_codes", reasons) object.__setattr__(self, "evidence", evidence) if self.auto_merge_allowed: raise ContractViolation("automatic duplicate merging is unavailable in Phase 2A.4") canonical = json.dumps( { "left": left_key, "reasons": reasons, "right": right_key, "strength": strength.value, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") object.__setattr__(self, "candidate_key", hashlib.sha256(canonical).hexdigest()) @dataclass(frozen=True, slots=True) class DuplicateCandidateReport: """Deterministic bounded report over normalized records.""" records_examined: int bucket_count: int candidates: tuple[DuplicateCandidate, ...] skipped_oversized_bucket_hashes: tuple[str, ...] = () report_digest: str = field(init=False) def __post_init__(self) -> None: for field_name in ("records_examined", "bucket_count"): value = getattr(self, field_name) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise ContractViolation(f"{field_name} must be a nonnegative integer") raw_candidates = tuple(self.candidates) if any(not isinstance(item, DuplicateCandidate) for item in raw_candidates): raise ContractViolation("candidates must contain DuplicateCandidate values") candidates = tuple(sorted(raw_candidates, key=lambda item: item.candidate_key)) if len({item.candidate_key for item in candidates}) != len(candidates): raise ContractViolation("candidate keys must be unique") candidate_regions = {item.left.region_id for item in candidates} | { item.right.region_id for item in candidates } if len(candidate_regions) > 1: raise ContractViolation("candidate report must contain exactly one source region") referenced_identities = { _identity_key(identity) for item in candidates for identity in (item.left, item.right) } if len(referenced_identities) > self.records_examined: raise ContractViolation("candidate report references more identities than examined") if candidates and self.bucket_count == 0: raise ContractViolation("candidate report with candidates requires at least one bucket") object.__setattr__(self, "candidates", candidates) hashes = tuple(sorted(set(self.skipped_oversized_bucket_hashes))) if any(not _SHA256_PATTERN.fullmatch(value) for value in hashes): raise ContractViolation("oversized bucket references must be SHA-256 hashes") object.__setattr__(self, "skipped_oversized_bucket_hashes", hashes) canonical = json.dumps( { "bucket_count": self.bucket_count, "candidate_keys": tuple(item.candidate_key for item in candidates), "records_examined": self.records_examined, "skipped": hashes, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") object.__setattr__(self, "report_digest", hashlib.sha256(canonical).hexdigest())