sync

rf4_research/waterbody_crosswalk.py

@@ -0,0 +1,59 @@
+"""Conservative, offline crosswalk suggestions for waterbody identities."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+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
+
+
+@dataclass(frozen=True, slots=True)
+class CrosswalkSuggestion:
+    source_system: str
+    external_id: str
+    external_name: str
+    status: str
+    canonical_keys: tuple[str, ...]
+
+
+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))