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>
54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
from pathlib import Path
|
|
|
|
from opengoods.adapters.gs1 import GS1Adapter
|
|
from opengoods.etl.supplement import apply_supplement, ensure_gs1_source
|
|
|
|
MAPPING = Path(__file__).parent / "fixtures" / "gs1_mapping.json"
|
|
GTIN = "06901234567892"
|
|
|
|
|
|
def test_gs1_adapter_offline_lookup():
|
|
adapter = GS1Adapter.from_file(MAPPING)
|
|
rec = adapter.fetch_barcode(GTIN)
|
|
assert rec["brand"] == "示例品牌"
|
|
assert rec["net_content_value"] == 550
|
|
assert rec["net_content_unit"] == "ml"
|
|
# An entry that only has empty values yields no supplement.
|
|
assert adapter.fetch_barcode("00000000000000") is None
|
|
# Unknown barcode -> None.
|
|
assert adapter.fetch_barcode("99999999999999") is None
|
|
|
|
|
|
def test_gs1_supplement_fills_only_gaps(db_conn):
|
|
pid = db_conn.execute(
|
|
"INSERT INTO product (gtin, name) VALUES (%s, %s) RETURNING id", (GTIN, "水")
|
|
).fetchone()[0]
|
|
|
|
adapter = GS1Adapter.from_file(MAPPING)
|
|
rec = adapter.fetch_barcode(GTIN)
|
|
source_id = ensure_gs1_source(db_conn)
|
|
|
|
filled = apply_supplement(db_conn, rec, source_id)
|
|
assert {"brand", "country_of_origin", "net_content"} <= set(filled)
|
|
|
|
brand_id, country, net_value, net_unit = db_conn.execute(
|
|
"""
|
|
SELECT brand_id, country_of_origin, net_content_value, net_content_unit
|
|
FROM product WHERE id = %s
|
|
""",
|
|
(pid,),
|
|
).fetchone()
|
|
assert brand_id is not None
|
|
assert country == "China"
|
|
assert float(net_value) == 550.0
|
|
assert net_unit == "ml"
|
|
|
|
fields = db_conn.execute(
|
|
"SELECT fields FROM product_source WHERE product_id = %s AND source_id = %s",
|
|
(pid, source_id),
|
|
).fetchone()[0]
|
|
assert "brand" in fields
|
|
|
|
# Re-applying does nothing because the gaps are now filled.
|
|
assert apply_supplement(db_conn, rec, source_id) == []
|