82 lines
3.3 KiB
Python
82 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from app.discovery.enums import MatchStatus
|
|
|
|
PRESENTATION_KEYS = {"title", "html", "presentation", "raw_title"}
|
|
|
|
|
|
def normalize_title(title: str) -> str:
|
|
return re.sub(r"\s+", " ", title.casefold().strip())
|
|
|
|
|
|
def spec_fingerprint(normalized_attributes: dict[str, Any], keys: tuple[str, ...] | None = None) -> str:
|
|
"""Fingerprint from whatever technical keys the observation actually has."""
|
|
use = keys if keys is not None else tuple(sorted(key for key in normalized_attributes if key not in PRESENTATION_KEYS))
|
|
parts = []
|
|
for key in use:
|
|
value = normalized_attributes.get(key)
|
|
parts.append(f"{key}={'' if value is None else str(value).casefold().strip()}")
|
|
return "|".join(parts)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IdentitySignals:
|
|
source_external_id: str | None
|
|
gtin_ean: str | None
|
|
manufacturer_sku: str | None
|
|
source_url: str | None
|
|
title: str
|
|
manufacturer: str | None
|
|
model: str | None
|
|
fingerprint: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IdentityDecision:
|
|
match_status: MatchStatus
|
|
existing_id: Any | None
|
|
reason: str
|
|
|
|
|
|
def decide_identity(incoming: IdentitySignals, known: list[tuple[Any, IdentitySignals]]) -> IdentityDecision:
|
|
"""Conservative matching. Similar titles never merge by themselves."""
|
|
for ref_id, existing in known:
|
|
if incoming.source_external_id and incoming.source_external_id == existing.source_external_id:
|
|
return IdentityDecision(MatchStatus.MATCHED, ref_id, "source_external_id")
|
|
if incoming.gtin_ean and incoming.gtin_ean == existing.gtin_ean:
|
|
return IdentityDecision(MatchStatus.MATCHED, ref_id, "gtin_ean")
|
|
if incoming.manufacturer_sku and incoming.manufacturer_sku == existing.manufacturer_sku:
|
|
return IdentityDecision(MatchStatus.MATCHED, ref_id, "manufacturer_sku")
|
|
if incoming.source_url and incoming.source_url == existing.source_url:
|
|
return IdentityDecision(MatchStatus.MATCHED, ref_id, "source_url")
|
|
|
|
title_hits = [
|
|
(ref_id, existing)
|
|
for ref_id, existing in known
|
|
if normalize_title(incoming.title) == normalize_title(existing.title)
|
|
]
|
|
if len(title_hits) == 1 and incoming.fingerprint and incoming.fingerprint == title_hits[0][1].fingerprint:
|
|
return IdentityDecision(MatchStatus.MATCHED, title_hits[0][0], "title+fingerprint")
|
|
|
|
manufacturer_model = [
|
|
(ref_id, existing)
|
|
for ref_id, existing in known
|
|
if incoming.manufacturer
|
|
and incoming.model
|
|
and incoming.manufacturer.casefold() == (existing.manufacturer or "").casefold()
|
|
and incoming.model.casefold() == (existing.model or "").casefold()
|
|
]
|
|
if len(manufacturer_model) == 1:
|
|
existing = manufacturer_model[0][1]
|
|
if incoming.fingerprint and existing.fingerprint and incoming.fingerprint != existing.fingerprint:
|
|
return IdentityDecision(MatchStatus.MATCH_UNCERTAIN, None, "manufacturer+model conflict with specs")
|
|
return IdentityDecision(MatchStatus.MATCHED, manufacturer_model[0][0], "manufacturer+model")
|
|
|
|
if title_hits:
|
|
return IdentityDecision(MatchStatus.MATCH_UNCERTAIN, None, "title only")
|
|
return IdentityDecision(MatchStatus.NEW_REFERENCE, None, "no match")
|