"""Deterministic local source-shadow reconciliation.""" from __future__ import annotations import hashlib import json import re from collections import Counter from dataclasses import dataclass, field from typing import Any from .domain.enums import ProfileEligibilityState from .domain.models import ProfileVersion, RegionScopedIdentity from .errors import ContractViolation from .normalization.models import NormalizationContext, NormalizedBusiness from .normalization.normalize import ( normalize_business, raw_business_fields_from_parsed_object, ) from .pbf.models import ( ParsedOsmObject, PbfObjectCounts, PbfParseSummary, VerifiedPbfInput, ) from .pbf.parser import OsmiumPbfParser from .profiles.engine import ProfileEvaluator from .profiles.models import ProfileDecision, ProfileDefinition _SHA256 = re.compile(r"^[0-9a-f]{64}$") _MAX_ITEMS = 1_000_000 _MAX_EVIDENCE = 256 def _text(value: object, field: str, maximum: int = 512) -> str: if not isinstance(value, str) or not value.strip(): raise ContractViolation(f"{field} must be a nonblank string") result = value.strip() if len(result) > maximum: raise ContractViolation(f"{field} exceeds the bounded length") return result def _digest(value: object, field: str) -> str: if not isinstance(value, str) or not _SHA256.fullmatch(value): raise ContractViolation(f"{field} must be lowercase SHA-256") return value def _evidence(values: tuple[str, ...], field: str) -> tuple[str, ...]: raw = tuple(values) if len(raw) > _MAX_EVIDENCE: raise ContractViolation(f"{field} exceeds the bounded count") return tuple(sorted({_text(value, field) for value in raw})) def _identity_key(identity: RegionScopedIdentity) -> tuple[str, str, int]: return identity.region_id, identity.osm_type.value, identity.osm_id @dataclass(frozen=True, slots=True) class ShadowLead: """One profile-active normalized OSM business.""" business: NormalizedBusiness profile: ProfileVersion definition_digest: str category: str subcategory: str | None matched_rule_ids: tuple[str, ...] = () evidence: tuple[str, ...] = () def __post_init__(self) -> None: if not isinstance(self.business, NormalizedBusiness): raise ContractViolation("business must be NormalizedBusiness") if not isinstance(self.profile, ProfileVersion): raise ContractViolation("profile must be ProfileVersion") object.__setattr__( self, "definition_digest", _digest(self.definition_digest, "definition_digest"), ) object.__setattr__(self, "category", _text(self.category, "category")) if self.subcategory is not None: object.__setattr__(self, "subcategory", _text(self.subcategory, "subcategory")) object.__setattr__( self, "matched_rule_ids", _evidence(self.matched_rule_ids, "matched_rule_ids"), ) object.__setattr__(self, "evidence", _evidence(self.evidence, "evidence")) @property def identity(self) -> RegionScopedIdentity: return self.business.identity def as_mapping(self) -> dict[str, Any]: business = self.business return { "address": business.address, "category": self.category, "city": business.city, "country": business.country, "domain": business.domain, "evidence": list(self.evidence), "identity_hash": business.identity_hash, "latitude_e7": business.latitude_e7, "longitude_e7": business.longitude_e7, "matched_rule_ids": list(self.matched_rule_ids), "name": business.name, "osm_id": business.identity.osm_id, "osm_type": business.identity.osm_type.value, "phone": business.phone, "postcode": business.postcode, "source_tags_hash": business.source_tags_hash, "state": business.state, "subcategory": self.subcategory, } @dataclass(frozen=True, slots=True) class ShadowReviewItem: """One profile decision that requires human review.""" decision: ProfileDecision def __post_init__(self) -> None: if not isinstance(self.decision, ProfileDecision): raise ContractViolation("decision must be ProfileDecision") if self.decision.state is not ProfileEligibilityState.PENDING_REVIEW: raise ContractViolation("review item must contain a pending-review decision") @property def identity(self) -> RegionScopedIdentity: return self.decision.identity def as_mapping(self) -> dict[str, Any]: return { "evidence": list(self.decision.evidence), "matched_rule_ids": list(self.decision.matched_rule_ids), "osm_id": self.identity.osm_id, "osm_type": self.identity.osm_type.value, } @dataclass(frozen=True, slots=True) class ShadowSourceReport: """Repeatable source/profile report from one verified local PBF.""" region_id: str profile: ProfileVersion definition_digest: str source_sha256: str source_size_bytes: int object_counts: PbfObjectCounts total_tags: int state_counts: tuple[tuple[str, int], ...] category_counts: tuple[tuple[str, int], ...] leads: tuple[ShadowLead, ...] = () review_items: tuple[ShadowReviewItem, ...] = () report_hash: str = field(init=False) def __post_init__(self) -> None: object.__setattr__(self, "region_id", _text(self.region_id, "region_id")) if not isinstance(self.profile, ProfileVersion): raise ContractViolation("profile must be ProfileVersion") object.__setattr__( self, "definition_digest", _digest(self.definition_digest, "definition_digest"), ) object.__setattr__(self, "source_sha256", _digest(self.source_sha256, "source_sha256")) if ( isinstance(self.source_size_bytes, bool) or not isinstance(self.source_size_bytes, int) or self.source_size_bytes <= 0 ): raise ContractViolation("source_size_bytes must be a positive integer") if not isinstance(self.object_counts, PbfObjectCounts): raise ContractViolation("object_counts must be PbfObjectCounts") if ( isinstance(self.total_tags, bool) or not isinstance(self.total_tags, int) or self.total_tags < 0 ): raise ContractViolation("total_tags must be a nonnegative integer") states = self._counts(self.state_counts, "state_counts") categories = self._counts(self.category_counts, "category_counts") if set(dict(states)) - {state.value for state in ProfileEligibilityState}: raise ContractViolation("state_counts contains an unsupported state") if sum(value for _, value in states) != self.object_counts.total: raise ContractViolation("state_counts must total the parsed object count") leads = tuple(sorted(tuple(self.leads), key=lambda item: _identity_key(item.identity))) reviews = tuple( sorted(tuple(self.review_items), key=lambda item: _identity_key(item.identity)) ) if len(leads) > _MAX_ITEMS or len(reviews) > _MAX_ITEMS: raise ContractViolation("shadow report exceeds the bounded item count") if any(not isinstance(item, ShadowLead) for item in leads): raise ContractViolation("leads must contain ShadowLead values") if any(not isinstance(item, ShadowReviewItem) for item in reviews): raise ContractViolation("review_items must contain ShadowReviewItem values") lead_keys = [_identity_key(item.identity) for item in leads] review_keys = [_identity_key(item.identity) for item in reviews] if len(lead_keys) != len(set(lead_keys)) or len(review_keys) != len(set(review_keys)): raise ContractViolation("shadow report contains duplicate source identities") if set(lead_keys) & set(review_keys): raise ContractViolation("source identity cannot be both active and pending review") if any(item.identity.region_id != self.region_id for item in leads + reviews): raise ContractViolation("all report items must belong to region_id") if any( item.profile != self.profile or item.definition_digest != self.definition_digest for item in leads ): raise ContractViolation("all leads must use the report profile definition") if any( item.decision.profile != self.profile or item.decision.definition_digest != self.definition_digest for item in reviews ): raise ContractViolation("all review items must use the report profile definition") active = dict(states).get(ProfileEligibilityState.ACTIVE.value, 0) pending = dict(states).get(ProfileEligibilityState.PENDING_REVIEW.value, 0) if len(leads) != active or len(reviews) != pending: raise ContractViolation("report items must match eligibility state counts") if sum(value for _, value in categories) != active: raise ContractViolation("category_counts must total the active state count") object.__setattr__(self, "state_counts", states) object.__setattr__(self, "category_counts", categories) object.__setattr__(self, "leads", leads) object.__setattr__(self, "review_items", reviews) canonical = json.dumps( self.as_mapping(include_hash=False), sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode() object.__setattr__(self, "report_hash", hashlib.sha256(canonical).hexdigest()) @staticmethod def _counts(values: tuple[tuple[str, int], ...], field: str) -> tuple[tuple[str, int], ...]: result: list[tuple[str, int]] = [] for item in tuple(values): if not isinstance(item, tuple) or len(item) != 2: raise ContractViolation(f"{field} contains an invalid entry") key, count = item if isinstance(count, bool) or not isinstance(count, int) or count < 0: raise ContractViolation(f"{field} contains an invalid count") result.append((_text(key, field), count)) if len({key for key, _ in result}) != len(result): raise ContractViolation(f"{field} keys must be unique") return tuple(sorted(result)) def as_mapping(self, *, include_hash: bool = True) -> dict[str, Any]: result: dict[str, Any] = { "category_counts": dict(self.category_counts), "definition_digest": self.definition_digest, "leads": [item.as_mapping() for item in self.leads], "object_counts": dict(self.object_counts.as_mapping()), "profile": {"id": self.profile.profile_id, "version": self.profile.version}, "region_id": self.region_id, "review_items": [item.as_mapping() for item in self.review_items], "source_sha256": self.source_sha256, "source_size_bytes": self.source_size_bytes, "state_counts": dict(self.state_counts), "total_tags": self.total_tags, } if include_hash: result["report_hash"] = self.report_hash return result def to_json(self) -> str: return json.dumps( self.as_mapping(), sort_keys=True, separators=(",", ":"), ensure_ascii=False ) class ProfileShadowSink: """Stream profile evaluation and normalization without persistence.""" def __init__( self, *, region_id: str, evaluator: ProfileEvaluator, definition: ProfileDefinition, normalization_context: NormalizationContext, max_leads: int = 500_000, max_review_items: int = 100_000, verify_unique_objects: bool = True, ) -> None: self._region_id = _text(region_id, "region_id") if not isinstance(evaluator, ProfileEvaluator): raise ContractViolation("evaluator must be ProfileEvaluator") if not isinstance(definition, ProfileDefinition): raise ContractViolation("definition must be ProfileDefinition") if not isinstance(normalization_context, NormalizationContext): raise ContractViolation("normalization_context must be NormalizationContext") for field_name, value in ( ("max_leads", max_leads), ("max_review_items", max_review_items), ): if ( isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= _MAX_ITEMS ): raise ContractViolation(f"{field_name} exceeds its bounded range") self._evaluator = evaluator self._definition = definition self._context = normalization_context self._max_leads = max_leads if not isinstance(verify_unique_objects, bool): raise ContractViolation("verify_unique_objects must be a boolean") self._max_reviews = max_review_items self._verify_unique_objects = verify_unique_objects self._states: Counter[str] = Counter() self._categories: Counter[str] = Counter() self._leads: list[ShadowLead] = [] self._reviews: list[ShadowReviewItem] = [] self._seen: set[tuple[str, int]] = set() self._object_count = 0 self._summary: PbfParseSummary | None = None self._aborted = False def accept(self, parsed_object: ParsedOsmObject) -> None: if self._summary is not None or self._aborted: raise ContractViolation("shadow sink is no longer accepting objects") if not isinstance(parsed_object, ParsedOsmObject): raise ContractViolation("parsed_object must be ParsedOsmObject") key = (parsed_object.identity.osm_type.value, parsed_object.identity.osm_id) if self._verify_unique_objects: if key in self._seen: raise ContractViolation("shadow input contains a duplicate source identity") self._seen.add(key) self._object_count += 1 decision = self._evaluator.evaluate( region_id=self._region_id, parsed_object=parsed_object, profile=self._definition.profile, ) self._states[decision.state.value] += 1 if decision.state is ProfileEligibilityState.ACTIVE: if len(self._leads) >= self._max_leads: raise ContractViolation("active shadow leads exceed max_leads") business = normalize_business( raw_business_fields_from_parsed_object( region_id=self._region_id, parsed_object=parsed_object, context=self._context, ), self._context, ) if decision.category is None: raise ContractViolation("active profile decision has no category") category_key = ( decision.category if decision.subcategory is None else f"{decision.category}/{decision.subcategory}" ) self._categories[category_key] += 1 self._leads.append( ShadowLead( business, decision.profile, decision.definition_digest, decision.category, decision.subcategory, decision.matched_rule_ids, decision.evidence, ) ) elif decision.state is ProfileEligibilityState.PENDING_REVIEW: if len(self._reviews) >= self._max_reviews: raise ContractViolation("shadow review items exceed max_review_items") self._reviews.append(ShadowReviewItem(decision)) def finish(self, summary: PbfParseSummary) -> None: if self._summary is not None or self._aborted: raise ContractViolation("shadow sink cannot be finalized twice") if not isinstance(summary, PbfParseSummary): raise ContractViolation("summary must be PbfParseSummary") if summary.counts.total != self._object_count: raise ContractViolation("parse summary count does not match evaluated objects") self._summary = summary def abort(self) -> None: self._leads.clear() self._reviews.clear() self._states.clear() self._categories.clear() self._seen.clear() self._object_count = 0 self._summary = None self._aborted = True def build_report(self) -> ShadowSourceReport: if self._aborted: raise ContractViolation("aborted shadow sink cannot build a report") if self._summary is None: raise ContractViolation("shadow sink must finish before building a report") return ShadowSourceReport( self._region_id, self._definition.profile, self._definition.definition_digest, self._summary.content_sha256, self._summary.size_bytes, self._summary.counts, self._summary.total_tags, tuple(self._states.items()), tuple(self._categories.items()), tuple(self._leads), tuple(self._reviews), ) def run_profile_shadow( *, approved_input: VerifiedPbfInput, parser: OsmiumPbfParser, evaluator: ProfileEvaluator, definition: ProfileDefinition, region_id: str, normalization_context: NormalizationContext, max_leads: int = 500_000, max_review_items: int = 100_000, verify_unique_objects: bool = True, ) -> ShadowSourceReport: """Parse a verified local PBF and return a deterministic source-shadow report.""" if not isinstance(parser, OsmiumPbfParser): raise ContractViolation("parser must be OsmiumPbfParser") sink = ProfileShadowSink( region_id=region_id, evaluator=evaluator, definition=definition, normalization_context=normalization_context, max_leads=max_leads, max_review_items=max_review_items, verify_unique_objects=verify_unique_objects, ) parser.parse(approved_input, sink) return sink.build_report() __all__ = [ "ProfileShadowSink", "ShadowLead", "ShadowReviewItem", "ShadowSourceReport", "run_profile_shadow", ]