bypos-collector: 持久化采集记录去重 + Goods 系统批量导入 #24
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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/天工库
|
||||
|
||||
|
||||
Executable
BIN
Binary file not shown.
@@ -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)<string[^>]*>(.*)</string>`)
|
||||
|
||||
// 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<<uint(attempt)) * 500 * time.Millisecond
|
||||
select {
|
||||
case <-time.After(backoff):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil, lastErr
|
||||
}
|
||||
|
||||
// ---- 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 {
|
||||
@@ -182,20 +219,20 @@ 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 {
|
||||
if sdogID == "" {
|
||||
sdogID = defaultSdogID
|
||||
}
|
||||
sdogID = resolveSdogID(sdogID)
|
||||
c := &Collector{
|
||||
client: &http.Client{Timeout: 25 * time.Second},
|
||||
sdogID: sdogID,
|
||||
seen: map[string]struct{}{},
|
||||
}
|
||||
c.current.Store("")
|
||||
c.loadCollected()
|
||||
return c
|
||||
}
|
||||
|
||||
@@ -207,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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -241,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) {
|
||||
@@ -308,6 +362,9 @@ func (c *Collector) Start(req JobReq) error {
|
||||
if req.SdogID != "" {
|
||||
c.sdogID = req.SdogID
|
||||
}
|
||||
if c.sdogID == "" {
|
||||
return fmt.Errorf("未指定 sdogid，请通过控制台输入框、环境变量 $BYPOS_SDOGID 或 -sdogid 参数配置")
|
||||
}
|
||||
|
||||
// Build the list of barcodes to query.
|
||||
var barcodes []string
|
||||
@@ -353,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)))
|
||||
@@ -390,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("")
|
||||
}()
|
||||
|
||||
@@ -438,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()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEan13Check_valid(t *testing.T) {
|
||||
tests := []struct {
|
||||
body string
|
||||
want string
|
||||
}{
|
||||
{"692045990501", "6920459905012"},
|
||||
{"690100000000", "6901000000004"},
|
||||
{"690100000099", "6901000000998"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
got, ok := ean13Check(tt.body)
|
||||
if !ok {
|
||||
t.Errorf("ean13Check(%q) returned not ok", tt.body)
|
||||
continue
|
||||
}
|
||||
if got != tt.want {
|
||||
t.Errorf("ean13Check(%q) = %q, want %q", tt.body, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestEan13Check_invalid(t *testing.T) {
|
||||
cases := []string{"", "12345", "12345678901", "1234567890123", "69010000a000"}
|
||||
for _, body := range cases {
|
||||
_, ok := ean13Check(body)
|
||||
if ok {
|
||||
t.Errorf("ean13Check(%q) should return not ok", body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMd5hex(t *testing.T) {
|
||||
got := md5hex("137966")
|
||||
if len(got) != 32 {
|
||||
t.Errorf("md5hex should return 32-char hex, got len=%d", len(got))
|
||||
}
|
||||
if got != md5hex("137966") {
|
||||
t.Error("md5hex should be deterministic")
|
||||
}
|
||||
if got == md5hex("other") {
|
||||
t.Error("md5hex should differ for different inputs")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSanitizeBarcodes(t *testing.T) {
|
||||
got := sanitizeBarcodes("6920459905012\n6901028941068 123")
|
||||
want := []string{"6920459905012", "6901028941068", "123"}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("len = %d, want %d", len(got), len(want))
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Errorf("got[%d] = %q, want %q", i, got[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSanitizeBarcodes_empty(t *testing.T) {
|
||||
got := sanitizeBarcodes(" \n\t ")
|
||||
if len(got) != 0 {
|
||||
t.Errorf("expected empty, got %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSdogID_envOverride(t *testing.T) {
|
||||
os.Setenv("BYPOS_SDOGID", "999999")
|
||||
defer os.Unsetenv("BYPOS_SDOGID")
|
||||
got := resolveSdogID("111111")
|
||||
if got != "999999" {
|
||||
t.Errorf("expected env var override, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSdogID_fallback(t *testing.T) {
|
||||
os.Unsetenv("BYPOS_SDOGID")
|
||||
got := resolveSdogID("111111")
|
||||
if got != "111111" {
|
||||
t.Errorf("expected fallback to explicit, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveSdogID_empty(t *testing.T) {
|
||||
os.Unsetenv("BYPOS_SDOGID")
|
||||
got := resolveSdogID("")
|
||||
if got != "" {
|
||||
t.Errorf("expected empty, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewCollector_sdogFromEnv(t *testing.T) {
|
||||
os.Setenv("BYPOS_SDOGID", "888888")
|
||||
defer os.Unsetenv("BYPOS_SDOGID")
|
||||
c := NewCollector("")
|
||||
if c.sdogID != "888888" {
|
||||
t.Errorf("expected sdogID from env, got %q", c.sdogID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartRejectsEmptySdogID(t *testing.T) {
|
||||
os.Unsetenv("BYPOS_SDOGID")
|
||||
c := NewCollector("")
|
||||
err := c.Start(JobReq{Mode: "list", List: "6920459905012"})
|
||||
if err == nil {
|
||||
t.Fatal("expected error for empty sdogid")
|
||||
}
|
||||
}
|
||||
+129
-12
@@ -1,6 +1,8 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"embed"
|
||||
"encoding/csv"
|
||||
"encoding/json"
|
||||
@@ -14,24 +16,55 @@ import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"runtime"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
//go:embed web/*
|
||||
var webFS embed.FS
|
||||
|
||||
var collector = NewCollector("")
|
||||
var collector *Collector
|
||||
|
||||
// waitExit keeps the console window open on Windows so the user can read errors.
|
||||
func waitExit() {
|
||||
if runtime.GOOS == "windows" {
|
||||
fmt.Println("\n按回车键退出...")
|
||||
bufio.NewReader(os.Stdin).ReadBytes('\n')
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
// Log to file so crashes are diagnosable even if console closes.
|
||||
lf, lfErr := os.OpenFile("bypos-collector.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
|
||||
if lfErr == nil {
|
||||
log.SetOutput(io.MultiWriter(os.Stderr, lf))
|
||||
defer lf.Close()
|
||||
}
|
||||
|
||||
sub, _ := fs.Sub(webFS, "web")
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("程序崩溃: %v", r)
|
||||
waitExit()
|
||||
}
|
||||
}()
|
||||
|
||||
log.Println("bypos-collector 启动中...")
|
||||
|
||||
addr := flag.String("addr", "127.0.0.1:8765", "本地监听地址")
|
||||
noOpen := flag.Bool("no-open", false, "不自动打开浏览器")
|
||||
sdog := flag.String("sdogid", "", "中心库账号 id（优先读 $BYPOS_SDOGID 环境变量）")
|
||||
flag.Parse()
|
||||
collector = NewCollector(*sdog)
|
||||
if collector.sdogID == "" {
|
||||
log.Println("警告: 未配置 sdogid，请通过 $BYPOS_SDOGID 环境变量、-sdogid 参数或控制台输入框指定")
|
||||
}
|
||||
|
||||
sub, err := fs.Sub(webFS, "web")
|
||||
if err != nil {
|
||||
log.Printf("错误: 无法加载内嵌 web 资源: %v", err)
|
||||
waitExit()
|
||||
return
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/", http.FileServer(http.FS(sub)))
|
||||
mux.HandleFunc("/api/start", handleStart)
|
||||
@@ -39,10 +72,17 @@ 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 {
|
||||
log.Fatalf("无法监听 %s: %v", *addr, err)
|
||||
log.Printf("端口 %s 被占用，自动选择可用端口...", *addr)
|
||||
ln, err = net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
log.Printf("错误: 无法监听: %v", err)
|
||||
waitExit()
|
||||
return
|
||||
}
|
||||
}
|
||||
realAddr := ln.Addr().String()
|
||||
urlStr := "http://" + realAddr + "/"
|
||||
@@ -51,10 +91,14 @@ func main() {
|
||||
fmt.Println(" 控制台: " + urlStr)
|
||||
fmt.Println(" 关闭本窗口即停止程序")
|
||||
fmt.Println("==============================================")
|
||||
log.Printf("监听地址: %s", realAddr)
|
||||
if !*noOpen {
|
||||
go openBrowser(urlStr)
|
||||
}
|
||||
log.Fatal(http.Serve(ln, mux))
|
||||
if err := http.Serve(ln, mux); err != nil {
|
||||
log.Printf("HTTP 服务异常退出: %v", err)
|
||||
waitExit()
|
||||
}
|
||||
}
|
||||
|
||||
func openBrowser(url string) {
|
||||
@@ -127,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 == "" {
|
||||
@@ -143,14 +260,14 @@ func handleExportCSV(w http.ResponseWriter, r *http.Request) {
|
||||
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"})
|
||||
cw.Write([]string{"barcode", "name", "spec", "unit", "area", "manufacturer", "license", "in_price", "sell_price", "status", "retmsg", "fetched_at", "source"})
|
||||
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.Write([]string{p.Barcode, p.Name, p.Spec, p.Unit, p.Area, p.Manufacturer, p.License, p.InPrice, p.SellPrice, p.Status, p.RetMsg, p.FetchedAt, p.Source})
|
||||
}
|
||||
cw.Flush()
|
||||
}
|
||||
|
||||
@@ -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; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
@@ -79,6 +84,10 @@
|
||||
<div class="card">
|
||||
<h3>② 采集参数</h3>
|
||||
<div class="row">
|
||||
<div>
|
||||
<label>中心库账号 sdogid</label>
|
||||
<input type="text" id="sdogid" placeholder="留空则使用环境变量或启动参数"/>
|
||||
</div>
|
||||
<div>
|
||||
<label>并发数</label>
|
||||
<input type="number" id="concurrency" value="3" min="1" max="20"/>
|
||||
@@ -93,11 +102,12 @@
|
||||
</div>
|
||||
</div>
|
||||
<label style="margin-top:12px"><input type="checkbox" id="logmiss"/> 同时记录未命中/无效条码(默认只存命中)</label>
|
||||
<div class="hint">速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(断点续采)。</div>
|
||||
<div class="hint">速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(跨次去重)。</div>
|
||||
</div>
|
||||
|
||||
<div class="card">
|
||||
<h3>③ 运行</h3>
|
||||
<div class="history-badge" id="history-badge">历史已采集: 0 个条码</div>
|
||||
<div style="margin-bottom:12px">
|
||||
<button class="primary" id="btnStart" onclick="start()">开始采集</button>
|
||||
<button class="danger" id="btnStop" onclick="stop()" disabled>停止</button>
|
||||
@@ -112,13 +122,36 @@
|
||||
<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 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>
|
||||
<h3>④ 导入到 Goods 系统</h3>
|
||||
<div class="hint" style="margin-top:0;margin-bottom:10px">采集完成后,将数据一键导入到 OpenGoods 商品档案系统。</div>
|
||||
<div class="row">
|
||||
<div>
|
||||
<label>Goods 管理后台地址</label>
|
||||
<input type="text" id="import-url" placeholder="如 http://192.168.1.100:8080/ping"/>
|
||||
</div>
|
||||
<div>
|
||||
<label>管理员用户名</label>
|
||||
<input type="text" id="import-user" placeholder="admin"/>
|
||||
</div>
|
||||
<div>
|
||||
<label>管理员密码</label>
|
||||
<input type="password" id="import-pass"/>
|
||||
</div>
|
||||
</div>
|
||||
<div style="margin-top:12px">
|
||||
<button class="primary" id="btnImport" onclick="doImport()">导入到 Goods</button>
|
||||
</div>
|
||||
<div id="import-result"></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>
|
||||
@@ -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 = '<div class="import-result fail">导入失败: '+esc(j.error.message||j.error)+'</div>';
|
||||
} else {
|
||||
res.innerHTML = '<div class="import-result ok">导入完成: 成功 '+j.loaded+' 条, 跳过 '+j.skipped+' 条, 失败 '+j.errored+' 条</div>';
|
||||
}
|
||||
} catch(e){
|
||||
res.innerHTML = '<div class="import-result fail">导入失败: '+esc(e.message)+'</div>';
|
||||
}
|
||||
btn.disabled = false;
|
||||
btn.textContent = '导入到 Goods';
|
||||
}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
|
||||
Reference in New Issue
Block a user