John Doe
a35bcd6647
- OFF incremental fetch via search API + persistent watermark (ingest_state, migration 0004) - GS1 barcode supplement adapter (offline mapping + GS1-style API) filling only gaps with field-level provenance - Non-GTIN dedup with canonical selection + merge_log; field-level conflict resolution (source trust > recency) - Quality scoring (0.4 completeness + 0.3 source trust + 0.2 multi-source + 0.1 freshness) wired into load/merge - Jobs: update_off, dedup, schedule; docs/ingestion-management.md - 19 new tests (pure + DB-integration), ruff clean Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""Field-level conflict resolution for multi-source records.
|
|
|
|
When more than one source describes the same product, each field may have
|
|
several candidate values. We pick a winner per field by source trust first,
|
|
then recency, ignoring empty values, and keep a provenance trail of which
|
|
source won each field.
|
|
|
|
These are pure functions (no DB / no network) so they are easy to unit-test;
|
|
the DB-level record merge lives in :mod:`opengoods.etl.dedup`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Candidate:
|
|
"""One source's proposed value for a field."""
|
|
|
|
value: object
|
|
source: str
|
|
trust: float = 0.5
|
|
fetched_at: datetime | None = None
|
|
|
|
|
|
@dataclass
|
|
class FieldResolution:
|
|
"""The winning value for a field plus the source it came from."""
|
|
|
|
value: object
|
|
source: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class MergedRecord:
|
|
"""A merged record with per-field provenance (field name -> source)."""
|
|
|
|
values: dict[str, object] = field(default_factory=dict)
|
|
provenance: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
def _is_empty(value: object) -> bool:
|
|
if value is None:
|
|
return True
|
|
if isinstance(value, str):
|
|
return value.strip() == ""
|
|
if isinstance(value, (list, dict, tuple, set)):
|
|
return len(value) == 0
|
|
return False
|
|
|
|
|
|
def _sort_key(c: Candidate) -> tuple[float, float]:
|
|
ts = c.fetched_at.timestamp() if c.fetched_at is not None else float("-inf")
|
|
return (c.trust, ts)
|
|
|
|
|
|
def resolve_field(candidates: list[Candidate]) -> FieldResolution | None:
|
|
"""Pick the best non-empty candidate for one field.
|
|
|
|
Ranking: highest source trust, then most recent ``fetched_at``. Returns
|
|
``None`` when there is no usable (non-empty) candidate.
|
|
"""
|
|
usable = [c for c in candidates if not _is_empty(c.value)]
|
|
if not usable:
|
|
return None
|
|
winner = max(usable, key=_sort_key)
|
|
return FieldResolution(value=winner.value, source=winner.source)
|
|
|
|
|
|
def merge_records(records: list[dict], *, fields: list[str] | None = None) -> MergedRecord:
|
|
"""Merge several ``{field: Candidate|value}`` records into one.
|
|
|
|
Each input record maps field name -> :class:`Candidate` (preferred) or a
|
|
bare value (treated as trust 0.5, no timestamp). The result keeps, for each
|
|
field, the winning value and the name of the source that supplied it.
|
|
"""
|
|
keys: list[str]
|
|
if fields is not None:
|
|
keys = list(fields)
|
|
else:
|
|
seen: dict[str, None] = {}
|
|
for rec in records:
|
|
for k in rec:
|
|
seen.setdefault(k, None)
|
|
keys = list(seen)
|
|
|
|
merged = MergedRecord()
|
|
for key in keys:
|
|
candidates: list[Candidate] = []
|
|
for rec in records:
|
|
if key not in rec:
|
|
continue
|
|
cand = rec[key]
|
|
if not isinstance(cand, Candidate):
|
|
cand = Candidate(value=cand, source="unknown")
|
|
candidates.append(cand)
|
|
resolution = resolve_field(candidates)
|
|
if resolution is not None:
|
|
merged.values[key] = resolution.value
|
|
if resolution.source is not None:
|
|
merged.provenance[key] = resolution.source
|
|
return merged
|