From f04da0a135778774aa28a1cc3dd827be79b8fec9 Mon Sep 17 00:00:00 2001 From: novaalphastrikeomegaz663 Date: Wed, 24 Jun 2026 02:33:45 +0000 Subject: [PATCH] feat(ingestion): import bypos-collector JSONL into goods Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- ingestion/opengoods/adapters/bypos.py | 116 +++++++++++++++++ ingestion/opengoods/etl/load.py | 155 +++++++++++++++++++++++ ingestion/opengoods/jobs/import_bypos.py | 77 +++++++++++ ingestion/tests/test_bypos.py | 155 +++++++++++++++++++++++ tools/bypos-collector/README.md | 15 +++ 5 files changed, 518 insertions(+) create mode 100644 ingestion/opengoods/adapters/bypos.py create mode 100644 ingestion/opengoods/jobs/import_bypos.py create mode 100644 ingestion/tests/test_bypos.py diff --git a/ingestion/opengoods/adapters/bypos.py b/ingestion/opengoods/adapters/bypos.py new file mode 100644 index 0000000..414f214 --- /dev/null +++ b/ingestion/opengoods/adapters/bypos.py @@ -0,0 +1,116 @@ +"""Adapter for the bypos-collector output (central product library zc.bypos.net). + +The ``bypos-collector`` tool (see ``tools/bypos-collector``) queries the same +central product library that the 云店 POS uses when adding a product by barcode, +and writes one JSON object per line (JSONL) with these fields:: + + barcode name spec unit area manufacturer license + in_price sell_price status retmsg fetched_at source + +This module turns one such record into the internal product shape consumed by +:func:`opengoods.etl.load.load_bypos_record`. It is a pure function (no DB, no +network) so it is easy to unit-test. + +The central library is a Chinese retail catalogue: it provides 品名/规格/单位/ +产地/厂商/建议价 but **not** ingredients or nutrition, so no ``food`` block is +produced. +""" + +from __future__ import annotations + +from decimal import Decimal, InvalidOperation + +from opengoods.etl.transform import _clamp, is_valid_gtin, parse_quantity +from opengoods.units import UnitError, normalize + +SOURCE_NAME = "bypos中心库" +SOURCE_HOMEPAGE = "https://zc.bypos.net" +# Vendor catalogue data; not an open-licensed dataset. Display facts only. +SOURCE_LICENSE = "proprietary" +SOURCE_TRUST = 0.6 + + +def _to_price(text: str | None) -> Decimal | None: + """Parse a price string to a positive Decimal, or None for empty/zero.""" + if not text: + return None + try: + amount = Decimal(str(text).strip()) + except (InvalidOperation, ValueError): + return None + if amount <= 0: + return None + return amount + + +def _net_content(spec: str | None) -> tuple[Decimal | None, str | None, Decimal | None]: + """Best-effort parse a spec like '500mL'/'5kg' to (value, unit, canonical). + + Packaging-style specs ('20支', '1X24', '盒') have no mass/volume unit and + yield ``(None, None, None)`` — the raw spec is kept in attributes instead. + """ + parsed = parse_quantity(spec or "") + if not parsed: + return None, None, None + value, unit = parsed + try: + norm = normalize(value, unit) + except UnitError: + return None, None, None + return norm.value, norm.unit, norm.canonical_value + + +def transform_bypos(rec: dict) -> dict | None: + """Transform one bypos-collector JSONL record into an internal product dict. + + Returns ``None`` for non-hit rows or rows without a usable name. + """ + if rec.get("status") != "hit": + return None + name = (rec.get("name") or "").strip() + if not name: + return None + + barcode = str(rec.get("barcode") or "").strip() + gtin = barcode if barcode and is_valid_gtin(barcode) else None + + net_value, net_unit, net_canonical = _net_content(rec.get("spec")) + + spec = (rec.get("spec") or "").strip() + pack_unit = (rec.get("unit") or "").strip() + area = (rec.get("area") or "").strip() + manufacturer = (rec.get("manufacturer") or "").strip() or None + license_no = (rec.get("license") or "").strip() + in_price = _to_price(rec.get("in_price")) + sell_price = _to_price(rec.get("sell_price")) + + attributes: dict[str, object] = {} + if spec: + attributes["spec"] = spec + if pack_unit: + attributes["pack_unit"] = pack_unit + if area: + attributes["origin_area"] = area + if license_no: + attributes["production_license"] = license_no + if in_price is not None: + attributes["suggested_in_price"] = float(in_price) + if sell_price is not None: + attributes["suggested_retail_price"] = float(sell_price) + + # Domestic GS1-China barcodes (69x) are China-made; area is a province/city, + # kept separately in attributes.origin_area. + country = "中国" if gtin and gtin.startswith("69") else None + + return { + "gtin": gtin, + "name": name, + "manufacturer": manufacturer, + "net_content_value": net_value, + "net_content_unit": _clamp(net_unit, 16), + "net_content_canonical": net_canonical, + "country_of_origin": country, + "attributes": attributes, + "msrp": sell_price, + "fetched_at": (rec.get("fetched_at") or "").strip() or None, + } diff --git a/ingestion/opengoods/etl/load.py b/ingestion/opengoods/etl/load.py index f90d5d0..0c54079 100644 --- a/ingestion/opengoods/etl/load.py +++ b/ingestion/opengoods/etl/load.py @@ -14,6 +14,10 @@ from typing import Any import psycopg from psycopg.types.json import Jsonb +from opengoods.adapters.bypos import SOURCE_HOMEPAGE as SOURCE_HOMEPAGE_BYPOS +from opengoods.adapters.bypos import SOURCE_LICENSE as SOURCE_LICENSE_BYPOS +from opengoods.adapters.bypos import SOURCE_NAME as SOURCE_NAME_BYPOS +from opengoods.adapters.bypos import SOURCE_TRUST as SOURCE_TRUST_BYPOS from opengoods.adapters.openfoodfacts import OFF_LICENSE, SOURCE_NAME from opengoods.etl.quality import update_quality @@ -73,6 +77,23 @@ def _ensure_brand(conn: psycopg.Connection, name: str | None) -> str | None: return row[0] +def _ensure_manufacturer( + conn: psycopg.Connection, name: str | None, country: str | None = None +) -> str | None: + if not name: + return None + row = conn.execute( + """ + INSERT INTO manufacturer (name, normalized_name, country) + VALUES (%s, %s, %s) + ON CONFLICT (normalized_name) DO UPDATE SET name = manufacturer.name + RETURNING id + """, + (name, _normalize_brand(name), country), + ).fetchone() + return row[0] + + def _category_id(conn: psycopg.Connection, path: str | None) -> tuple[str | None, str | None]: if not path: return None, None @@ -218,6 +239,140 @@ def load_record_safe( return False +def ensure_bypos_source(conn: psycopg.Connection) -> str: + """Upsert the bypos central-library source row and return its id.""" + return ensure_source_named( + conn, + SOURCE_NAME_BYPOS, + SOURCE_HOMEPAGE_BYPOS, + SOURCE_LICENSE_BYPOS, + SOURCE_TRUST_BYPOS, + ) + + +def load_bypos_record( + conn: psycopg.Connection, rec: dict[str, Any], source_id: str, raw: dict +) -> str: + """Upsert one transformed bypos record; return the product id. + + Unlike OFF records these have no ingredients/nutrition, so no ``food_detail`` + row is written. The suggested retail price (if any) is stored as a CNY MSRP + snapshot, and provenance/MSRP rows are keyed by source so a re-import + refreshes rather than duplicates them. + """ + manufacturer_id = _ensure_manufacturer( + conn, rec.get("manufacturer"), rec.get("country_of_origin") + ) + attrs = rec.get("attributes") or {} + fields = ["name", "net_content", "country_of_origin"] + if manufacturer_id: + fields.append("manufacturer") + if attrs: + fields.append("attributes") + + if rec.get("gtin"): + fields.append("gtin") + prod = conn.execute( + """ + INSERT INTO product (gtin, name, manufacturer_id, + net_content_value, net_content_unit, net_content_canonical, + country_of_origin, attributes) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (gtin) WHERE gtin IS NOT NULL DO UPDATE SET + name = EXCLUDED.name, + manufacturer_id = COALESCE(EXCLUDED.manufacturer_id, product.manufacturer_id), + net_content_value = COALESCE(EXCLUDED.net_content_value, product.net_content_value), + net_content_unit = COALESCE(EXCLUDED.net_content_unit, product.net_content_unit), + net_content_canonical = COALESCE( + EXCLUDED.net_content_canonical, product.net_content_canonical), + country_of_origin = COALESCE(EXCLUDED.country_of_origin, product.country_of_origin), + attributes = product.attributes || EXCLUDED.attributes + RETURNING id + """, + ( + rec["gtin"], + rec["name"], + manufacturer_id, + rec.get("net_content_value"), + rec.get("net_content_unit"), + rec.get("net_content_canonical"), + rec.get("country_of_origin"), + Jsonb(attrs), + ), + ).fetchone() + else: + prod = conn.execute( + """ + INSERT INTO product (name, manufacturer_id, + net_content_value, net_content_unit, net_content_canonical, + country_of_origin, attributes) + VALUES (%s,%s,%s,%s,%s,%s,%s) + RETURNING id + """, + ( + rec["name"], + manufacturer_id, + rec.get("net_content_value"), + rec.get("net_content_unit"), + rec.get("net_content_canonical"), + rec.get("country_of_origin"), + Jsonb(attrs), + ), + ).fetchone() + product_id = prod[0] + + # Refresh this source's MSRP snapshot (suggested retail price, CNY). + conn.execute( + "DELETE FROM product_msrp WHERE product_id = %s AND source_id = %s", + (product_id, source_id), + ) + if rec.get("msrp") is not None: + conn.execute( + """ + INSERT INTO product_msrp (product_id, amount, currency, region, source_id, source_url) + VALUES (%s,%s,'CNY','CN',%s,%s) + """, + (product_id, rec["msrp"], source_id, SOURCE_HOMEPAGE_BYPOS), + ) + fields.append("msrp") + + # Refresh this source's provenance row (one per source for idempotency). + conn.execute( + "DELETE FROM product_source WHERE product_id = %s AND source_id = %s", + (product_id, source_id), + ) + conn.execute( + """ + INSERT INTO product_source (product_id, source_id, url, fields, fetched_at, raw) + VALUES (%s,%s,%s,%s, COALESCE(%s::timestamptz, now()), %s) + """, + ( + product_id, + source_id, + SOURCE_HOMEPAGE_BYPOS, + fields, + rec.get("fetched_at"), + Jsonb(_jsonable(raw)), + ), + ) + + update_quality(conn, product_id) + return product_id + + +def load_bypos_record_safe( + conn: psycopg.Connection, rec: dict[str, Any], source_id: str, raw: dict +) -> bool: + """Load one bypos record inside a savepoint (see :func:`load_record_safe`).""" + try: + with conn.transaction(): + load_bypos_record(conn, rec, source_id, raw) + return True + except Exception as exc: # noqa: BLE001 - per-record isolation is intentional + logger.warning("skipping bypos record gtin=%s: %s", rec.get("gtin"), exc) + return False + + def _jsonable(raw: dict) -> dict: """Drop values that are not JSON-serializable from a raw record.""" try: diff --git a/ingestion/opengoods/jobs/import_bypos.py b/ingestion/opengoods/jobs/import_bypos.py new file mode 100644 index 0000000..d466955 --- /dev/null +++ b/ingestion/opengoods/jobs/import_bypos.py @@ -0,0 +1,77 @@ +"""Import products collected by the bypos-collector tool into OpenGoods. + +The ``tools/bypos-collector`` program writes one product per line (JSONL). This +job reads such a file, transforms each ``hit`` record into the internal product +shape, and upserts it into the database under the ``bypos中心库`` source with +field-level provenance. + +Usage:: + + python -m opengoods.jobs.import_bypos --input products.jsonl + python -m opengoods.jobs.import_bypos --input products.jsonl --limit 500 + +Re-running is safe: products are upserted by GTIN and the source's MSRP/ +provenance rows are refreshed rather than duplicated. +""" + +from __future__ import annotations + +import argparse +import gzip +import json +import sys +from collections.abc import Iterator + +import psycopg + +from opengoods.adapters.bypos import transform_bypos +from opengoods.etl.load import default_dsn, ensure_bypos_source, load_bypos_record_safe + + +def _read_jsonl(path: str) -> Iterator[dict]: + opener = gzip.open if path.endswith(".gz") else open + with opener(path, "rt", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + yield json.loads(line) + except json.JSONDecodeError: + continue + + +def run(args: argparse.Namespace) -> int: + loaded = skipped = errored = 0 + with psycopg.connect(args.dsn, autocommit=False) as conn: + source_id = ensure_bypos_source(conn) + yielded = 0 + for raw in _read_jsonl(args.input): + if args.limit and yielded >= args.limit: + break + rec = transform_bypos(raw) + if rec is None: + skipped += 1 + continue + yielded += 1 + if load_bypos_record_safe(conn, rec, source_id, raw): + loaded += 1 + else: + errored += 1 + conn.commit() + print(f"loaded={loaded} skipped={skipped} errored={errored}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Import bypos-collector JSONL into OpenGoods") + parser.add_argument( + "--input", required=True, help="path to a bypos-collector JSONL (.jsonl or .jsonl.gz)" + ) + parser.add_argument("--limit", type=int, default=0, help="max hit records to load (0 = all)") + parser.add_argument("--dsn", default=default_dsn(), help="PostgreSQL DSN") + return run(parser.parse_args(argv)) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ingestion/tests/test_bypos.py b/ingestion/tests/test_bypos.py new file mode 100644 index 0000000..01187fb --- /dev/null +++ b/ingestion/tests/test_bypos.py @@ -0,0 +1,155 @@ +"""Tests for the bypos-collector adapter/transform and DB loader. + +The pure-function tests run anywhere; the DB roundtrip is skipped automatically +when no database is reachable or migrations are not applied. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from opengoods.adapters.bypos import transform_bypos + +psycopg = pytest.importorskip("psycopg") + +# A real central-library "hit" row as emitted by the collector. +HIT = { + "barcode": "6901028941068", + "name": "泰山合悦", + "spec": "20支", + "unit": "盒", + "area": "广东", + "manufacturer": "", + "license": "", + "in_price": "22.50", + "sell_price": "28.00", + "status": "hit", + "retmsg": "获取商品信息成功", + "fetched_at": "2026-06-24T02:08:34Z", + "source": "zc.bypos.net", +} + +# A hit with a real mass/volume spec that should normalize to net content. +HIT_VOLUME = { + "barcode": "6920459905012", + "name": "康师傅冰红茶490ml", + "spec": "490毫升", + "unit": "瓶", + "area": "浙江杭州", + "manufacturer": "", + "license": "", + "in_price": "2.20", + "sell_price": "3.00", + "status": "hit", + "fetched_at": "2026-06-24T02:08:34Z", + "source": "zc.bypos.net", +} + + +def test_transform_skips_non_hit(): + assert transform_bypos({**HIT, "status": "miss", "name": ""}) is None + assert transform_bypos({**HIT, "status": "invalid"}) is None + + +def test_transform_skips_missing_name(): + assert transform_bypos({**HIT, "name": " "}) is None + + +def test_transform_packaging_spec_kept_raw(): + rec = transform_bypos(HIT) + assert rec is not None + assert rec["gtin"] == "6901028941068" + assert rec["name"] == "泰山合悦" + # '20支' is a count, not mass/volume -> no net_content, raw spec retained. + assert rec["net_content_value"] is None + assert rec["net_content_unit"] is None + assert rec["attributes"]["spec"] == "20支" + assert rec["attributes"]["pack_unit"] == "盒" + assert rec["attributes"]["origin_area"] == "广东" + assert rec["attributes"]["suggested_in_price"] == 22.5 + assert rec["attributes"]["suggested_retail_price"] == 28.0 + assert rec["msrp"] == Decimal("28.00") + assert rec["country_of_origin"] == "中国" + + +def test_transform_volume_spec_normalized(): + rec = transform_bypos(HIT_VOLUME) + assert rec is not None + assert rec["net_content_value"] == Decimal("490") + assert rec["net_content_unit"] == "ml" + assert rec["net_content_canonical"] == Decimal("490") + assert rec["attributes"]["origin_area"] == "浙江杭州" + + +def test_transform_zero_price_dropped(): + rec = transform_bypos({**HIT, "in_price": "0.00", "sell_price": "0.00"}) + assert rec is not None + assert rec["msrp"] is None + assert "suggested_in_price" not in rec["attributes"] + assert "suggested_retail_price" not in rec["attributes"] + + +def test_transform_invalid_barcode_no_gtin(): + rec = transform_bypos({**HIT, "barcode": "123"}) + assert rec is not None + assert rec["gtin"] is None + assert rec["country_of_origin"] is None + + +# --- DB roundtrip (skipped without a database) --------------------------------- + + +@pytest.fixture() +def conn(): + from opengoods.etl.load import default_dsn + + try: + c = psycopg.connect(default_dsn(), connect_timeout=3) + except psycopg.OperationalError as exc: # pragma: no cover - env dependent + pytest.skip(f"no database available: {exc}") + has_product = c.execute("SELECT to_regclass('public.product') IS NOT NULL").fetchone()[0] + if not has_product: + c.close() + pytest.skip("migrations not applied") + yield c + c.rollback() + c.close() + + +def test_load_bypos_roundtrip(conn): + from opengoods.etl.load import ensure_bypos_source, load_bypos_record + + source_id = ensure_bypos_source(conn) + rec = transform_bypos(HIT_VOLUME) + product_id = load_bypos_record(conn, rec, source_id, HIT_VOLUME) + + row = conn.execute( + "SELECT name, gtin, net_content_unit, country_of_origin, attributes ->> 'pack_unit' " + "FROM product WHERE id = %s", + (product_id,), + ).fetchone() + assert row[0] == "康师傅冰红茶490ml" + assert row[1] == "6920459905012" + assert row[2] == "ml" + assert row[3] == "中国" + assert row[4] == "瓶" + + msrp = conn.execute( + "SELECT amount, currency FROM product_msrp WHERE product_id = %s AND source_id = %s", + (product_id, source_id), + ).fetchone() + assert msrp[0] == Decimal("3.00") + assert msrp[1] == "CNY" + + # Re-import is idempotent: still one MSRP and one provenance row per source. + load_bypos_record(conn, rec, source_id, HIT_VOLUME) + counts = conn.execute( + "SELECT (SELECT count(*) FROM product_msrp WHERE product_id=%s AND source_id=%s), " + "(SELECT count(*) FROM product_source WHERE product_id=%s AND source_id=%s)", + (product_id, source_id, product_id, source_id), + ).fetchone() + assert counts == (1, 1) + + conn.rollback() diff --git a/tools/bypos-collector/README.md b/tools/bypos-collector/README.md index cd5cc84..6039816 100644 --- a/tools/bypos-collector/README.md +++ b/tools/bypos-collector/README.md @@ -62,6 +62,21 @@ GET http://zc.bypos.net/byGoodsService/byMessage.asmx/GetGoodsinfo `collect.go` 的 `defaultSdogID`,也可用 `-sdogid` 参数或控制台覆盖。 **这是账号级凭证**——若本仓库对外公开,建议改为从环境变量/外部配置读取。 +## 导入 goods/天工库 + +采集得到的 JSONL 用 ingestion 里的导入任务入库(只导入 `status=hit` 的记录, +按 GTIN 去重 upsert,重复导入幂等): + +```bash +cd ingestion +python -m opengoods.jobs.import_bypos --input products.jsonl +# 可选:--limit N 限制条数;--dsn 指定数据库 +``` + +字段映射:`barcode→gtin`、`name→name`、`spec`(可解析的质量/体积如 500mL/5kg) +`→net_content`、其余规格/单位/产地/许可/进价留存到 `attributes`、`sell_price→` +CNY MSRP 快照。来源记为 `bypos中心库`,带字段级 provenance。 + ## 注意 批量自动查询比页面逐条更"重",上游可能对账号限频。请低速、分前缀/品类分批采集。