feat(admin): add /api/import/bypos endpoint for bypos-collector import
Brings the bypos JSONL import backend (handler + store) into main so the bypos-collector tool's import feature works end-to-end. Records are upserted on GTIN with manufacturer/MSRP/barcode/source provenance, quality recomputed per product. Only status=hit records with a name are imported. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user