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

@@ -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,
+    }

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:

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())

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()

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。
+
 ## 注意
 
 批量自动查询比页面逐条更"重",上游可能对账号限频。请低速、分前缀/品类分批采集。