novaalphastrikeomegaz663
f04da0a135
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
"""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,
|
|
}
|