87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
"""Conservative, offline crosswalk suggestions for waterbody identities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Iterable, Protocol
|
|
|
|
from .media_assets import normalize_entity_label
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CanonicalWaterbody:
|
|
key: str
|
|
name: str
|
|
aliases: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class WaterbodyIdentity:
|
|
source_system: str
|
|
external_id: str
|
|
name: str
|
|
|
|
|
|
class WaterbodyCandidate(Protocol):
|
|
"""Minimum shape shared by parsed secondary-source candidates."""
|
|
|
|
source_system: str
|
|
source_external_id: str
|
|
name: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CrosswalkSuggestion:
|
|
source_system: str
|
|
external_id: str
|
|
external_name: str
|
|
status: str
|
|
canonical_keys: tuple[str, ...]
|
|
|
|
|
|
def identities_from_candidates(
|
|
candidates: Iterable[WaterbodyCandidate],
|
|
) -> list[WaterbodyIdentity]:
|
|
"""Convert parser rows without selecting a canonical key.
|
|
|
|
Metrics, URLs, and other source facts stay outside the crosswalk
|
|
decision and must not influence automatic canonical matching.
|
|
"""
|
|
return [
|
|
WaterbodyIdentity(
|
|
source_system=row.source_system,
|
|
external_id=row.source_external_id,
|
|
name=row.name,
|
|
)
|
|
for row in candidates
|
|
]
|
|
|
|
|
|
def suggest_waterbody_crosswalk(
|
|
canonical: list[CanonicalWaterbody], identities: list[WaterbodyIdentity],
|
|
) -> list[CrosswalkSuggestion]:
|
|
"""Suggest only unique exact normalized-name matches.
|
|
|
|
``ambiguous`` and ``unmatched`` rows intentionally have no selected key;
|
|
callers must not turn this report into aliases without human review.
|
|
"""
|
|
by_name: dict[str, set[str]] = {}
|
|
for item in canonical:
|
|
for name in (item.name, *item.aliases):
|
|
by_name.setdefault(normalize_entity_label(name), set()).add(item.key)
|
|
result: list[CrosswalkSuggestion] = []
|
|
for identity in identities:
|
|
keys = tuple(sorted(by_name.get(normalize_entity_label(identity.name), set())))
|
|
status = "exact" if len(keys) == 1 else "ambiguous" if keys else "unmatched"
|
|
result.append(CrosswalkSuggestion(
|
|
source_system=identity.source_system, external_id=identity.external_id,
|
|
external_name=identity.name, status=status,
|
|
canonical_keys=keys if status == "exact" else (),
|
|
))
|
|
return result
|
|
|
|
|
|
def missing_external_ids(expected: set[str], observed: set[str]) -> tuple[str, ...]:
|
|
"""Return stable missing IDs without interpreting absence as deletion."""
|
|
return tuple(sorted(expected - observed))
|