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/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 7ecb619..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"
@@ -182,19 +183,22 @@ func (c *Collector) lookup(ctx context.Context, barcode string) (*Product, error
// ---- job / collector state ----
+const collectedFile = "collected.txt"
+
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"`
+ 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"`
+ HistoryCount int `json:"history_count"`
}
type Collector struct {
@@ -215,8 +219,9 @@ type Collector struct {
startedAt string
message string
- seen map[string]struct{} // barcodes already in output (dedupe / resume)
- recent []Product // ring of last results for UI
+ seen map[string]struct{} // barcodes already collected (dedupe)
+ collectedFp *os.File // persistent history file handle
+ recent []Product // ring of last results for UI
}
func NewCollector(sdogID string) *Collector {
@@ -227,6 +232,7 @@ func NewCollector(sdogID string) *Collector {
seen: map[string]struct{}{},
}
c.current.Store("")
+ c.loadCollected()
return c
}
@@ -238,20 +244,22 @@ func (c *Collector) snapshot() Stats {
msg := c.message
out := c.outPath
started := c.startedAt
+ hist := len(c.seen)
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,
+ 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,
+ HistoryCount: hist,
}
}
@@ -272,9 +280,24 @@ func (c *Collector) pushRecent(p Product) {
c.mu.Unlock()
}
-// loadSeen reads an existing output file to build the dedupe set (for resume).
+// loadCollected reads the persistent history file (one barcode per line).
+func (c *Collector) loadCollected() {
+ f, err := os.Open(collectedFile)
+ if err != nil {
+ return
+ }
+ defer f.Close()
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ bc := strings.TrimSpace(sc.Text())
+ if bc != "" {
+ c.seen[bc] = struct{}{}
+ }
+ }
+}
+
+// loadSeen reads an existing output file and adds to 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) {
@@ -387,6 +410,7 @@ func (c *Collector) Start(req JobReq) error {
}
c.outFile = f
c.outPath = abs
+ c.collectedFp, _ = os.OpenFile(collectedFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
// reset counters
atomic.StoreInt64(&c.total, int64(len(barcodes)))
@@ -424,6 +448,11 @@ func (c *Collector) run(ctx context.Context, barcodes []string, req JobReq) {
c.outFile.Close()
c.outFile = nil
}
+ if c.collectedFp != nil {
+ c.collectedFp.Sync()
+ c.collectedFp.Close()
+ c.collectedFp = nil
+ }
c.current.Store("")
}()
@@ -472,14 +501,17 @@ func (c *Collector) run(ctx context.Context, barcodes []string, req JobReq) {
}
atomic.AddInt64(&c.done, 1)
c.pushRecent(*p)
+ writeMu.Lock()
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()
}
+ c.seen[bc] = struct{}{}
+ if c.collectedFp != nil {
+ c.collectedFp.WriteString(bc + "\n")
+ }
+ writeMu.Unlock()
}
}
diff --git a/tools/bypos-collector/main.go b/tools/bypos-collector/main.go
index 009ebfd..469c9de 100644
--- a/tools/bypos-collector/main.go
+++ b/tools/bypos-collector/main.go
@@ -2,6 +2,7 @@ package main
import (
"bufio"
+ "bytes"
"embed"
"encoding/csv"
"encoding/json"
@@ -15,6 +16,7 @@ import (
"os"
"os/exec"
"runtime"
+ "strings"
"time"
)
@@ -70,6 +72,7 @@ func main() {
mux.HandleFunc("/api/stats", handleStats)
mux.HandleFunc("/api/download", handleDownload)
mux.HandleFunc("/api/export.csv", handleExportCSV)
+ mux.HandleFunc("/api/import", handleImport)
ln, err := net.Listen("tcp", *addr)
if err != nil {
@@ -168,6 +171,79 @@ func handleDownload(w http.ResponseWriter, r *http.Request) {
io.Copy(w, f)
}
+func handleImport(w http.ResponseWriter, r *http.Request) {
+ if r.Method != "POST" {
+ http.Error(w, "method", 405)
+ return
+ }
+ var req struct {
+ APIURL string `json:"api_url"`
+ Username string `json:"username"`
+ Password string `json:"password"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ writeJSON(w, 400, map[string]string{"error": "请求格式错误"})
+ return
+ }
+ req.APIURL = strings.TrimRight(req.APIURL, "/")
+ if req.APIURL == "" || req.Username == "" || req.Password == "" {
+ writeJSON(w, 400, map[string]string{"error": "请填写完整的 API 地址、用户名和密码"})
+ return
+ }
+
+ s := collector.snapshot()
+ if s.OutFile == "" {
+ writeJSON(w, 400, map[string]string{"error": "没有采集数据可导入"})
+ return
+ }
+
+ client := &http.Client{Timeout: 60 * time.Second}
+
+ loginBody, _ := json.Marshal(map[string]string{"username": req.Username, "password": req.Password})
+ loginResp, err := client.Post(req.APIURL+"/api/login", "application/json", bytes.NewReader(loginBody))
+ if err != nil {
+ writeJSON(w, 502, map[string]string{"error": "无法连接 Goods 系统: " + err.Error()})
+ return
+ }
+ defer loginResp.Body.Close()
+ var loginResult struct {
+ Token string `json:"token"`
+ Error *struct {
+ Message string `json:"message"`
+ } `json:"error"`
+ }
+ json.NewDecoder(loginResp.Body).Decode(&loginResult)
+ if loginResp.StatusCode != 200 || loginResult.Token == "" {
+ msg := "登录失败"
+ if loginResult.Error != nil {
+ msg = loginResult.Error.Message
+ }
+ writeJSON(w, 401, map[string]string{"error": msg})
+ return
+ }
+
+ jsonlData, err := os.ReadFile(s.OutFile)
+ if err != nil {
+ writeJSON(w, 500, map[string]string{"error": "读取采集文件失败: " + err.Error()})
+ return
+ }
+
+ importReq, _ := http.NewRequest("POST", req.APIURL+"/api/import/bypos", bytes.NewReader(jsonlData))
+ importReq.Header.Set("Content-Type", "application/x-ndjson")
+ importReq.Header.Set("Authorization", "Bearer "+loginResult.Token)
+ importResp, err := client.Do(importReq)
+ if err != nil {
+ writeJSON(w, 502, map[string]string{"error": "导入请求失败: " + err.Error()})
+ return
+ }
+ defer importResp.Body.Close()
+ respBody, _ := io.ReadAll(importResp.Body)
+
+ w.Header().Set("Content-Type", "application/json; charset=utf-8")
+ w.WriteHeader(importResp.StatusCode)
+ w.Write(respBody)
+}
+
func handleExportCSV(w http.ResponseWriter, r *http.Request) {
s := collector.snapshot()
if s.OutFile == "" {
diff --git a/tools/bypos-collector/web/index.html b/tools/bypos-collector/web/index.html
index 972db23..ec06654 100644
--- a/tools/bypos-collector/web/index.html
+++ b/tools/bypos-collector/web/index.html
@@ -37,6 +37,11 @@
.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; }
+ .history-badge { display:inline-block; background:#e8f0fe; color:#1f6feb; padding:4px 12px; border-radius:12px; font-size:13px; font-weight:600; margin-bottom:10px; }
+ .import-result { margin-top:10px; padding:10px; border-radius:6px; font-size:13px; }
+ .import-result.ok { background:#dcfce7; color:#166534; }
+ .import-result.fail { background:#fee2e2; color:#991b1b; }
+ input[type=password] { width:100%; padding:8px 10px; border:1px solid #cdd5e0; border-radius:6px; font-size:14px; }
@@ -97,11 +102,12 @@
- 速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(断点续采)。
+ 速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(跨次去重)。
③ 运行
+
历史已采集: 0 个条码
@@ -116,13 +122,36 @@
-
+
-
④ 实时结果(最近 60 条)
+
④ 导入到 Goods 系统
+
采集完成后,将数据一键导入到 OpenGoods 商品档案系统。
+
+
+
+
+
+
+
⑤ 实时结果(最近 60 条)
| 条码 | 品名 | 规格 | 单位 | 产地 | 进价 | 零售价 | 状态 |
|---|
@@ -196,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;
@@ -206,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';
+}