"""Deterministic, bounded, report-only existing-row adoption scanner.""" from __future__ import annotations from collections import defaultdict from collections.abc import Iterable from itertools import islice from ..domain.enums import AcceptedMappingOutcome, CandidateReviewOutcome from ..domain.models import OsmIdentity, RegionScopedIdentity from ..errors import ContractViolation from ..noco.models import NocoReadAudit, NocoReadPage, NocoSourceRecord, WorkflowProtectionEvidence from ..normalization.models import NormalizedBusiness from ..normalization.normalize import ( normalize_address, normalize_country_code, normalize_domain, normalize_name, normalize_phone, ) from ..ports.noco_read import NocoReadRepository from .models import ( AdoptionCandidate, AdoptionScanReport, AdoptionScanRequest, ExistingActiveMapping, NormalizedNocoBusiness, ) _WORKFLOW_FIELDS = ( "Enriched_Luxeillum", "Email_Found", "Blocked", "Enriched_Saas", "Blocked_SaaS", "Website_Found_By_Scraper", "website_last_checked_at", "website_enrichment_status", "website_enrichment_reason", "Website_Review_Candidates", "Website_Review_Summary", "Website_Review_Status", "Website_Review_Selected_URL", "Offer_URL", "Prospect_URL", ) _NON_OSM_IDENTITY_FIELDS = ( "place_id", "cid", "data_id", "maps_link", "foursquare_id", "foursquare_url", ) def _meaningful(value: object) -> bool: if value is None or value is False: return False if isinstance(value, str): return bool(value.strip()) and value.strip().casefold() not in {"false", "null", "0"} if isinstance(value, (int, float)): return value != 0 return True def _protection(record: NocoSourceRecord) -> WorkflowProtectionEvidence: reasons = {"source_baseline:absent"} for field_name in _WORKFLOW_FIELDS: if _meaningful(record.value(field_name)): reasons.add(f"workflow_field:{field_name}") for field_name in _NON_OSM_IDENTITY_FIELDS: if _meaningful(record.value(field_name)): reasons.add(f"non_osm_identity:{field_name}") source = record.text("source") if source is None: reasons.add("source:unknown") elif source.casefold() != "osm": reasons.add(f"source:non_osm:{source.casefold()}") return WorkflowProtectionEvidence( noco_id=record.noco_id, protected=True, reasons=tuple(reasons), ) def _normalize_noco( record: NocoSourceRecord, request: AdoptionScanRequest, ) -> NormalizedNocoBusiness: return NormalizedNocoBusiness( noco_id=record.noco_id, osm_identity=record.osm_identity(), name=normalize_name(record.text("name")), domain=normalize_domain(record.text("website")), phone=normalize_phone(record.text("phone"), request.normalization_context), address=normalize_address(record.text("address")), city=normalize_address(record.text("city")), country=( normalize_country_code(record.text("country_code")) or normalize_country_code(record.text("lead_country")) or request.normalization_context.country_code ), category=normalize_name(record.text("category")), protection=_protection(record), ) def _source_key(identity: RegionScopedIdentity) -> tuple[str, int]: return identity.osm_type.value, identity.osm_id def _matched_fields(source: NormalizedBusiness, row: NormalizedNocoBusiness) -> tuple[str, ...]: fields: list[str] = [] for field_name in ("name", "domain", "phone", "address", "city", "country"): left = getattr(source, field_name) right = getattr(row, field_name) if left is not None and left == right: fields.append(field_name) return tuple(fields) def _classify( source: NormalizedBusiness, row: NormalizedNocoBusiness, ) -> tuple[CandidateReviewOutcome, AcceptedMappingOutcome | None, str, int, tuple[str, ...]] | None: fields = _matched_fields(source, row) if row.osm_identity == source.identity.osm_identity: return ( CandidateReviewOutcome.ACCEPTED, AcceptedMappingOutcome.EXACT, "exact_osm_identity", 1000, ("osm_identity",) + fields, ) matched = set(fields) if {"domain", "phone", "country", "name"}.issubset(matched): return ( CandidateReviewOutcome.ACCEPTED, AcceptedMappingOutcome.HIGH_CONFIDENCE, "domain_phone_country_name", 980, fields, ) if {"domain", "name", "address"}.issubset(matched): return ( CandidateReviewOutcome.ACCEPTED, AcceptedMappingOutcome.HIGH_CONFIDENCE, "domain_name_address", 960, fields, ) if {"phone", "name", "address"}.issubset(matched): return ( CandidateReviewOutcome.ACCEPTED, AcceptedMappingOutcome.HIGH_CONFIDENCE, "phone_name_address", 950, fields, ) if len(fields) >= 2: return CandidateReviewOutcome.AMBIGUOUS, None, "multiple_partial_signals", 500, fields return None def scan_adoption_candidates( *, repository: NocoReadRepository, request: AdoptionScanRequest, sources: Iterable[NormalizedBusiness], existing_mappings: Iterable[ExistingActiveMapping] = (), ) -> AdoptionScanReport: """Read Noco rows once and emit immutable adoption evidence only.""" if not isinstance(repository, NocoReadRepository): raise ContractViolation("repository must implement NocoReadRepository") if not isinstance(request, AdoptionScanRequest): raise ContractViolation("request must be AdoptionScanRequest") schema = repository.read_schema() if schema.table_id != request.table_id: raise ContractViolation("Noco table id does not match the scan request") if schema.fingerprint != request.expected_schema_fingerprint: raise ContractViolation("Noco schema fingerprint mismatch") source_rows = tuple(islice(iter(sources), request.max_sources + 1)) if len(source_rows) > request.max_sources: raise ContractViolation("source count exceeds max_sources") if any(not isinstance(source, NormalizedBusiness) for source in source_rows): raise ContractViolation("sources must contain NormalizedBusiness values") if any(source.identity.region_id != request.region_id for source in source_rows): raise ContractViolation("all sources must belong to the requested region") source_by_key: dict[tuple[str, int], NormalizedBusiness] = {} for source in source_rows: key = _source_key(source.identity) if key in source_by_key: raise ContractViolation("source identities must be unique") source_by_key[key] = source mappings = tuple(islice(iter(existing_mappings), request.max_existing_mappings + 1)) if len(mappings) > request.max_existing_mappings: raise ContractViolation("existing mapping count exceeds max_existing_mappings") if any(not isinstance(mapping, ExistingActiveMapping) for mapping in mappings): raise ContractViolation("existing_mappings must contain ExistingActiveMapping values") mapping_by_noco: dict[int, ExistingActiveMapping] = {} mapping_by_source: dict[tuple[str, int], ExistingActiveMapping] = {} for mapping in mappings: if mapping.identity.region_id != request.region_id: raise ContractViolation("existing mappings must belong to the requested region") source_key = _source_key(mapping.identity) if mapping.noco_id in mapping_by_noco or source_key in mapping_by_source: raise ContractViolation("existing active mappings violate uniqueness") mapping_by_noco[mapping.noco_id] = mapping mapping_by_source[source_key] = mapping exact_index = source_by_key domain_index: dict[str, set[tuple[str, int]]] = defaultdict(set) phone_index: dict[str, set[tuple[str, int]]] = defaultdict(set) name_index: dict[str, set[tuple[str, int]]] = defaultdict(set) for key, source in source_by_key.items(): if source.domain: domain_index[source.domain].add(key) if source.phone: phone_index[source.phone].add(key) if source.name: name_index[source.name].add(key) candidates_by_source: dict[tuple[str, int], list[AdoptionCandidate]] = defaultdict(list) sources_by_noco: dict[int, set[tuple[str, int]]] = defaultdict(set) rows_read = 0 pages_read = 0 candidate_count = 0 max_noco_id: int | None = None previous_noco_id = request.read_request.start_after_id terminal_page_seen = False maximum_pages = ( request.read_request.max_rows + request.read_request.page_size - 1 ) // request.read_request.page_size for page in repository.iter_pages(request.read_request): if not isinstance(page, NocoReadPage): raise ContractViolation("repository pages must be NocoReadPage values") if terminal_page_seen: raise ContractViolation("repository yielded rows after a terminal page") pages_read += 1 if pages_read > maximum_pages: raise ContractViolation("repository exceeded the bounded page count") if not page.records: raise ContractViolation("repository yielded an empty page") remaining_rows = request.read_request.max_rows - rows_read expected_limit = min(request.read_request.page_size, remaining_rows) if len(page.records) > expected_limit: raise ContractViolation("repository exceeded the bounded row count") if page.records[0].noco_id <= previous_noco_id: raise ContractViolation("repository violated cross-page keyset ordering") if page.next_after_id is not None and len(page.records) != expected_limit: raise ContractViolation("nonterminal repository page must fill its bounded limit") for raw_row in page.records: rows_read += 1 max_noco_id = raw_row.noco_id row = _normalize_noco(raw_row, request) possible: set[tuple[str, int]] = set() if isinstance(row.osm_identity, OsmIdentity): possible.add((row.osm_identity.osm_type.value, row.osm_identity.osm_id)) if row.domain: possible.update(domain_index.get(row.domain, ())) if row.phone: possible.update(phone_index.get(row.phone, ())) if row.name: possible.update(name_index.get(row.name, ())) if len(possible) > request.max_sources_per_noco_row: raise ContractViolation("Noco row exceeds max_sources_per_noco_row") for source_key in sorted(possible): candidate_source = exact_index.get(source_key) if candidate_source is None: continue classified = _classify(candidate_source, row) if classified is None: continue status, proposed, rule, confidence, fields = classified collision_reasons: set[str] = set() mapped_noco = mapping_by_noco.get(row.noco_id) if mapped_noco and _source_key(mapped_noco.identity) != source_key: collision_reasons.add("noco_row_already_maps_to_other_source") mapped_source = mapping_by_source.get(source_key) if mapped_source and mapped_source.noco_id != row.noco_id: collision_reasons.add("source_already_maps_to_other_noco_row") candidate = AdoptionCandidate( source_identity=candidate_source.identity, noco_id=row.noco_id, result_status=( CandidateReviewOutcome.COLLISION if collision_reasons else status ), proposed_outcome=None if collision_reasons else proposed, rule=rule, confidence_milli=confidence, matched_fields=fields, protection=row.protection, collision_reasons=tuple(collision_reasons), ) bucket = candidates_by_source[source_key] if len(bucket) >= request.max_candidates_per_source: raise ContractViolation("candidate count exceeds max_candidates_per_source") bucket.append(candidate) candidate_count += 1 sources_by_noco[row.noco_id].add(source_key) if candidate_count > request.max_candidates: raise ContractViolation("candidate count exceeds max_candidates") previous_noco_id = page.records[-1].noco_id terminal_page_seen = page.next_after_id is None if pages_read and not terminal_page_seen: raise ContractViolation("repository ended before its declared continuation") output: list[AdoptionCandidate] = [] for source_key, source in sorted(source_by_key.items()): bucket = candidates_by_source.get(source_key, []) if not bucket: output.append( AdoptionCandidate( source_identity=source.identity, noco_id=None, result_status=CandidateReviewOutcome.UNMATCHED, proposed_outcome=None, rule="no_reliable_match", confidence_milli=0, matched_fields=(), protection=None, ) ) continue accepted = [ item for item in bucket if item.result_status is CandidateReviewOutcome.ACCEPTED ] ambiguous = [ item for item in bucket if item.result_status is CandidateReviewOutcome.AMBIGUOUS ] collision_candidates = [ item for item in bucket if item.result_status is CandidateReviewOutcome.COLLISION ] cross_source_rows = { item.noco_id for item in bucket if item.noco_id is not None and len(sources_by_noco[item.noco_id]) > 1 } if len(accepted) > 1 or cross_source_rows: for item in bucket: reasons = set(item.collision_reasons) if len(accepted) > 1: reasons.add("multiple_accepted_noco_matches") if item.noco_id in cross_source_rows: reasons.add("noco_row_matches_multiple_sources") output.append( AdoptionCandidate( source_identity=item.source_identity, noco_id=item.noco_id, result_status=CandidateReviewOutcome.COLLISION, proposed_outcome=None, rule=item.rule, confidence_milli=item.confidence_milli, matched_fields=item.matched_fields, protection=item.protection, collision_reasons=tuple(reasons), ) ) elif accepted: output.append(accepted[0]) output.extend(collision_candidates) elif collision_candidates: output.extend(collision_candidates) else: output.extend(ambiguous) if len(output) > request.max_candidates: raise ContractViolation("report candidate count exceeds max_candidates") audit = NocoReadAudit( schema_fingerprint=schema.fingerprint, rows_read=rows_read, pages_read=pages_read, max_noco_id=max_noco_id, repository_kind=type(repository).__name__, ) report = AdoptionScanReport( run_id=request.run_id, region_id=request.region_id, table_id=request.table_id, schema_fingerprint=schema.fingerprint, candidates=tuple(output), read_audit=audit, ) return report