87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""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
|