feat(ingestion): import bypos-collector JSONL into goods

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

admin-frontend/src/api.ts

@@ -51,14 +51,23 @@ export const api = {
       body: JSON.stringify({ username, password }),
     }),
   me: () => request<{ username: string }>("/me"),
-  listProducts: (q: string, page: number, size: number) =>
+  listProducts: (
+    q: string,
+    page: number,
+    size: number,
+    sort?: string,
+    order?: string,
+  ) =>
     request<{
       items: import("./types").ProductRow[];
       page: number;
       size: number;
       total: number;
       completeness_fields: string[];
-    }>(`/products?q=${encodeURIComponent(q)}&page=${page}&size=${size}`),
+    }>(
+      `/products?q=${encodeURIComponent(q)}&page=${page}&size=${size}` +
+        (sort ? `&sort=${sort}&order=${order || "asc"}` : ""),
+    ),
   getProduct: (id: string) =>
     request<import("./types").ProductDetail>(`/products/${id}`),
   createProduct: (body: unknown) =>

admin-frontend/src/components/ProductList.tsx

@@ -1,7 +1,15 @@
 import { useEffect, useState } from "react";
 import { api, ApiError } from "../api";
 import { Brand, Category, FIELD_LABELS, ProductRow } from "../types";
-import { Search, AlertCircle, Plus } from "lucide-react";
+import { Search, AlertCircle, Plus, ChevronUp, ChevronDown, ChevronsUpDown } from "lucide-react";
+
+type SortKey =
+  | "name"
+  | "brand"
+  | "gtin"
+  | "category_path"
+  | "status"
+  | "quality_score";
 
 const STATUS_LABEL: Record<string, string> = {
   active: "在用",
@@ -24,6 +32,42 @@ function QualityBadge({ score }: { score: number }) {
   );
 }
 
+function SortableTh({
+  label,
+  sortKey,
+  sort,
+  order,
+  onSort,
+}: {
+  label: string;
+  sortKey: SortKey;
+  sort: SortKey | "";
+  order: "asc" | "desc";
+  onSort: (key: SortKey) => void;
+}) {
+  const active = sort === sortKey;
+  return (
+    <th className="px-4 py-3">
+      <button
+        type="button"
+        onClick={() => onSort(sortKey)}
+        className={`flex items-center gap-1 uppercase hover:text-gray-700 ${
+          active ? "text-emerald-600" : ""
+        }`}
+      >
+        {label}
+        {!active ? (
+          <ChevronsUpDown className="h-3.5 w-3.5 text-gray-300" />
+        ) : order === "asc" ? (
+          <ChevronUp className="h-3.5 w-3.5" />
+        ) : (
+          <ChevronDown className="h-3.5 w-3.5" />
+        )}
+      </button>
+    </th>
+  );
+}
+
 export default function ProductList({
   onOpen,
 }: {
@@ -33,6 +77,8 @@ export default function ProductList({
   const [input, setInput] = useState("");
   const [page, setPage] = useState(1);
   const [size, setSize] = useState(20);
+  const [sort, setSort] = useState<SortKey | "">("");
+  const [order, setOrder] = useState<"asc" | "desc">("asc");
   const [jump, setJump] = useState("");
   const [rows, setRows] = useState<ProductRow[]>([]);
   const [total, setTotal] = useState(0);
@@ -49,7 +95,7 @@ export default function ProductList({
     setLoading(true);
     setError("");
     api
-      .listProducts(q, page, size)
+      .listProducts(q, page, size, sort || undefined, order)
       .then((r) => {
         setRows(r.items);
         setTotal(r.total);
@@ -61,7 +107,20 @@ export default function ProductList({
   useEffect(() => {
     setSelected(new Set());
     reload();
-  }, [q, page, size]);
+  }, [q, page, size, sort, order]);
+
+  function toggleSort(key: SortKey) {
+    setPage(1);
+    if (sort !== key) {
+      setSort(key);
+      setOrder("asc");
+    } else if (order === "asc") {
+      setOrder("desc");
+    } else {
+      setSort("");
+      setOrder("asc");
+    }
+  }
 
   useEffect(() => {
     api.listCategories().then((r) => setCategories(r.items)).catch(() => {});
@@ -245,12 +304,12 @@ export default function ProductList({
                   aria-label="全选"
                 />
               </th>
-              <th className="px-4 py-3">名称</th>
-              <th className="px-4 py-3">品牌</th>
-              <th className="px-4 py-3">条码</th>
-              <th className="px-4 py-3">品类</th>
-              <th className="px-4 py-3">状态</th>
-              <th className="px-4 py-3">质量分</th>
+              <SortableTh label="名称" sortKey="name" sort={sort} order={order} onSort={toggleSort} />
+              <SortableTh label="品牌" sortKey="brand" sort={sort} order={order} onSort={toggleSort} />
+              <SortableTh label="条码" sortKey="gtin" sort={sort} order={order} onSort={toggleSort} />
+              <SortableTh label="品类" sortKey="category_path" sort={sort} order={order} onSort={toggleSort} />
+              <SortableTh label="状态" sortKey="status" sort={sort} order={order} onSort={toggleSort} />
+              <SortableTh label="质量分" sortKey="quality_score" sort={sort} order={order} onSort={toggleSort} />
               <th className="px-4 py-3">缺失字段</th>
             </tr>
           </thead>

api/internal/adminhandler/handler.go

@@ -164,8 +164,10 @@ func (h *Handler) Me(w http.ResponseWriter, r *http.Request) {
 // ListProducts returns a paginated product list.
 func (h *Handler) ListProducts(w http.ResponseWriter, r *http.Request) {
 	q := r.URL.Query().Get("q")
+	sort := r.URL.Query().Get("sort")
+	order := r.URL.Query().Get("order")
 	page, size := pageParams(r)
-	items, total, err := h.store.ListProducts(r.Context(), q, size, (page-1)*size)
+	items, total, err := h.store.ListProducts(r.Context(), q, sort, order, size, (page-1)*size)
 	if h.handleErr(w, err) {
 		return
 	}

api/internal/adminstore/adminstore.go

@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -50,8 +51,34 @@ type ProductRow struct {
 	UpdatedAt    string   `json:"updated_at"`
 }
 
+// productSortColumns whitelists the sortable list columns, mapping the API sort
+// key to a SQL expression. NULLs sort last regardless of direction.
+var productSortColumns = map[string]string{
+	"name":          "p.name",
+	"brand":         "b.name",
+	"gtin":          "p.gtin",
+	"category_path": "c.path",
+	"status":        "p.status",
+	"quality_score": "p.quality_score",
+	"updated_at":    "p.updated_at",
+}
+
+// productOrderBy returns a safe ORDER BY clause for the given sort key/direction,
+// falling back to the default (most recently updated first) for unknown keys.
+func productOrderBy(sort, order string) string {
+	col, ok := productSortColumns[sort]
+	if !ok {
+		return "p.updated_at DESC"
+	}
+	dir := "ASC"
+	if strings.EqualFold(order, "desc") {
+		dir = "DESC"
+	}
+	return col + " " + dir + " NULLS LAST, p.updated_at DESC"
+}
+
 // ListProducts returns a paginated, optionally name/gtin-filtered list.
-func (s *Store) ListProducts(ctx context.Context, q string, limit, offset int) ([]ProductRow, int, error) {
+func (s *Store) ListProducts(ctx context.Context, q, sort, order string, limit, offset int) ([]ProductRow, int, error) {
 	args := []any{}
 	where := "WHERE 1=1"
 	if q != "" {
@@ -84,7 +111,8 @@ FROM product p
 LEFT JOIN brand b ON b.id = p.brand_id
 LEFT JOIN category c ON c.id = p.category_id
 LEFT JOIN food_detail f ON f.product_id = p.id ` + where +
-		" ORDER BY p.updated_at DESC LIMIT $" + strconv.Itoa(len(args)-1) + " OFFSET $" + strconv.Itoa(len(args))
+		" ORDER BY " + productOrderBy(sort, order) +
+		" LIMIT $" + strconv.Itoa(len(args)-1) + " OFFSET $" + strconv.Itoa(len(args))
 
 	rows, err := s.pool.Query(ctx, sql, args...)
 	if err != nil {

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/.gitignore

@@ -0,0 +1,9 @@
+# build artifacts
+bypos-collector.exe
+bypos-collector
+bypos-collector-linux
+*.exe
+# collected data / outputs
+*.jsonl
+*.csv
+*.log

tools/bypos-collector/README.md

@@ -0,0 +1,82 @@
+# bypos-collector
+
+按条码批量采集商品档案的小工具(单文件 Windows/Linux 程序,自带本地 Web 控制台)。
+数据来源是云店「新增商品」输入条码时所查的同一个中心商品库 `zc.bypos.net`。
+采集结果落地为 JSONL,供后续导入本项目(天工/goods)。
+
+> 这是一个**独立模块**(有自己的 `go.mod`),与 `api/` 主服务互不影响,
+> CI 不会编译它。放在 `tools/` 下仅作代码留存与后期迭代。
+
+## 目录
+
+| 文件 | 说明 |
+| --- | --- |
+| `main.go` | 入口:本地 HTTP 服务 + 启动浏览器 + API 路由(start/stop/stats/download/export.csv) |
+| `collect.go` | 核心:签名、EAN-13 校验位、范围/清单枚举、并发+限速、JSONL 落库、断点续采 |
+| `web/index.html` | 内嵌(`go:embed`)的控制台界面 |
+| `build.sh` | 交叉编译出 `bypos-collector.exe`(windows/amd64)与 linux 测试二进制 |
+| `使用说明.md` | 面向使用者的操作说明 |
+
+## 构建
+
+```bash
+./build.sh
+# 产物:bypos-collector.exe(发给 Windows 用户)/ bypos-collector-linux(本地测试)
+```
+
+二进制与采集产物(`*.jsonl`/`*.csv`)已在 `.gitignore` 中排除,不入库。
+
+## 接口与签名(逆向所得,后期迭代参考)
+
+云店新增商品页输入条码时,前端经服务端代理 `/prod-api/ZmSvr/httpUtil/getGet`
+转发到中心库:
+
+```
+GET http://zc.bypos.net/byGoodsService/byMessage.asmx/GetGoodsinfo
+    ?sdogid=<账号id>&regnum=1&barcode=<条码>
+    &sparm1=<md5(sdogid)>            # 常量,随账号固定
+    &sparm2=<md5(barcode + tsMs)>    # tsMs = 当前秒*1000(末尾恒为 000)
+    &sparm3=<tsMs 前 10 位 = 秒级时间戳>
+    &sparm4=&barcodetype=yunpos
+```
+
+返回 `<string>{...json...}</string>`,内层 JSON 字段:
+
+| 上游字段 | 含义 | 归一化字段 |
+| --- | --- | --- |
+| item_name | 品名 | name |
+| item_size | 规格 | spec |
+| unit_no | 单位 | unit |
+| item_area | 产地/地区 | area |
+| birth_com | 生产企业(常空) | manufacturer |
+| birth_doc | 生产许可(常空) | license |
+| inprice | 建议进价 | in_price |
+| sellprice | 建议零售价 | sell_price |
+| retcode | 1=命中,0=失败 | status(hit/miss/invalid) |
+
+`retmsg` 含「非国标条码 / 参数异常」=> invalid;含「条码不存在」=> miss。
+
+## 配置
+
+- `sdogid`:中心库账号 id(本项目所属云店账号的授权 id)。默认值见
+  `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。
+
+## 注意
+
+批量自动查询比页面逐条更"重",上游可能对账号限频。请低速、分前缀/品类分批采集。

tools/bypos-collector/build.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+# Build the bypos-collector for Windows (and a Linux binary for local testing).
+set -euo pipefail
+cd "$(dirname "$0")"
+gofmt -w ./*.go
+go vet ./...
+echo "building windows/amd64 .exe ..."
+GOOS=windows GOARCH=amd64 go build -ldflags "-s -w" -o bypos-collector.exe .
+echo "building linux/amd64 (for testing) ..."
+go build -o bypos-collector-linux .
+ls -la bypos-collector.exe bypos-collector-linux

tools/bypos-collector/collect.go

@@ -0,0 +1,481 @@
+package main
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// ---- upstream config ----
+// sdogid is the 云店 account license id observed in the live request. It is
+// configurable so the tool is not tied to a single account.
+const defaultSdogID = "137966"
+
+const endpoint = "http://zc.bypos.net/byGoodsService/byMessage.asmx/GetGoodsinfo"
+
+var stringTagRe = regexp.MustCompile(`(?s)<string[^>]*>(.*)</string>`)
+
+// Product is the normalized record we persist (one JSON object per line).
+type Product struct {
+	Barcode      string `json:"barcode"`
+	Name         string `json:"name"`         // item_name 品名
+	Spec         string `json:"spec"`         // item_size 规格
+	Unit         string `json:"unit"`         // unit_no 单位
+	Area         string `json:"area"`         // item_area 产地/地区
+	Manufacturer string `json:"manufacturer"` // birth_com 生产企业
+	License      string `json:"license"`      // birth_doc 生产许可
+	InPrice      string `json:"in_price"`     // 建议进价
+	SellPrice    string `json:"sell_price"`   // 建议零售价
+	Status       string `json:"status"`       // hit / miss / invalid / error
+	RetMsg       string `json:"retmsg"`       // 原始返回信息
+	FetchedAt    string `json:"fetched_at"`   // RFC3339
+	Source       string `json:"source"`       // zc.bypos.net
+}
+
+// upstream raw fields
+type rawResp struct {
+	RetCode   string `json:"retcode"`
+	RetMsg    string `json:"retmsg"`
+	Barcode   string `json:"barcode"`
+	ItemName  string `json:"item_name"`
+	UnitNo    string `json:"unit_no"`
+	ItemSize  string `json:"item_size"`
+	ItemArea  string `json:"item_area"`
+	BirthCom  string `json:"birth_com"`
+	BirthDoc  string `json:"birth_doc"`
+	InPrice   string `json:"inprice"`
+	SellPrice string `json:"sellprice"`
+}
+
+func md5hex(s string) string {
+	h := md5.Sum([]byte(s))
+	return hex.EncodeToString(h[:])
+}
+
+// ean13Check computes the EAN-13 check digit for a 12-digit body.
+func ean13Check(body string) (string, bool) {
+	if len(body) != 12 {
+		return "", false
+	}
+	sum := 0
+	for i := 0; i < 12; i++ {
+		c := body[i]
+		if c < '0' || c > '9' {
+			return "", false
+		}
+		d := int(c - '0')
+		if i%2 == 0 {
+			sum += d
+		} else {
+			sum += d * 3
+		}
+	}
+	chk := (10 - (sum % 10)) % 10
+	return body + strconv.Itoa(chk), true
+}
+
+// lookup queries the upstream central library for one barcode.
+func (c *Collector) lookup(ctx context.Context, barcode string) (*Product, error) {
+	tsMs := strconv.FormatInt(time.Now().Unix()*1000, 10) // always ends in 000
+	sparm1 := md5hex(c.sdogID)
+	sparm2 := md5hex(barcode + tsMs)
+	sparm3 := tsMs[:10]
+	q := url.Values{}
+	q.Set("sdogid", c.sdogID)
+	q.Set("regnum", "1")
+	q.Set("barcode", barcode)
+	q.Set("sparm1", sparm1)
+	q.Set("sparm2", sparm2)
+	q.Set("sparm3", sparm3)
+	q.Set("sparm4", "")
+	q.Set("barcodetype", "yunpos")
+	reqURL := endpoint + "?" + q.Encode()
+
+	req, _ := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
+	req.Header.Set("User-Agent", "Mozilla/5.0")
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	b, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	inner := b
+	if m := stringTagRe.FindSubmatch(b); m != nil {
+		inner = m[1]
+	}
+	var r rawResp
+	if err := json.Unmarshal(inner, &r); err != nil {
+		return nil, fmt.Errorf("parse: %v (body=%.120s)", err, string(b))
+	}
+	p := &Product{
+		Barcode:   barcode,
+		RetMsg:    r.RetMsg,
+		FetchedAt: time.Now().Format(time.RFC3339),
+		Source:    "zc.bypos.net",
+	}
+	if r.RetCode == "1" {
+		p.Status = "hit"
+		p.Name = strings.TrimSpace(r.ItemName)
+		p.Spec = strings.TrimSpace(r.ItemSize)
+		p.Unit = strings.TrimSpace(r.UnitNo)
+		p.Area = strings.TrimSpace(r.ItemArea)
+		p.Manufacturer = strings.TrimSpace(r.BirthCom)
+		p.License = strings.TrimSpace(r.BirthDoc)
+		p.InPrice = strings.TrimSpace(r.InPrice)
+		p.SellPrice = strings.TrimSpace(r.SellPrice)
+	} else if strings.Contains(r.RetMsg, "非国标") || strings.Contains(r.RetMsg, "参数异常") {
+		p.Status = "invalid"
+	} else {
+		p.Status = "miss"
+	}
+	return p, nil
+}
+
+// ---- job / collector state ----
+
+type Stats struct {
+	Running   bool   `json:"running"`
+	Total     int64  `json:"total"`
+	Done      int64  `json:"done"`
+	Hits      int64  `json:"hits"`
+	Miss      int64  `json:"miss"`
+	Invalid   int64  `json:"invalid"`
+	Errors    int64  `json:"errors"`
+	Skipped   int64  `json:"skipped"`
+	Current   string `json:"current"`
+	OutFile   string `json:"out_file"`
+	StartedAt string `json:"started_at"`
+	Message   string `json:"message"`
+}
+
+type Collector struct {
+	mu      sync.Mutex
+	client  *http.Client
+	sdogID  string
+	outPath string
+	outFile *os.File
+
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+
+	// atomic counters
+	total, done, hits, miss, invalid, errors, skipped int64
+	running                                           int32
+
+	current   atomic.Value // string
+	startedAt string
+	message   string
+
+	seen   map[string]struct{} // barcodes already in output (dedupe / resume)
+	recent []Product           // ring of last results for UI
+}
+
+func NewCollector(sdogID string) *Collector {
+	if sdogID == "" {
+		sdogID = defaultSdogID
+	}
+	c := &Collector{
+		client: &http.Client{Timeout: 25 * time.Second},
+		sdogID: sdogID,
+		seen:   map[string]struct{}{},
+	}
+	c.current.Store("")
+	return c
+}
+
+func (c *Collector) isRunning() bool { return atomic.LoadInt32(&c.running) == 1 }
+
+func (c *Collector) snapshot() Stats {
+	cur, _ := c.current.Load().(string)
+	c.mu.Lock()
+	msg := c.message
+	out := c.outPath
+	started := c.startedAt
+	c.mu.Unlock()
+	return Stats{
+		Running:   c.isRunning(),
+		Total:     atomic.LoadInt64(&c.total),
+		Done:      atomic.LoadInt64(&c.done),
+		Hits:      atomic.LoadInt64(&c.hits),
+		Miss:      atomic.LoadInt64(&c.miss),
+		Invalid:   atomic.LoadInt64(&c.invalid),
+		Errors:    atomic.LoadInt64(&c.errors),
+		Skipped:   atomic.LoadInt64(&c.skipped),
+		Current:   cur,
+		OutFile:   out,
+		StartedAt: started,
+		Message:   msg,
+	}
+}
+
+func (c *Collector) recentResults() []Product {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	out := make([]Product, len(c.recent))
+	copy(out, c.recent)
+	return out
+}
+
+func (c *Collector) pushRecent(p Product) {
+	c.mu.Lock()
+	c.recent = append(c.recent, p)
+	if len(c.recent) > 60 {
+		c.recent = c.recent[len(c.recent)-60:]
+	}
+	c.mu.Unlock()
+}
+
+// loadSeen reads an existing output file to build the dedupe set (for resume).
+func (c *Collector) loadSeen(path string) error {
+	c.seen = map[string]struct{}{}
+	f, err := os.Open(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+	defer f.Close()
+	dec := json.NewDecoder(f)
+	for {
+		var p Product
+		if err := dec.Decode(&p); err != nil {
+			break
+		}
+		if p.Barcode != "" {
+			c.seen[p.Barcode] = struct{}{}
+		}
+	}
+	return nil
+}
+
+type JobReq struct {
+	Mode        string `json:"mode"`        // "range" | "list"
+	StartBody   string `json:"start_body"`  // 12-digit body (range mode)
+	EndBody     string `json:"end_body"`    // 12-digit body (range mode)
+	List        string `json:"list"`        // newline/space separated barcodes (list mode)
+	Concurrency int    `json:"concurrency"` // parallel requests
+	DelayMs     int    `json:"delay_ms"`    // min interval between request starts
+	LogMiss     bool   `json:"log_miss"`    // also write miss/invalid lines
+	OutFile     string `json:"out_file"`
+	SdogID      string `json:"sdog_id"`
+}
+
+func sanitizeBarcodes(s string) []string {
+	fields := regexp.MustCompile(`[^0-9]+`).Split(s, -1)
+	var out []string
+	for _, f := range fields {
+		if f != "" {
+			out = append(out, f)
+		}
+	}
+	return out
+}
+
+// Start launches a collection job. Returns error if validation fails or busy.
+func (c *Collector) Start(req JobReq) error {
+	if c.isRunning() {
+		return fmt.Errorf("已有任务在运行")
+	}
+	if req.Concurrency <= 0 {
+		req.Concurrency = 3
+	}
+	if req.Concurrency > 20 {
+		req.Concurrency = 20
+	}
+	if req.DelayMs < 0 {
+		req.DelayMs = 0
+	}
+	if req.OutFile == "" {
+		req.OutFile = "products.jsonl"
+	}
+	if req.SdogID != "" {
+		c.sdogID = req.SdogID
+	}
+
+	// Build the list of barcodes to query.
+	var barcodes []string
+	switch req.Mode {
+	case "list":
+		barcodes = sanitizeBarcodes(req.List)
+		if len(barcodes) == 0 {
+			return fmt.Errorf("条码清单为空")
+		}
+	case "range":
+		start, err := strconv.ParseInt(req.StartBody, 10, 64)
+		if err != nil || len(req.StartBody) != 12 {
+			return fmt.Errorf("起始码必须是 12 位数字(不含校验位)")
+		}
+		end, err := strconv.ParseInt(req.EndBody, 10, 64)
+		if err != nil || len(req.EndBody) != 12 {
+			return fmt.Errorf("结束码必须是 12 位数字(不含校验位)")
+		}
+		if end < start {
+			return fmt.Errorf("结束码不能小于起始码")
+		}
+		if end-start+1 > 5_000_000 {
+			return fmt.Errorf("单次范围过大(>500万),请缩小区间分批采集")
+		}
+		for v := start; v <= end; v++ {
+			body := fmt.Sprintf("%012d", v)
+			full, ok := ean13Check(body)
+			if ok {
+				barcodes = append(barcodes, full)
+			}
+		}
+	default:
+		return fmt.Errorf("未知模式: %s", req.Mode)
+	}
+
+	abs, _ := filepath.Abs(req.OutFile)
+	if err := c.loadSeen(abs); err != nil {
+		return fmt.Errorf("读取已有文件失败: %v", err)
+	}
+	f, err := os.OpenFile(abs, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		return fmt.Errorf("打开输出文件失败: %v", err)
+	}
+	c.outFile = f
+	c.outPath = abs
+
+	// reset counters
+	atomic.StoreInt64(&c.total, int64(len(barcodes)))
+	atomic.StoreInt64(&c.done, 0)
+	atomic.StoreInt64(&c.hits, 0)
+	atomic.StoreInt64(&c.miss, 0)
+	atomic.StoreInt64(&c.invalid, 0)
+	atomic.StoreInt64(&c.errors, 0)
+	atomic.StoreInt64(&c.skipped, 0)
+	c.mu.Lock()
+	c.recent = nil
+	c.startedAt = time.Now().Format(time.RFC3339)
+	c.message = ""
+	c.mu.Unlock()
+	atomic.StoreInt32(&c.running, 1)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	c.cancel = cancel
+
+	go c.run(ctx, barcodes, req)
+	return nil
+}
+
+func (c *Collector) Stop() {
+	if c.cancel != nil {
+		c.cancel()
+	}
+}
+
+func (c *Collector) run(ctx context.Context, barcodes []string, req JobReq) {
+	defer func() {
+		atomic.StoreInt32(&c.running, 0)
+		if c.outFile != nil {
+			c.outFile.Sync()
+			c.outFile.Close()
+			c.outFile = nil
+		}
+		c.current.Store("")
+	}()
+
+	jobs := make(chan string, req.Concurrency*2)
+	var writeMu sync.Mutex
+
+	// global rate limiter: one token every DelayMs
+	var ticker *time.Ticker
+	if req.DelayMs > 0 {
+		ticker = time.NewTicker(time.Duration(req.DelayMs) * time.Millisecond)
+		defer ticker.Stop()
+	}
+
+	worker := func() {
+		defer c.wg.Done()
+		for bc := range jobs {
+			if ctx.Err() != nil {
+				return
+			}
+			if ticker != nil {
+				select {
+				case <-ticker.C:
+				case <-ctx.Done():
+					return
+				}
+			}
+			c.current.Store(bc)
+			p, err := c.lookup(ctx, bc)
+			if err != nil {
+				if ctx.Err() != nil {
+					return
+				}
+				atomic.AddInt64(&c.errors, 1)
+				atomic.AddInt64(&c.done, 1)
+				ep := Product{Barcode: bc, Status: "error", RetMsg: err.Error(), FetchedAt: time.Now().Format(time.RFC3339), Source: "zc.bypos.net"}
+				c.pushRecent(ep)
+				continue
+			}
+			switch p.Status {
+			case "hit":
+				atomic.AddInt64(&c.hits, 1)
+			case "miss":
+				atomic.AddInt64(&c.miss, 1)
+			case "invalid":
+				atomic.AddInt64(&c.invalid, 1)
+			}
+			atomic.AddInt64(&c.done, 1)
+			c.pushRecent(*p)
+			if p.Status == "hit" || req.LogMiss {
+				line, _ := json.Marshal(p)
+				writeMu.Lock()
+				c.outFile.Write(line)
+				c.outFile.Write([]byte("\n"))
+				c.seen[bc] = struct{}{}
+				writeMu.Unlock()
+			}
+		}
+	}
+
+	for i := 0; i < req.Concurrency; i++ {
+		c.wg.Add(1)
+		go worker()
+	}
+
+	for _, bc := range barcodes {
+		if ctx.Err() != nil {
+			break
+		}
+		if _, ok := c.seen[bc]; ok {
+			atomic.AddInt64(&c.skipped, 1)
+			atomic.AddInt64(&c.done, 1)
+			continue
+		}
+		select {
+		case jobs <- bc:
+		case <-ctx.Done():
+		}
+	}
+	close(jobs)
+	c.wg.Wait()
+
+	c.mu.Lock()
+	if ctx.Err() != nil {
+		c.message = "已停止"
+	} else {
+		c.message = "采集完成"
+	}
+	c.mu.Unlock()
+}

tools/bypos-collector/go.mod

@@ -0,0 +1,3 @@
+module byposcollector
+
+go 1.23.4

tools/bypos-collector/main.go

@@ -0,0 +1,156 @@
+package main
+
+import (
+	"embed"
+	"encoding/csv"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"io/fs"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"runtime"
+	"time"
+)
+
+//go:embed web/*
+var webFS embed.FS
+
+var collector = NewCollector("")
+
+func main() {
+	addr := flag.String("addr", "127.0.0.1:8765", "本地监听地址")
+	noOpen := flag.Bool("no-open", false, "不自动打开浏览器")
+	sdog := flag.String("sdogid", "", "中心库账号 id(默认使用内置值)")
+	flag.Parse()
+	if *sdog != "" {
+		collector.sdogID = *sdog
+	}
+
+	sub, _ := fs.Sub(webFS, "web")
+	mux := http.NewServeMux()
+	mux.Handle("/", http.FileServer(http.FS(sub)))
+	mux.HandleFunc("/api/start", handleStart)
+	mux.HandleFunc("/api/stop", handleStop)
+	mux.HandleFunc("/api/stats", handleStats)
+	mux.HandleFunc("/api/download", handleDownload)
+	mux.HandleFunc("/api/export.csv", handleExportCSV)
+
+	ln, err := net.Listen("tcp", *addr)
+	if err != nil {
+		log.Fatalf("无法监听 %s: %v", *addr, err)
+	}
+	realAddr := ln.Addr().String()
+	urlStr := "http://" + realAddr + "/"
+	fmt.Println("==============================================")
+	fmt.Println(" 中心库商品采集器  bypos-collector")
+	fmt.Println(" 控制台: " + urlStr)
+	fmt.Println(" 关闭本窗口即停止程序")
+	fmt.Println("==============================================")
+	if !*noOpen {
+		go openBrowser(urlStr)
+	}
+	log.Fatal(http.Serve(ln, mux))
+}
+
+func openBrowser(url string) {
+	time.Sleep(600 * time.Millisecond)
+	var cmd string
+	var args []string
+	switch runtime.GOOS {
+	case "windows":
+		cmd = "rundll32"
+		args = []string{"url.dll,FileProtocolHandler", url}
+	case "darwin":
+		cmd = "open"
+		args = []string{url}
+	default:
+		cmd = "xdg-open"
+		args = []string{url}
+	}
+	_ = exec.Command(cmd, args...).Start()
+}
+
+func writeJSON(w http.ResponseWriter, code int, v interface{}) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(code)
+	json.NewEncoder(w).Encode(v)
+}
+
+func handleStart(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.Error(w, "method", 405)
+		return
+	}
+	var req JobReq
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeJSON(w, 400, map[string]string{"error": "请求格式错误"})
+		return
+	}
+	if err := collector.Start(req); err != nil {
+		writeJSON(w, 400, map[string]string{"error": err.Error()})
+		return
+	}
+	writeJSON(w, 200, map[string]string{"ok": "started"})
+}
+
+func handleStop(w http.ResponseWriter, r *http.Request) {
+	collector.Stop()
+	writeJSON(w, 200, map[string]string{"ok": "stopping"})
+}
+
+func handleStats(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, 200, map[string]interface{}{
+		"stats":  collector.snapshot(),
+		"recent": collector.recentResults(),
+	})
+}
+
+func handleDownload(w http.ResponseWriter, r *http.Request) {
+	s := collector.snapshot()
+	if s.OutFile == "" {
+		http.Error(w, "no output yet", 404)
+		return
+	}
+	f, err := os.Open(s.OutFile)
+	if err != nil {
+		http.Error(w, err.Error(), 404)
+		return
+	}
+	defer f.Close()
+	w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8")
+	w.Header().Set("Content-Disposition", "attachment; filename=products.jsonl")
+	io.Copy(w, f)
+}
+
+func handleExportCSV(w http.ResponseWriter, r *http.Request) {
+	s := collector.snapshot()
+	if s.OutFile == "" {
+		http.Error(w, "no output yet", 404)
+		return
+	}
+	f, err := os.Open(s.OutFile)
+	if err != nil {
+		http.Error(w, err.Error(), 404)
+		return
+	}
+	defer f.Close()
+	w.Header().Set("Content-Type", "text/csv; charset=utf-8")
+	w.Header().Set("Content-Disposition", "attachment; filename=products.csv")
+	w.Write([]byte{0xEF, 0xBB, 0xBF}) // UTF-8 BOM so Excel reads Chinese correctly
+	cw := csv.NewWriter(w)
+	cw.Write([]string{"barcode", "name", "spec", "unit", "area", "manufacturer", "license", "in_price", "sell_price", "status", "fetched_at"})
+	dec := json.NewDecoder(f)
+	for {
+		var p Product
+		if err := dec.Decode(&p); err != nil {
+			break
+		}
+		cw.Write([]string{p.Barcode, p.Name, p.Spec, p.Unit, p.Area, p.Manufacturer, p.License, p.InPrice, p.SellPrice, p.Status, p.FetchedAt})
+	}
+	cw.Flush()
+}

tools/bypos-collector/web/index.html

@@ -0,0 +1,206 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+<meta charset="utf-8"/>
+<meta name="viewport" content="width=device-width, initial-scale=1"/>
+<title>中心库商品采集器</title>
+<style>
+  * { box-sizing: border-box; }
+  body { font-family: -apple-system, "Microsoft YaHei", Arial, sans-serif; margin: 0; background:#f4f6f9; color:#222; }
+  header { background:#1f6feb; color:#fff; padding:14px 22px; font-size:18px; font-weight:600; }
+  .wrap { max-width:1080px; margin:18px auto; padding:0 16px; }
+  .card { background:#fff; border:1px solid #e3e8ef; border-radius:10px; padding:18px 20px; margin-bottom:16px; }
+  .card h3 { margin:0 0 12px; font-size:15px; color:#1f6feb; }
+  label { display:block; font-size:13px; color:#555; margin:8px 0 4px; }
+  input[type=text], input[type=number], textarea, select {
+    width:100%; padding:8px 10px; border:1px solid #cdd5e0; border-radius:6px; font-size:14px;
+  }
+  textarea { height:90px; font-family:monospace; }
+  .row { display:flex; gap:14px; flex-wrap:wrap; }
+  .row > div { flex:1; min-width:160px; }
+  .tabs { display:flex; gap:8px; margin-bottom:12px; }
+  .tab { padding:7px 16px; border:1px solid #cdd5e0; border-radius:20px; cursor:pointer; font-size:13px; background:#fff; }
+  .tab.active { background:#1f6feb; color:#fff; border-color:#1f6feb; }
+  button.primary { background:#1f6feb; color:#fff; border:none; padding:10px 22px; border-radius:6px; font-size:14px; cursor:pointer; }
+  button.danger { background:#d1242f; color:#fff; border:none; padding:10px 22px; border-radius:6px; font-size:14px; cursor:pointer; }
+  button.ghost { background:#fff; color:#1f6feb; border:1px solid #1f6feb; padding:8px 16px; border-radius:6px; cursor:pointer; font-size:13px; }
+  button:disabled { opacity:.5; cursor:not-allowed; }
+  .stats { display:flex; gap:10px; flex-wrap:wrap; }
+  .stat { flex:1; min-width:90px; background:#f7f9fc; border:1px solid #e3e8ef; border-radius:8px; padding:10px; text-align:center; }
+  .stat .n { font-size:22px; font-weight:700; }
+  .stat .l { font-size:12px; color:#777; margin-top:2px; }
+  .bar { height:10px; background:#e3e8ef; border-radius:6px; overflow:hidden; margin:10px 0; }
+  .bar > div { height:100%; background:#2da44e; width:0%; transition:width .4s; }
+  table { width:100%; border-collapse:collapse; font-size:13px; }
+  th, td { text-align:left; padding:6px 8px; border-bottom:1px solid #eef1f5; white-space:nowrap; overflow:hidden; text-overflow:ellipsis; max-width:180px; }
+  th { color:#888; font-weight:600; }
+  .hit { color:#2da44e; } .miss { color:#999; } .invalid { color:#d1242f; } .error { color:#bf8700; }
+  .hint { font-size:12px; color:#888; margin-top:6px; line-height:1.5; }
+  .est { font-size:13px; color:#1f6feb; margin-top:6px; }
+</style>
+</head>
+<body>
+<header>中心库商品采集器 · bypos-collector</header>
+<div class="wrap">
+
+  <div class="card">
+    <h3>① 规划采集范围</h3>
+    <div class="tabs">
+      <div class="tab active" data-mode="range" onclick="setMode('range')">按条码范围</div>
+      <div class="tab" data-mode="list" onclick="setMode('list')">按条码清单</div>
+    </div>
+
+    <div id="pane-range">
+      <div class="hint">EAN-13 国标条码共 13 位,最后一位是校验位由程序自动计算。下面填<b>前 12 位</b>(本体),程序逐个枚举并补校验位查询。常见前缀:69 开头为中国大陆。</div>
+      <label>快捷填充前缀(可选)</label>
+      <div class="row">
+        <div><input type="text" id="prefix" placeholder="如 690100,点下方按钮自动算区间"/></div>
+        <div style="flex:0"><button class="ghost" onclick="fillFromPrefix()">用前缀填充区间</button></div>
+      </div>
+      <div class="row">
+        <div>
+          <label>起始本体(12 位)</label>
+          <input type="text" id="start" value="690100000000" maxlength="12"/>
+        </div>
+        <div>
+          <label>结束本体(12 位)</label>
+          <input type="text" id="end" value="690100000999" maxlength="12"/>
+        </div>
+      </div>
+      <div class="est" id="est"></div>
+    </div>
+
+    <div id="pane-list" style="display:none">
+      <label>粘贴条码清单(每行一个,或用空格/逗号分隔)</label>
+      <textarea id="list" placeholder="6901028941068&#10;6920202888883"></textarea>
+    </div>
+  </div>
+
+  <div class="card">
+    <h3>② 采集参数</h3>
+    <div class="row">
+      <div>
+        <label>并发数</label>
+        <input type="number" id="concurrency" value="3" min="1" max="20"/>
+      </div>
+      <div>
+        <label>每次请求间隔(毫秒)</label>
+        <input type="number" id="delay" value="300" min="0"/>
+      </div>
+      <div>
+        <label>输出文件名</label>
+        <input type="text" id="outfile" value="products.jsonl"/>
+      </div>
+    </div>
+    <label style="margin-top:12px"><input type="checkbox" id="logmiss"/> 同时记录未命中/无效条码(默认只存命中)</label>
+    <div class="hint">速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(断点续采)。</div>
+  </div>
+
+  <div class="card">
+    <h3>③ 运行</h3>
+    <div style="margin-bottom:12px">
+      <button class="primary" id="btnStart" onclick="start()">开始采集</button>
+      <button class="danger" id="btnStop" onclick="stop()" disabled>停止</button>
+      <button class="ghost" onclick="window.open('/api/download')">下载 JSONL</button>
+      <button class="ghost" onclick="window.open('/api/export.csv')">导出 CSV(Excel)</button>
+    </div>
+    <div class="bar"><div id="prog"></div></div>
+    <div class="stats">
+      <div class="stat"><div class="n" id="s-done">0</div><div class="l">已处理</div></div>
+      <div class="stat"><div class="n" id="s-total">0</div><div class="l">总计</div></div>
+      <div class="stat"><div class="n hit" id="s-hits">0</div><div class="l">命中</div></div>
+      <div class="stat"><div class="n miss" id="s-miss">0</div><div class="l">未命中</div></div>
+      <div class="stat"><div class="n invalid" id="s-invalid">0</div><div class="l">无效</div></div>
+      <div class="stat"><div class="n error" id="s-errors">0</div><div class="l">错误</div></div>
+      <div class="stat"><div class="n" id="s-skipped">0</div><div class="l">跳过</div></div>
+    </div>
+    <div class="hint" id="msg"></div>
+  </div>
+
+  <div class="card">
+    <h3>④ 实时结果(最近 60 条)</h3>
+    <div style="max-height:340px; overflow:auto">
+      <table>
+        <thead><tr><th>条码</th><th>品名</th><th>规格</th><th>单位</th><th>产地</th><th>进价</th><th>零售价</th><th>状态</th></tr></thead>
+        <tbody id="rows"></tbody>
+      </table>
+    </div>
+  </div>
+</div>
+
+<script>
+let mode = 'range';
+function setMode(m){
+  mode = m;
+  document.querySelectorAll('.tab').forEach(t=>t.classList.toggle('active', t.dataset.mode===m));
+  document.getElementById('pane-range').style.display = m==='range'?'block':'none';
+  document.getElementById('pane-list').style.display = m==='list'?'block':'none';
+}
+function fillFromPrefix(){
+  let p = document.getElementById('prefix').value.replace(/[^0-9]/g,'');
+  if(!p){ alert('请先填前缀'); return; }
+  if(p.length>=12){ alert('前缀太长,应少于 12 位'); return; }
+  let pad = 12 - p.length;
+  document.getElementById('start').value = p + '0'.repeat(pad);
+  document.getElementById('end').value   = p + '9'.repeat(pad);
+  updateEst();
+}
+function updateEst(){
+  let s = document.getElementById('start').value.replace(/[^0-9]/g,'');
+  let e = document.getElementById('end').value.replace(/[^0-9]/g,'');
+  if(s.length===12 && e.length===12){
+    let n = (BigInt(e) - BigInt(s)) + 1n;
+    document.getElementById('est').textContent = '本次将查询约 ' + n.toString() + ' 个条码';
+  } else {
+    document.getElementById('est').textContent = '';
+  }
+}
+document.getElementById('start').addEventListener('input', updateEst);
+document.getElementById('end').addEventListener('input', updateEst);
+updateEst();
+
+async function start(){
+  let body = {
+    mode: mode,
+    start_body: document.getElementById('start').value.trim(),
+    end_body: document.getElementById('end').value.trim(),
+    list: document.getElementById('list').value,
+    concurrency: parseInt(document.getElementById('concurrency').value)||3,
+    delay_ms: parseInt(document.getElementById('delay').value)||0,
+    log_miss: document.getElementById('logmiss').checked,
+    out_file: document.getElementById('outfile').value.trim()
+  };
+  let r = await fetch('/api/start', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(body)});
+  let j = await r.json();
+  if(j.error){ alert('启动失败: ' + j.error); return; }
+}
+async function stop(){ await fetch('/api/stop', {method:'POST'}); }
+
+function esc(s){ return (s||'').replace(/[&<>]/g, c=>({'&':'&amp;','<':'&lt;','>':'&gt;'}[c])); }
+
+async function poll(){
+  try{
+    let r = await fetch('/api/stats'); let j = await r.json();
+    let s = j.stats;
+    document.getElementById('s-done').textContent = s.done;
+    document.getElementById('s-total').textContent = s.total;
+    document.getElementById('s-hits').textContent = s.hits;
+    document.getElementById('s-miss').textContent = s.miss;
+    document.getElementById('s-invalid').textContent = s.invalid;
+    document.getElementById('s-errors').textContent = s.errors;
+    document.getElementById('s-skipped').textContent = s.skipped;
+    let pct = s.total>0 ? Math.floor(s.done*100/s.total) : 0;
+    document.getElementById('prog').style.width = pct + '%';
+    document.getElementById('msg').textContent = (s.running? ('采集中… 当前 '+s.current) : (s.message||'空闲'));
+    document.getElementById('btnStart').disabled = s.running;
+    document.getElementById('btnStop').disabled = !s.running;
+    let rows = (j.recent||[]).slice().reverse().map(p=>
+      '<tr><td>'+esc(p.barcode)+'</td><td>'+esc(p.name)+'</td><td>'+esc(p.spec)+'</td><td>'+esc(p.unit)+'</td><td>'+esc(p.area)+'</td><td>'+esc(p.in_price)+'</td><td>'+esc(p.sell_price)+'</td><td class="'+p.status+'">'+esc(p.status)+'</td></tr>'
+    ).join('');
+    document.getElementById('rows').innerHTML = rows;
+  }catch(e){}
+}
+setInterval(poll, 1000); poll();
+</script>
+</body>
+</html>

tools/bypos-collector/使用说明.md

@@ -0,0 +1,68 @@
+# 中心库商品采集器 bypos-collector 使用说明
+
+一个单文件 Windows 小程序,通过云店「新增商品」用到的同一个中心商品库
+(`zc.bypos.net`)按条码批量采集商品档案(品名/规格/单位/产地/厂商/建议进价/建议零售价),
+存到本地,供后续导入天工(goods)系统。
+
+## 一、运行
+
+1. 把 `bypos-collector.exe` 放到任意空文件夹(采集结果会生成在同一文件夹)。
+2. 双击运行。会弹出一个黑色命令行窗口(不要关它),并自动打开浏览器控制台
+   `http://127.0.0.1:8765/`。
+   - 若没自动打开,手动在浏览器输入上面这个地址。
+3. 用完直接关掉那个命令行窗口即可退出。
+
+## 二、采集
+
+控制台分四步:
+
+**① 规划采集范围** —— 两种方式二选一:
+- **按条码范围**:EAN-13 国标条码共 13 位,最后一位是校验位,程序自动算。
+  你只填**前 12 位**的起止区间。可在「快捷填充前缀」里填如 `690100`,
+  点按钮自动生成区间(`690100000000` ~ `690100999999`)。
+- **按条码清单**:直接粘贴一批条码(每行一个,或空格/逗号分隔)。
+
+**② 采集参数**:
+- 并发数(默认 3)、请求间隔(默认 300ms):**越慢越安全**,上游可能对账号限频。
+- 输出文件名(默认 `products.jsonl`)。
+- 「同时记录未命中/无效条码」:默认只存命中的;勾上会把未命中也记下来。
+
+**③ 运行**:点「开始采集」。进度、命中/未命中/错误实时显示。
+已经采过的条码会自动跳过(可随时停了再开,断点续采)。
+
+**④ 实时结果**:最近 60 条滚动显示。
+
+## 三、导出
+
+- 「下载 JSONL」:原始数据(每行一个 JSON),用于导入天工系统。
+- 「导出 CSV」:Excel 可直接打开查看。
+
+## 四、字段说明(JSONL 每行)
+
+| 字段 | 含义 |
+| --- | --- |
+| barcode | 条码(GTIN/EAN-13) |
+| name | 品名 |
+| spec | 规格 |
+| unit | 单位 |
+| area | 产地/地区 |
+| manufacturer | 生产企业(常为空) |
+| license | 生产许可(常为空) |
+| in_price | 建议进价 |
+| sell_price | 建议零售价 |
+| status | hit=命中 / miss=不存在 / invalid=非国标条码 / error=请求出错 |
+| fetched_at | 采集时间 |
+
+## 五、注意
+
+- 这是用云店账号授权去查上游中心库,**批量自动**比页面里一条条查更"重",
+  上游厂商可能对账号做频控/限额。请低速、分批("一点点采"),发现大量报错就降速。
+- 全量 69 段是个天文数字,不要无脑全跑;建议按你关心的品牌/品类前缀分批。
+
+## 六、命令行参数(可选)
+
+```
+bypos-collector.exe -addr 127.0.0.1:8765   # 改监听端口
+bypos-collector.exe -no-open               # 不自动开浏览器
+bypos-collector.exe -sdogid 137966         # 指定中心库账号 id
+```