feat: preserve gear components through catch imports

rf4_research/gear_components.py

@@ -0,0 +1,46 @@
+"""Ordered gear evidence extracted from source records without auto-mapping."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .community_sources import EquipmentItem
+
+
+@dataclass(frozen=True, slots=True)
+class GearComponentIdentity:
+    role: str
+    position: int
+    raw_value: str
+    source_external_id: str | None
+
+
+def _role_from_kind(kind: str) -> str:
+    value = kind.casefold()
+    terms = {
+        "удилищ": "rod", "катуш": "reel", "леск": "line", "крюч": "hook",
+        "поплав": "float", "груз": "sinker", "монтаж": "rig", "rig": "rig",
+        "приманк": "lure", "нажив": "bait", "bait": "bait", "lure": "lure",
+    }
+    return next((role for term, role in terms.items() if term in value), "other")
+
+
+def from_equipment(items: tuple[EquipmentItem, ...]) -> tuple[GearComponentIdentity, ...]:
+    """Keep source order and raw text; canonical mapping happens only after review."""
+    return tuple(
+        GearComponentIdentity(
+            role=_role_from_kind(item.kind), position=index,
+            raw_value=item.name, source_external_id=item.external_id,
+        )
+        for index, item in enumerate(items)
+        if item.name.strip()
+    )
+
+
+def from_catch_fields(*, bait: str | None, rig_type: str | None) -> tuple[GearComponentIdentity, ...]:
+    values: list[GearComponentIdentity] = []
+    if bait and bait.strip():
+        values.append(GearComponentIdentity("lure", len(values), bait.strip(), None))
+    if rig_type and rig_type.strip():
+        values.append(GearComponentIdentity("rig", len(values), rig_type.strip(), None))
+    return tuple(values)

rf4_research/gear_crosswalk.py

@@ -0,0 +1,86 @@
+"""Conservative offline crosswalk suggestions for gear identities."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .community_sources import GEAR_CATEGORIES
+from .media_assets import normalize_entity_label
+
+
+@dataclass(frozen=True, slots=True)
+class CanonicalGear:
+    key: str
+    name: str
+    category: str
+    aliases: tuple[str, ...] = ()
+    brand: str | None = None
+    family: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class GearIdentity:
+    source_system: str
+    external_id: str
+    name: str
+    category: str
+    brand: str | None = None
+    family: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class GearCrosswalkSuggestion:
+    source_system: str
+    external_id: str
+    external_name: str
+    category: str
+    status: str
+    canonical_keys: tuple[str, ...]
+
+
+def _compatible_category(value: str) -> bool:
+    return value.casefold().strip() in GEAR_CATEGORIES
+
+
+def suggest_gear_crosswalk(
+    canonical: list[CanonicalGear], identities: list[GearIdentity],
+) -> list[GearCrosswalkSuggestion]:
+    """Suggest only unique exact normalized-name matches in the same category.
+
+    Brand/family differences never create an automatic match. Ambiguous,
+    category-mismatched and unmatched identities retain no canonical key.
+    """
+    by_name: dict[str, set[str]] = {}
+    by_key = {item.key: item for item in canonical}
+    for item in canonical:
+        if not _compatible_category(item.category):
+            raise ValueError(f"invalid canonical gear category: {item.category!r}")
+        for name in (item.name, *item.aliases):
+            by_name.setdefault(normalize_entity_label(name), set()).add(item.key)
+    result: list[GearCrosswalkSuggestion] = []
+    for identity in identities:
+        keys = tuple(sorted(by_name.get(normalize_entity_label(identity.name), set())))
+        compatible = tuple(
+            key for key in keys
+            if by_key[key].category.casefold() == identity.category.casefold()
+            and not (by_key[key].brand and identity.brand and normalize_entity_label(by_key[key].brand) != normalize_entity_label(identity.brand))
+            and not (by_key[key].family and identity.family and normalize_entity_label(by_key[key].family) != normalize_entity_label(identity.family))
+        )
+        if not _compatible_category(identity.category):
+            status = "category_mismatch"
+            selected: tuple[str, ...] = ()
+        elif len(compatible) == 1:
+            status = "exact"
+            selected = compatible
+        elif len(compatible) > 1:
+            status = "ambiguous"
+            selected = ()
+        else:
+            status = "unmatched"
+            selected = ()
+        result.append(GearCrosswalkSuggestion(
+            source_system=identity.source_system, external_id=identity.external_id,
+            external_name=identity.name, category=identity.category,
+            status=status, canonical_keys=selected,
+        ))
+    return result