"""Validated immutable inputs for the controlled insert-only writer.""" from __future__ import annotations import hashlib import json import re from dataclasses import dataclass from datetime import datetime from types import MappingProxyType from typing import Mapping from ..errors import ContractViolation _SHA256 = re.compile(r"^[0-9a-f]{64}$") _OSM_TYPES = frozenset({"node", "way", "relation"}) _MAX_TEXT = 4096 def _mapping(value: object, field: str) -> Mapping[str, object]: if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ContractViolation(f"{field} must be an object with string keys") return value def _text( value: object, field: str, *, required: bool = True, maximum: int = _MAX_TEXT, ) -> str | None: if value is None and not required: return None if not isinstance(value, str): raise ContractViolation(f"{field} must be a string") result = value.strip() if required and not result: raise ContractViolation(f"{field} must be nonblank") if len(result) > maximum: raise ContractViolation(f"{field} exceeds the bounded length") return result or None def _integer(value: object, field: str, *, positive: bool = False) -> int: if isinstance(value, bool) or not isinstance(value, int): raise ContractViolation(f"{field} must be an integer") if positive and value <= 0: raise ContractViolation(f"{field} must be positive") return value def _digest(value: object, field: str) -> str: result = _text(value, field, maximum=64) if result is None or not _SHA256.fullmatch(result): raise ContractViolation(f"{field} must be lowercase SHA-256") return result def _optional_text(value: object, field: str, maximum: int = _MAX_TEXT) -> str | None: return _text(value, field, required=False, maximum=maximum) def _iso_timestamp(value: str) -> str: normalized = value.strip() try: parsed = datetime.fromisoformat(normalized.replace("Z", "+00:00")) except ValueError as exc: raise ContractViolation("run_timestamp must be ISO-8601") from exc if parsed.tzinfo is None: raise ContractViolation("run_timestamp must include a timezone") return normalized @dataclass(frozen=True, slots=True) class InsertCandidate: """One validated source lead eligible for an insert-only attempt.""" region_id: str profile_id: str profile_version: str report_hash: str source_sha256: str osm_type: str osm_id: int identity_hash: str source_tags_hash: str name: str category: str subcategory: str | None domain: str | None phone: str | None address: str | None city: str | None state: str | None postcode: str | None country: str latitude_e7: int longitude_e7: int def __post_init__(self) -> None: for field in ("region_id", "profile_id", "profile_version", "name", "category", "country"): value = _text(getattr(self, field), field, maximum=512) object.__setattr__(self, field, value) object.__setattr__(self, "report_hash", _digest(self.report_hash, "report_hash")) object.__setattr__(self, "source_sha256", _digest(self.source_sha256, "source_sha256")) object.__setattr__(self, "identity_hash", _digest(self.identity_hash, "identity_hash")) object.__setattr__( self, "source_tags_hash", _digest(self.source_tags_hash, "source_tags_hash"), ) object.__setattr__(self, "osm_type", _text(self.osm_type, "osm_type", maximum=16)) if self.osm_type not in _OSM_TYPES: raise ContractViolation("osm_type is unsupported") object.__setattr__(self, "osm_id", _integer(self.osm_id, "osm_id", positive=True)) for field in ("subcategory", "domain", "phone", "address", "city", "state", "postcode"): object.__setattr__(self, field, _optional_text(getattr(self, field), field)) latitude = _integer(self.latitude_e7, "latitude_e7") longitude = _integer(self.longitude_e7, "longitude_e7") if not -900_000_000 <= latitude <= 900_000_000: raise ContractViolation("latitude_e7 is outside the valid range") if not -1_800_000_000 <= longitude <= 1_800_000_000: raise ContractViolation("longitude_e7 is outside the valid range") @property def osm_url(self) -> str: return f"https://www.openstreetmap.org/{self.osm_type}/{self.osm_id}" @property def website(self) -> str | None: return None if self.domain is None else f"https://{self.domain}" @property def idempotency_key(self) -> str: material = ( f"insert:{self.report_hash}:{self.region_id}:{self.profile_id}:" f"{self.profile_version}:{self.osm_type}:{self.osm_id}" ) return hashlib.sha256(material.encode()).hexdigest() @property def payload_hash(self) -> str: payload = { "address": self.address, "category": self.category, "city": self.city, "country": self.country, "domain": self.domain, "identity_hash": self.identity_hash, "latitude_e7": self.latitude_e7, "longitude_e7": self.longitude_e7, "name": self.name, "osm_id": self.osm_id, "osm_type": self.osm_type, "phone": self.phone, "postcode": self.postcode, "profile_id": self.profile_id, "profile_version": self.profile_version, "region_id": self.region_id, "source_tags_hash": self.source_tags_hash, "state": self.state, "subcategory": self.subcategory, } canonical = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode() return hashlib.sha256(canonical).hexdigest() def noco_payload( self, *, run_id: str, run_timestamp: str, include_website: bool = True, ) -> Mapping[str, object]: timestamp = _iso_timestamp(run_timestamp) identifier = _text(run_id, "run_id", maximum=256) if identifier is None: raise ContractViolation("run_id must be nonblank") payload: dict[str, object] = { "address": self.address or "", "category": self.category, "city": self.city or "", "country": self.country, "country_code": self.country, "country_name": self.country, "industry_slug": self.profile_id, "job_id": identifier, "job_status": "completed", "lead_country": self.country, "name": self.name, "osm_id": str(self.osm_id), "osm_type": self.osm_type, "osm_url": self.osm_url, "phone": self.phone or "", "postcode": self.postcode or "", "query_display_name": self.profile_id, "query_name": self.profile_id, "scraped_at": timestamp, "selector_id": f"{self.profile_id}:{self.profile_version}", "source": "OSM", "state_region": self.state or "", "subcategory": self.subcategory or "", "website": self.website if include_website and self.website else "", } return MappingProxyType(payload) @classmethod def from_mapping( cls, *, lead: Mapping[str, object], region_id: str, profile_id: str, profile_version: str, report_hash: str, source_sha256: str, ) -> "InsertCandidate": return cls( region_id=region_id, profile_id=profile_id, profile_version=profile_version, report_hash=report_hash, source_sha256=source_sha256, osm_type=str(lead.get("osm_type", "")), osm_id=_integer(lead.get("osm_id"), "osm_id", positive=True), identity_hash=_digest(lead.get("identity_hash"), "identity_hash"), source_tags_hash=_digest(lead.get("source_tags_hash"), "source_tags_hash"), name=str(lead.get("name", "")), category=str(lead.get("category", "")), subcategory=_optional_text(lead.get("subcategory"), "subcategory"), domain=_optional_text(lead.get("domain"), "domain"), phone=_optional_text(lead.get("phone"), "phone"), address=_optional_text(lead.get("address"), "address"), city=_optional_text(lead.get("city"), "city"), state=_optional_text(lead.get("state"), "state"), postcode=_optional_text(lead.get("postcode"), "postcode"), country=str(lead.get("country", "")), latitude_e7=_integer(lead.get("latitude_e7"), "latitude_e7"), longitude_e7=_integer(lead.get("longitude_e7"), "longitude_e7"), ) @dataclass(frozen=True, slots=True) class ShadowReportDocument: """A canonical Phase 2A.6 report accepted as Phase 2B input.""" region_id: str profile_id: str profile_version: str source_sha256: str report_hash: str candidates: tuple[InsertCandidate, ...] @classmethod def from_json( cls, content: str, *, expected_report_hash: str, max_candidates: int, ) -> "ShadowReportDocument": expected = _digest(expected_report_hash, "expected_report_hash") if isinstance(max_candidates, bool) or not 1 <= max_candidates <= 1_000_000: raise ContractViolation("max_candidates exceeds the bounded range") try: parsed: object = json.loads(content) except json.JSONDecodeError as exc: raise ContractViolation("shadow report is not valid JSON") from exc root = _mapping(parsed, "shadow report") embedded = _digest(root.get("report_hash"), "report_hash") without_hash = dict(root) without_hash.pop("report_hash", None) canonical = json.dumps( without_hash, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode() calculated = hashlib.sha256(canonical).hexdigest() if embedded != calculated or expected != calculated: raise ContractViolation("shadow report hash does not match canonical content") region_id = _text(root.get("region_id"), "region_id", maximum=512) source_sha256 = _digest(root.get("source_sha256"), "source_sha256") profile = _mapping(root.get("profile"), "profile") profile_id = _text(profile.get("id"), "profile.id", maximum=512) profile_version = _text(profile.get("version"), "profile.version", maximum=512) raw_leads = root.get("leads") if not isinstance(raw_leads, list): raise ContractViolation("leads must be an array") if len(raw_leads) > max_candidates: raise ContractViolation("shadow report leads exceed max_candidates") if region_id is None or profile_id is None or profile_version is None: raise ContractViolation("shadow report identity fields must be nonblank") candidates = tuple( InsertCandidate.from_mapping( lead=_mapping(item, "lead"), region_id=region_id, profile_id=profile_id, profile_version=profile_version, report_hash=calculated, source_sha256=source_sha256, ) for item in raw_leads ) keys = {(item.osm_type, item.osm_id) for item in candidates} if len(keys) != len(candidates): raise ContractViolation("shadow report contains duplicate candidate identities") ordered = tuple(sorted(candidates, key=lambda item: (item.osm_type, item.osm_id))) return cls( region_id=region_id, profile_id=profile_id, profile_version=profile_version, source_sha256=source_sha256, report_hash=calculated, candidates=ordered, ) __all__ = ["InsertCandidate", "ShadowReportDocument"]