"""Immutable adoption-scan requests, evidence, and reports.""" from __future__ import annotations import hashlib import json import re from dataclasses import dataclass, field from itertools import islice from types import MappingProxyType from typing import Mapping from ..domain.enums import AcceptedMappingOutcome, CandidateReviewOutcome from ..domain.models import OsmIdentity, RegionScopedIdentity from ..errors import ContractViolation from ..noco.models import NocoReadAudit, NocoReadRequest, WorkflowProtectionEvidence from ..normalization.models import NormalizationContext _SHA256 = re.compile(r"^[0-9a-f]{64}$") _MAX_MATCHED_FIELDS = 128 _MAX_COLLISION_REASONS = 128 def _text(value: object, field_name: str, max_length: int = 256) -> str: if not isinstance(value, str) or not value.strip(): raise ContractViolation(f"{field_name} must be a nonblank string") result = value.strip() if len(result) > max_length: raise ContractViolation(f"{field_name} exceeds the bounded length") return result def _positive(value: object, field_name: str) -> int: if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise ContractViolation(f"{field_name} must be a positive integer") return value @dataclass(frozen=True, slots=True) class ExistingActiveMapping: """Read evidence for one currently active sidecar mapping.""" noco_id: int identity: RegionScopedIdentity def __post_init__(self) -> None: object.__setattr__(self, "noco_id", _positive(self.noco_id, "noco_id")) if not isinstance(self.identity, RegionScopedIdentity): raise ContractViolation("identity must be RegionScopedIdentity") @dataclass(frozen=True, slots=True) class NormalizedNocoBusiness: """Canonical comparison fields detached from one Noco row.""" noco_id: int osm_identity: OsmIdentity | None name: str | None domain: str | None phone: str | None address: str | None city: str | None country: str | None category: str | None protection: WorkflowProtectionEvidence def __post_init__(self) -> None: object.__setattr__(self, "noco_id", _positive(self.noco_id, "noco_id")) if self.osm_identity is not None and not isinstance(self.osm_identity, OsmIdentity): raise ContractViolation("osm_identity must be OsmIdentity or null") for field_name in ("name", "domain", "phone", "address", "city", "country", "category"): value = getattr(self, field_name) if value is not None and (not isinstance(value, str) or not value): raise ContractViolation(f"{field_name} must be nonblank or null") if not isinstance(self.protection, WorkflowProtectionEvidence): raise ContractViolation("protection must be WorkflowProtectionEvidence") if self.protection.noco_id != self.noco_id: raise ContractViolation("protection evidence must reference the same Noco row") @dataclass(frozen=True, slots=True) class AdoptionScanRequest: """Explicit bounds and schema evidence for one report-only scan.""" run_id: str region_id: str table_id: str expected_schema_fingerprint: str normalization_context: NormalizationContext read_request: NocoReadRequest = NocoReadRequest() max_sources: int = 100_000 max_candidates: int = 500_000 max_existing_mappings: int = 100_000 max_sources_per_noco_row: int = 256 max_candidates_per_source: int = 16 def __post_init__(self) -> None: for field_name in ("run_id", "region_id", "table_id"): object.__setattr__(self, field_name, _text(getattr(self, field_name), field_name)) if not _SHA256.fullmatch(self.expected_schema_fingerprint): raise ContractViolation("expected_schema_fingerprint must be lowercase SHA-256") if not isinstance(self.normalization_context, NormalizationContext): raise ContractViolation("normalization_context must be NormalizationContext") if not isinstance(self.read_request, NocoReadRequest): raise ContractViolation("read_request must be NocoReadRequest") for field_name, upper in ( ("max_sources", 1_000_000), ("max_candidates", 2_000_000), ("max_existing_mappings", 1_000_000), ("max_sources_per_noco_row", 4_096), ("max_candidates_per_source", 128), ): value = getattr(self, field_name) if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= upper: raise ContractViolation(f"{field_name} exceeds its bounded range") @dataclass(frozen=True, slots=True) class AdoptionCandidate: """Immutable report evidence; never an active mapping or executable write.""" source_identity: RegionScopedIdentity noco_id: int | None result_status: CandidateReviewOutcome proposed_outcome: AcceptedMappingOutcome | None rule: str confidence_milli: int matched_fields: tuple[str, ...] protection: WorkflowProtectionEvidence | None collision_reasons: tuple[str, ...] = () def __post_init__(self) -> None: if not isinstance(self.source_identity, RegionScopedIdentity): raise ContractViolation("source_identity must be RegionScopedIdentity") if self.noco_id is not None: object.__setattr__(self, "noco_id", _positive(self.noco_id, "noco_id")) try: status = ( self.result_status if isinstance(self.result_status, CandidateReviewOutcome) else CandidateReviewOutcome(self.result_status) ) except (TypeError, ValueError) as exc: raise ContractViolation("result_status is unsupported") from exc object.__setattr__(self, "result_status", status) if self.proposed_outcome is not None: try: proposed = ( self.proposed_outcome if isinstance(self.proposed_outcome, AcceptedMappingOutcome) else AcceptedMappingOutcome(self.proposed_outcome) ) except (TypeError, ValueError) as exc: raise ContractViolation("proposed_outcome is unsupported") from exc object.__setattr__(self, "proposed_outcome", proposed) if status is CandidateReviewOutcome.ACCEPTED and self.proposed_outcome is None: raise ContractViolation("accepted candidate requires a proposed outcome") if status is not CandidateReviewOutcome.ACCEPTED and self.proposed_outcome is not None: raise ContractViolation("nonaccepted candidate cannot propose an accepted outcome") object.__setattr__(self, "rule", _text(self.rule, "rule")) if ( isinstance(self.confidence_milli, bool) or not isinstance(self.confidence_milli, int) or not 0 <= self.confidence_milli <= 1000 ): raise ContractViolation("confidence_milli must be within [0, 1000]") raw_fields = tuple(islice(iter(self.matched_fields), _MAX_MATCHED_FIELDS + 1)) if len(raw_fields) > _MAX_MATCHED_FIELDS: raise ContractViolation("matched_fields exceed the bounded count") raw_collisions = tuple(islice(iter(self.collision_reasons), _MAX_COLLISION_REASONS + 1)) if len(raw_collisions) > _MAX_COLLISION_REASONS: raise ContractViolation("collision_reasons exceed the bounded count") fields = tuple(sorted({_text(value, "matched_field", 128) for value in raw_fields})) collisions = tuple( sorted({_text(value, "collision_reason", 256) for value in raw_collisions}) ) object.__setattr__(self, "matched_fields", fields) object.__setattr__(self, "collision_reasons", collisions) if status is CandidateReviewOutcome.COLLISION and not collisions: raise ContractViolation("collision candidate requires collision reasons") if status is not CandidateReviewOutcome.COLLISION and collisions: raise ContractViolation("collision reasons require collision status") if self.protection is not None: if not isinstance(self.protection, WorkflowProtectionEvidence): raise ContractViolation("protection must be WorkflowProtectionEvidence or null") if self.noco_id != self.protection.noco_id: raise ContractViolation("protection must reference the candidate Noco row") @dataclass(frozen=True, slots=True) class AdoptionScanReport: """Deterministic, secret-free result of one bounded adoption scan.""" run_id: str region_id: str table_id: str schema_fingerprint: str candidates: tuple[AdoptionCandidate, ...] read_audit: NocoReadAudit counts: Mapping[str, int] = field(init=False) report_hash: str = field(init=False) def __post_init__(self) -> None: for field_name in ("run_id", "region_id", "table_id"): object.__setattr__(self, field_name, _text(getattr(self, field_name), field_name)) if not _SHA256.fullmatch(self.schema_fingerprint): raise ContractViolation("schema_fingerprint must be lowercase SHA-256") if not isinstance(self.read_audit, NocoReadAudit): raise ContractViolation("read_audit must be NocoReadAudit") if self.read_audit.schema_fingerprint != self.schema_fingerprint: raise ContractViolation("read audit schema fingerprint does not match report") raw_candidates = tuple(islice(iter(self.candidates), 2_000_001)) if len(raw_candidates) > 2_000_000: raise ContractViolation("candidates exceed the hard report bound") if any(not isinstance(candidate, AdoptionCandidate) for candidate in raw_candidates): raise ContractViolation("candidates must contain AdoptionCandidate values") candidates = tuple( sorted( raw_candidates, key=lambda item: ( item.source_identity.osm_type.value, item.source_identity.osm_id, item.noco_id or 0, item.result_status.value, item.rule, ), ) ) candidate_keys = [ ( candidate.source_identity.region_id, candidate.source_identity.osm_type.value, candidate.source_identity.osm_id, candidate.noco_id, candidate.result_status.value, candidate.proposed_outcome.value if candidate.proposed_outcome else None, candidate.rule, candidate.confidence_milli, candidate.matched_fields, candidate.protection.reasons if candidate.protection else (), candidate.collision_reasons, ) for candidate in candidates ] if len(candidate_keys) != len(set(candidate_keys)): raise ContractViolation("report contains duplicate candidate evidence") object.__setattr__(self, "candidates", candidates) counts: dict[str, int] = {status.value: 0 for status in CandidateReviewOutcome} for candidate in candidates: counts[candidate.result_status.value] += 1 object.__setattr__(self, "counts", MappingProxyType(counts)) canonical = json.dumps( { "candidates": [ { "collision_reasons": candidate.collision_reasons, "confidence_milli": candidate.confidence_milli, "matched_fields": candidate.matched_fields, "noco_id": candidate.noco_id, "proposed_outcome": ( candidate.proposed_outcome.value if candidate.proposed_outcome else None ), "protection": ( { "protected": candidate.protection.protected, "reasons": candidate.protection.reasons, } if candidate.protection else None ), "rule": candidate.rule, "source": ( candidate.source_identity.region_id, candidate.source_identity.osm_type.value, candidate.source_identity.osm_id, ), "status": candidate.result_status.value, } for candidate in candidates ], "read": { "max_noco_id": self.read_audit.max_noco_id, "pages_read": self.read_audit.pages_read, "repository_kind": self.read_audit.repository_kind, "rows_read": self.read_audit.rows_read, }, "region_id": self.region_id, "run_id": self.run_id, "schema_fingerprint": self.schema_fingerprint, "table_id": self.table_id, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") object.__setattr__(self, "report_hash", hashlib.sha256(canonical).hexdigest())