diff --git a/api/internal/adminhandler/handler.go b/api/internal/adminhandler/handler.go index 5ab5ed7..475225a 100644 --- a/api/internal/adminhandler/handler.go +++ b/api/internal/adminhandler/handler.go @@ -90,6 +90,8 @@ func (h *Handler) Router() http.Handler { r.Put("/api/categories/{id}", h.UpdateCategory) r.Delete("/api/categories/{id}", h.DeleteCategory) + r.Post("/api/import/bypos", h.ImportBypos) + r.Get("/api/submissions", h.ListSubmissions) r.Get("/api/submissions/{id}", h.GetSubmission) r.Post("/api/submissions/{id}/approve", h.ApproveSubmission) diff --git a/api/internal/adminhandler/import_bypos.go b/api/internal/adminhandler/import_bypos.go new file mode 100644 index 0000000..b70c7fa --- /dev/null +++ b/api/internal/adminhandler/import_bypos.go @@ -0,0 +1,30 @@ +package adminhandler + +import ( + "encoding/json" + "net/http" + + "github.com/baicai2026-baicai/goods/api/internal/adminstore" +) + +func (h *Handler) ImportBypos(w http.ResponseWriter, r *http.Request) { + var records []adminstore.ByposRecord + dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 50<<20)) + for dec.More() { + var rec adminstore.ByposRecord + if err := dec.Decode(&rec); err != nil { + continue + } + records = append(records, rec) + } + if len(records) == 0 { + writeError(w, http.StatusBadRequest, "empty", "\u6ca1\u6709\u53ef\u5bfc\u5165\u7684\u8bb0\u5f55") + return + } + result, err := h.store.ImportByposRecords(r.Context(), records) + if err != nil { + writeError(w, http.StatusInternalServerError, "import_error", err.Error()) + return + } + writeJSON(w, http.StatusOK, result) +} diff --git a/api/internal/adminstore/import_bypos.go b/api/internal/adminstore/import_bypos.go new file mode 100644 index 0000000..26e5e9b --- /dev/null +++ b/api/internal/adminstore/import_bypos.go @@ -0,0 +1,217 @@ +package adminstore + +import ( + "context" + "encoding/json" + "strings" + + "github.com/jackc/pgx/v5" +) + +type ByposRecord struct { + Barcode string `json:"barcode"` + Name string `json:"name"` + Spec string `json:"spec"` + Unit string `json:"unit"` + Area string `json:"area"` + Manufacturer string `json:"manufacturer"` + License string `json:"license"` + InPrice string `json:"in_price"` + SellPrice string `json:"sell_price"` + Status string `json:"status"` + RetMsg string `json:"retmsg"` + FetchedAt string `json:"fetched_at"` + Source string `json:"source"` +} + +type ImportByposResult struct { + Loaded int `json:"loaded"` + Skipped int `json:"skipped"` + Errored int `json:"errored"` +} + +const byposSourceName = "bypos\u4e2d\u5fc3\u5e93" +const byposSourceURL = "https://zc.bypos.net" + +func (s *Store) ensureByposSource(ctx context.Context, tx pgx.Tx) (string, error) { + var id string + err := tx.QueryRow(ctx, ` +INSERT INTO source (name, homepage, license, trust_weight) +VALUES ($1, $2, 'proprietary', 0.6) +ON CONFLICT (name) DO UPDATE SET homepage = EXCLUDED.homepage +RETURNING id`, byposSourceName, byposSourceURL).Scan(&id) + return id, err +} + +func (s *Store) ensureManufacturer(ctx context.Context, tx pgx.Tx, name string, country *string) (string, error) { + norm := strings.Join(strings.Fields(strings.ToLower(name)), " ") + var id string + err := tx.QueryRow(ctx, ` +INSERT INTO manufacturer (name, normalized_name, country) +VALUES ($1, $2, $3) +ON CONFLICT (normalized_name) DO UPDATE SET name = manufacturer.name +RETURNING id`, name, norm, country).Scan(&id) + return id, err +} + +func (s *Store) importByposRecord(ctx context.Context, tx pgx.Tx, rec ByposRecord, sourceID string) error { + if rec.Status != "hit" || strings.TrimSpace(rec.Name) == "" { + return nil + } + + barcode := strings.TrimSpace(rec.Barcode) + name := strings.TrimSpace(rec.Name) + + var country *string + if strings.HasPrefix(barcode, "69") { + c := "\u4e2d\u56fd" + country = &c + } + + var manufacturerID *string + if mfr := strings.TrimSpace(rec.Manufacturer); mfr != "" { + mid, err := s.ensureManufacturer(ctx, tx, mfr, country) + if err != nil { + return err + } + manufacturerID = &mid + } + + attrs := map[string]interface{}{} + if v := strings.TrimSpace(rec.Spec); v != "" { + attrs["spec"] = v + } + if v := strings.TrimSpace(rec.Unit); v != "" { + attrs["pack_unit"] = v + } + if v := strings.TrimSpace(rec.Area); v != "" { + attrs["origin_area"] = v + } + if v := strings.TrimSpace(rec.License); v != "" { + attrs["production_license"] = v + } + attrsJSON, _ := json.Marshal(attrs) + + var productID string + if barcode != "" { + err := tx.QueryRow(ctx, ` +INSERT INTO product (gtin, name, manufacturer_id, country_of_origin, attributes, status) +VALUES ($1, $2, $3, $4, $5, 'active') +ON CONFLICT (gtin) WHERE gtin IS NOT NULL DO UPDATE SET + name = EXCLUDED.name, + manufacturer_id = COALESCE(EXCLUDED.manufacturer_id, product.manufacturer_id), + country_of_origin = COALESCE(EXCLUDED.country_of_origin, product.country_of_origin), + attributes = product.attributes || EXCLUDED.attributes +RETURNING id`, barcode, name, manufacturerID, country, attrsJSON).Scan(&productID) + if err != nil { + return err + } + } else { + err := tx.QueryRow(ctx, ` +INSERT INTO product (name, manufacturer_id, country_of_origin, attributes, status) +VALUES ($1, $2, $3, $4, 'active') +RETURNING id`, name, manufacturerID, country, attrsJSON).Scan(&productID) + if err != nil { + return err + } + } + + _, _ = tx.Exec(ctx, "DELETE FROM product_msrp WHERE product_id = $1 AND source_id = $2", productID, sourceID) + if sp := strings.TrimSpace(rec.SellPrice); sp != "" && sp != "0" { + _, _ = tx.Exec(ctx, ` +INSERT INTO product_msrp (product_id, amount, currency, region, source_id, source_url) +VALUES ($1, $2::numeric, 'CNY', 'CN', $3, $4)`, productID, sp, sourceID, byposSourceURL) + } + + _, _ = tx.Exec(ctx, "DELETE FROM product_source WHERE product_id = $1 AND source_id = $2", productID, sourceID) + fields := []string{"name", "country_of_origin"} + if manufacturerID != nil { + fields = append(fields, "manufacturer") + } + if len(attrs) > 0 { + fields = append(fields, "attributes") + } + if barcode != "" { + fields = append(fields, "gtin") + } + rawJSON, _ := json.Marshal(rec) + fetchedAt := strings.TrimSpace(rec.FetchedAt) + if fetchedAt == "" { + fetchedAt = "" + } + + if fetchedAt != "" { + _, err := tx.Exec(ctx, ` +INSERT INTO product_source (product_id, source_id, url, fields, fetched_at, raw) +VALUES ($1, $2, $3, $4, $5::timestamptz, $6)`, + productID, sourceID, byposSourceURL, fields, fetchedAt, rawJSON) + if err != nil { + return err + } + } else { + _, err := tx.Exec(ctx, ` +INSERT INTO product_source (product_id, source_id, url, fields, fetched_at, raw) +VALUES ($1, $2, $3, $4, now(), $5)`, + productID, sourceID, byposSourceURL, fields, rawJSON) + if err != nil { + return err + } + } + + if barcode != "" { + gtinType := "EAN13" + switch len(barcode) { + case 8: + gtinType = "EAN8" + case 12: + gtinType = "UPC" + case 14: + gtinType = "GTIN14" + } + _, _ = tx.Exec(ctx, ` +INSERT INTO product_barcode (product_id, gtin, gtin_type, pack_level, is_primary, source_id) +VALUES ($1, $2, $3, 'each', true, $4) +ON CONFLICT (gtin) DO NOTHING`, productID, barcode, gtinType, sourceID) + } + + _, err := s.recomputeQualityTx(ctx, tx, productID) + return err +} + +func (s *Store) ImportByposRecords(ctx context.Context, records []ByposRecord) (*ImportByposResult, error) { + result := &ImportByposResult{} + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + sourceID, err := s.ensureByposSource(ctx, tx) + if err != nil { + return nil, err + } + + for _, rec := range records { + if rec.Status != "hit" || strings.TrimSpace(rec.Name) == "" { + result.Skipped++ + continue + } + sp, spErr := tx.Begin(ctx) + if spErr != nil { + result.Errored++ + continue + } + if err := s.importByposRecord(ctx, sp, rec, sourceID); err != nil { + sp.Rollback(ctx) + result.Errored++ + } else { + sp.Commit(ctx) + result.Loaded++ + } + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return result, nil +} diff --git a/tools/bypos-collector/README.md b/tools/bypos-collector/README.md index 6039816..34c5b8d 100644 --- a/tools/bypos-collector/README.md +++ b/tools/bypos-collector/README.md @@ -58,9 +58,10 @@ GET http://zc.bypos.net/byGoodsService/byMessage.asmx/GetGoodsinfo ## 配置 -- `sdogid`:中心库账号 id(本项目所属云店账号的授权 id)。默认值见 - `collect.go` 的 `defaultSdogID`,也可用 `-sdogid` 参数或控制台覆盖。 - **这是账号级凭证**——若本仓库对外公开,建议改为从环境变量/外部配置读取。 +- `sdogid`：中心库账号 id（本项目所属云店账号的授权 id）。 + 优先从环境变量 `BYPOS_SDOGID` 读取，其次是命令行 `-sdogid` 参数， + 也可以在 Web 控制台的输入框中填写。**三者都未设时启动会警告、开始采集时会报错。** +- 网络错误自动重试（最多 3 次，指数退避 500ms/1s/2s）。 ## 导入 goods/天工库 diff --git a/tools/bypos-collector/bypos-collector.exe b/tools/bypos-collector/bypos-collector.exe new file mode 100755 index 0000000..12c37f8 Binary files /dev/null and b/tools/bypos-collector/bypos-collector.exe differ diff --git a/tools/bypos-collector/collect.go b/tools/bypos-collector/collect.go index ebe6672..3583f87 100644 --- a/tools/bypos-collector/collect.go +++ b/tools/bypos-collector/collect.go @@ -1,6 +1,7 @@ package main import ( + "bufio" "context" "crypto/md5" "encoding/hex" @@ -20,12 +21,21 @@ import ( ) // ---- 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" +// maxRetries is the number of retry attempts for transient network errors. +const maxRetries = 3 + +// resolveSdogID reads the account id from $BYPOS_SDOGID, falling back to the +// explicit argument (CLI flag or Web UI input). Returns empty if neither set. +func resolveSdogID(explicit string) string { + if v := os.Getenv("BYPOS_SDOGID"); v != "" { + return v + } + return explicit +} + var stringTagRe = regexp.MustCompile(`(?s)]*>(.*)`) // Product is the normalized record we persist (one JSON object per line). @@ -87,8 +97,8 @@ func ean13Check(body string) (string, bool) { 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) { +// lookupOnce performs a single HTTP request to the upstream central library. +func (c *Collector) lookupOnce(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) @@ -147,21 +157,48 @@ func (c *Collector) lookup(ctx context.Context, barcode string) (*Product, error return p, nil } +// lookup queries the upstream with up to maxRetries retries on transient errors. +func (c *Collector) lookup(ctx context.Context, barcode string) (*Product, error) { + var lastErr error + for attempt := 0; attempt <= maxRetries; attempt++ { + if ctx.Err() != nil { + return nil, ctx.Err() + } + p, err := c.lookupOnce(ctx, barcode) + if err == nil { + return p, nil + } + lastErr = err + if attempt < maxRetries { + backoff := time.Duration(1< @@ -79,6 +84,10 @@

② 采集参数

+
+ + +
@@ -93,11 +102,12 @@
-
速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(断点续采)。
+
速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(跨次去重)。

③ 运行

+
历史已采集: 0 个条码
@@ -112,13 +122,36 @@
0
未命中
0
无效
0
错误
-
0
跳过
+
0
跳过(已采)
-

④ 实时结果(最近 60 条)

+

④ 导入到 Goods 系统

+
采集完成后,将数据一键导入到 OpenGoods 商品档案系统。
+
+
+ + +
+
+ + +
+
+ + +
+
+
+ +
+
+
+ +
+

⑤ 实时结果(最近 60 条)

@@ -165,6 +198,7 @@ async function start(){ start_body: document.getElementById('start').value.trim(), end_body: document.getElementById('end').value.trim(), list: document.getElementById('list').value, + sdog_id: document.getElementById('sdogid').value.trim(), concurrency: parseInt(document.getElementById('concurrency').value)||3, delay_ms: parseInt(document.getElementById('delay').value)||0, log_miss: document.getElementById('logmiss').checked, @@ -191,6 +225,7 @@ async function poll(){ 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('history-badge').textContent = '历史已采集: ' + (s.history_count||0) + ' 个条码'; document.getElementById('msg').textContent = (s.running? ('采集中… 当前 '+s.current) : (s.message||'空闲')); document.getElementById('btnStart').disabled = s.running; document.getElementById('btnStop').disabled = !s.running; @@ -201,6 +236,35 @@ async function poll(){ }catch(e){} } setInterval(poll, 1000); poll(); + +async function doImport(){ + let url = document.getElementById('import-url').value.trim(); + let user = document.getElementById('import-user').value.trim(); + let pass = document.getElementById('import-pass').value; + if(!url || !user || !pass){ alert('请填写完整的 Goods 系统地址、用户名和密码'); return; } + let btn = document.getElementById('btnImport'); + let res = document.getElementById('import-result'); + btn.disabled = true; + btn.textContent = '导入中...'; + res.innerHTML = ''; + try { + let r = await fetch('/api/import', { + method:'POST', + headers:{'Content-Type':'application/json'}, + body:JSON.stringify({api_url:url, username:user, password:pass}) + }); + let j = await r.json(); + if(j.error){ + res.innerHTML = '
导入失败: '+esc(j.error.message||j.error)+'
'; + } else { + res.innerHTML = '
导入完成: 成功 '+j.loaded+' 条, 跳过 '+j.skipped+' 条, 失败 '+j.errored+' 条
'; + } + } catch(e){ + res.innerHTML = '
导入失败: '+esc(e.message)+'
'; + } + btn.disabled = false; + btn.textContent = '导入到 Goods'; +}
条码品名规格单位产地进价零售价状态