feat: 实现商品目录批量导入接口 (#133)
This commit is contained in:
chengma
@@ -0,0 +1,295 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"cmautobuy/admin/model"
|
||||
"cmautobuy/admin/repository"
|
||||
)
|
||||
|
||||
const (
|
||||
CatalogMaxProducts = 500
|
||||
CatalogMaxSKUs = 5000
|
||||
)
|
||||
|
||||
type CatalogShopeeProduct struct {
|
||||
GoodsID string `json:"goods_id"`
|
||||
Title string `json:"title"`
|
||||
Status string `json:"status"`
|
||||
MainSKUCode string `json:"main_sku_code"`
|
||||
}
|
||||
type CatalogShopeeSKU struct {
|
||||
SKUID string `json:"sku_id"`
|
||||
GoodsID string `json:"goods_id"`
|
||||
SpecRaw string `json:"spec_raw"`
|
||||
Color string `json:"color"`
|
||||
Size string `json:"size"`
|
||||
Advice string `json:"advice"`
|
||||
ParseOK bool `json:"parse_ok"`
|
||||
SKUCode string `json:"sku_code"`
|
||||
}
|
||||
type CatalogPddDimension struct {
|
||||
Key string `json:"key"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
type CatalogPddSKU struct {
|
||||
Options map[string]string `json:"options"`
|
||||
PriceCent *int64 `json:"price_cent"`
|
||||
ListPriceCent *int64 `json:"list_price_cent"`
|
||||
Available bool `json:"available"`
|
||||
}
|
||||
type CatalogPddProduct struct {
|
||||
GoodsID string `json:"goods_id"`
|
||||
URL string `json:"url"`
|
||||
Title string `json:"title"`
|
||||
ShopName string `json:"shop_name"`
|
||||
Dimensions []CatalogPddDimension `json:"dimensions"`
|
||||
SKUs []CatalogPddSKU `json:"skus"`
|
||||
}
|
||||
type CatalogAssociation struct {
|
||||
ShopeeGoodsID string `json:"shopee_goods_id"`
|
||||
PddGoodsID string `json:"pdd_goods_id"`
|
||||
}
|
||||
type CatalogBatchRequest struct {
|
||||
SchemaVersion int `json:"schema_version"`
|
||||
BatchID string `json:"batch_id"`
|
||||
ObservedAt string `json:"observed_at"`
|
||||
ShopeeProducts []CatalogShopeeProduct `json:"shopee_products"`
|
||||
ShopeeSKUs []CatalogShopeeSKU `json:"shopee_skus"`
|
||||
PddProducts []CatalogPddProduct `json:"pdd_products"`
|
||||
Associations []CatalogAssociation `json:"associations"`
|
||||
}
|
||||
type CatalogCounts struct {
|
||||
ShopeeCreated int `json:"shopee_created"`
|
||||
ShopeeUpdated int `json:"shopee_updated"`
|
||||
SKUCreated int `json:"sku_created"`
|
||||
SKUUpdated int `json:"sku_updated"`
|
||||
PddCreated int `json:"pdd_created"`
|
||||
PddUpdated int `json:"pdd_updated"`
|
||||
AssociationCreated int `json:"association_created"`
|
||||
AssociationUnchanged int `json:"association_unchanged"`
|
||||
}
|
||||
type CatalogBatchResponse struct {
|
||||
BatchID string `json:"batch_id"`
|
||||
Status string `json:"status"`
|
||||
Replayed bool `json:"replayed"`
|
||||
Counts CatalogCounts `json:"counts"`
|
||||
FinishedAt string `json:"finished_at"`
|
||||
}
|
||||
|
||||
type CatalogError struct {
|
||||
Status int
|
||||
Code, Message string
|
||||
Retryable bool
|
||||
Details any
|
||||
}
|
||||
|
||||
func (e *CatalogError) Error() string { return e.Message }
|
||||
func catalogInvalid(code, message string, details any) error {
|
||||
return &CatalogError{Status: http.StatusUnprocessableEntity, Code: code, Message: message, Details: details}
|
||||
}
|
||||
|
||||
func ValidateCatalogBatch(req CatalogBatchRequest) error {
|
||||
if req.SchemaVersion != 1 {
|
||||
return catalogInvalid("UNSUPPORTED_SCHEMA", "schema_version 只支持 1", nil)
|
||||
}
|
||||
if len(req.BatchID) < 1 || len(req.BatchID) > 191 {
|
||||
return catalogInvalid("INVALID_BATCH_ID", "batch_id 长度必须为 1-191", nil)
|
||||
}
|
||||
if _, err := time.Parse(time.RFC3339Nano, req.ObservedAt); err != nil {
|
||||
return catalogInvalid("INVALID_OBSERVED_AT", "observed_at 必须是带时区 ISO 8601", nil)
|
||||
}
|
||||
if len(req.ShopeeProducts)+len(req.PddProducts) > CatalogMaxProducts {
|
||||
return catalogInvalid("TOO_MANY_PRODUCTS", "每批商品总数不能超过 500", nil)
|
||||
}
|
||||
if len(req.ShopeeSKUs) > CatalogMaxSKUs {
|
||||
return catalogInvalid("TOO_MANY_SKUS", "每批 SKU 不能超过 5000", nil)
|
||||
}
|
||||
seenProducts, seenSKUs, seenPDD := map[string]bool{}, map[string]bool{}, map[string]bool{}
|
||||
for i, p := range req.ShopeeProducts {
|
||||
p.GoodsID = strings.TrimSpace(p.GoodsID)
|
||||
if p.GoodsID == "" || strings.TrimSpace(p.Title) == "" || seenProducts[p.GoodsID] {
|
||||
return catalogInvalid("INVALID_SHOPEE_PRODUCT", "蝦皮商品 ID/标题不能为空且批内不能重复", map[string]any{"index": i, "goods_id": p.GoodsID})
|
||||
}
|
||||
seenProducts[p.GoodsID] = true
|
||||
}
|
||||
for i, s := range req.ShopeeSKUs {
|
||||
if strings.TrimSpace(s.SKUID) == "" || strings.TrimSpace(s.GoodsID) == "" || strings.TrimSpace(s.SpecRaw) == "" || seenSKUs[s.SKUID] {
|
||||
return catalogInvalid("INVALID_SHOPEE_SKU", "蝦皮 SKU ID、商品 ID、spec_raw 必填且 SKU 不能重复", map[string]any{"index": i, "sku_id": s.SKUID})
|
||||
}
|
||||
seenSKUs[s.SKUID] = true
|
||||
}
|
||||
for i, p := range req.PddProducts {
|
||||
if strings.TrimSpace(p.GoodsID) == "" || strings.TrimSpace(p.URL) == "" || seenPDD[p.GoodsID] {
|
||||
return catalogInvalid("INVALID_PDD_PRODUCT", "PDD 商品 ID/URL 必填且批内不能重复", map[string]any{"index": i, "goods_id": p.GoodsID})
|
||||
}
|
||||
seenPDD[p.GoodsID] = true
|
||||
for j, s := range p.SKUs {
|
||||
if len(s.Options) == 0 || (s.PriceCent != nil && *s.PriceCent < 0) || (s.ListPriceCent != nil && *s.ListPriceCent < 0) {
|
||||
return catalogInvalid("INVALID_PDD_SKU", "PDD 规格选项不能为空且金额不能为负", map[string]any{"product_index": i, "sku_index": j})
|
||||
}
|
||||
}
|
||||
}
|
||||
for i, a := range req.Associations {
|
||||
if strings.TrimSpace(a.ShopeeGoodsID) == "" || strings.TrimSpace(a.PddGoodsID) == "" {
|
||||
return catalogInvalid("INVALID_ASSOCIATION", "关联双方商品 ID 必填", map[string]any{"index": i})
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func ImportCatalogBatch(db *sql.DB, source string, req CatalogBatchRequest, raw []byte) (CatalogBatchResponse, error) {
|
||||
if err := ValidateCatalogBatch(req); err != nil {
|
||||
return CatalogBatchResponse{}, err
|
||||
}
|
||||
observed, _ := time.Parse(time.RFC3339Nano, req.ObservedAt)
|
||||
req.ObservedAt = observed.UTC().Format(model.TimeLayout)
|
||||
h := sha256.Sum256(raw)
|
||||
hash := hex.EncodeToString(h[:])
|
||||
now := model.NowISO()
|
||||
tx, err := db.BeginTx(context.Background(), &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||||
if err != nil {
|
||||
return CatalogBatchResponse{}, err
|
||||
}
|
||||
run := model.CatalogImportRun{Source: source, BatchID: req.BatchID, RequestHash: hash, Status: model.CatalogImportProcessing, ObservedAt: req.ObservedAt, LastRequestAt: now, CreatedAt: now}
|
||||
if err = repository.InsertCatalogImportRun(tx, run); err != nil {
|
||||
tx.Rollback()
|
||||
if errors.Is(err, repository.ErrCatalogImportRunExists) {
|
||||
return replayCatalogBatch(db, source, req.BatchID, hash, now)
|
||||
}
|
||||
return CatalogBatchResponse{}, err
|
||||
}
|
||||
counts := CatalogCounts{}
|
||||
fail := func(importErr error) (CatalogBatchResponse, error) {
|
||||
tx.Rollback()
|
||||
run.Status = model.CatalogImportFailed
|
||||
run.FailureCount = 1
|
||||
run.ErrorSummary = truncateCatalogError(importErr.Error())
|
||||
var catalogErr *CatalogError
|
||||
if errors.As(importErr, &catalogErr) {
|
||||
stored, _ := json.Marshal(catalogErr)
|
||||
run.ResponseBody = string(stored)
|
||||
}
|
||||
run.FinishedAt = model.NowISO()
|
||||
if insertErr := repository.InsertCatalogImportRun(db, run); insertErr == nil {
|
||||
_ = repository.CompleteCatalogImportRun(db, run)
|
||||
}
|
||||
return CatalogBatchResponse{}, importErr
|
||||
}
|
||||
for _, p := range req.ShopeeProducts {
|
||||
c, u, e := repository.UpsertCatalogShopeeProduct(tx, p.GoodsID, p.Title, p.Status, p.MainSKUCode, req.ObservedAt, now)
|
||||
if e != nil {
|
||||
return fail(e)
|
||||
}
|
||||
if c {
|
||||
counts.ShopeeCreated++
|
||||
}
|
||||
if u {
|
||||
counts.ShopeeUpdated++
|
||||
}
|
||||
}
|
||||
for _, p := range req.PddProducts {
|
||||
var skus string
|
||||
if len(p.SKUs) > 0 {
|
||||
b, _ := json.Marshal(map[string]any{"schema_version": 1, "goods_id": p.GoodsID, "title": p.Title, "shop_name": p.ShopName, "dimensions": p.Dimensions, "skus": p.SKUs})
|
||||
skus = string(b)
|
||||
}
|
||||
c, u, e := repository.UpsertCatalogPddProduct(tx, p.GoodsID, p.URL, p.Title, p.ShopName, skus, req.ObservedAt, now)
|
||||
if e != nil {
|
||||
return fail(e)
|
||||
}
|
||||
if c {
|
||||
counts.PddCreated++
|
||||
}
|
||||
if u {
|
||||
counts.PddUpdated++
|
||||
}
|
||||
}
|
||||
for _, s := range req.ShopeeSKUs {
|
||||
c, u, e := repository.UpsertCatalogShopeeSKU(tx, s.SKUID, s.GoodsID, s.SpecRaw, s.Color, s.Size, s.Advice, s.ParseOK, s.SKUCode, req.ObservedAt, now)
|
||||
if e != nil {
|
||||
return fail(&CatalogError{Status: 409, Code: "SKU_OWNERSHIP_CONFLICT", Message: e.Error()})
|
||||
}
|
||||
if c {
|
||||
counts.SKUCreated++
|
||||
}
|
||||
if u {
|
||||
counts.SKUUpdated++
|
||||
}
|
||||
}
|
||||
for _, a := range req.Associations {
|
||||
c, u, e := repository.ApplyCatalogAssociation(tx, a.ShopeeGoodsID, a.PddGoodsID, now)
|
||||
if e != nil {
|
||||
return fail(&CatalogError{Status: 409, Code: "ASSOCIATION_CONFLICT", Message: e.Error()})
|
||||
}
|
||||
if c {
|
||||
counts.AssociationCreated++
|
||||
}
|
||||
if u {
|
||||
counts.AssociationUnchanged++
|
||||
}
|
||||
}
|
||||
resp := CatalogBatchResponse{BatchID: req.BatchID, Status: "succeeded", Counts: counts, FinishedAt: model.NowISO()}
|
||||
body, _ := json.Marshal(resp)
|
||||
run.Status = model.CatalogImportSucceeded
|
||||
run.FinishedAt = resp.FinishedAt
|
||||
run.ResponseBody = string(body)
|
||||
run.ShopeeCreated = counts.ShopeeCreated
|
||||
run.ShopeeUpdated = counts.ShopeeUpdated
|
||||
run.SKUCreated = counts.SKUCreated
|
||||
run.SKUUpdated = counts.SKUUpdated
|
||||
run.PddCreated = counts.PddCreated
|
||||
run.PddUpdated = counts.PddUpdated
|
||||
run.AssociationCreated = counts.AssociationCreated
|
||||
run.AssociationUnchanged = counts.AssociationUnchanged
|
||||
if err = repository.CompleteCatalogImportRun(tx, run); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return CatalogBatchResponse{}, err
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func replayCatalogBatch(db *sql.DB, source, batchID, hash, now string) (CatalogBatchResponse, error) {
|
||||
run, err := repository.GetCatalogImportRun(db, source, batchID)
|
||||
if err != nil {
|
||||
return CatalogBatchResponse{}, err
|
||||
}
|
||||
if run.RequestHash != hash {
|
||||
_ = repository.RecordCatalogImportConflict(db, source, batchID, now)
|
||||
return CatalogBatchResponse{}, &CatalogError{Status: 409, Code: "IDEMPOTENCY_CONFLICT", Message: "同一 batch_id 已提交不同内容"}
|
||||
}
|
||||
_ = repository.RecordCatalogImportReplay(db, source, batchID, now)
|
||||
if run.Status == model.CatalogImportProcessing {
|
||||
return CatalogBatchResponse{}, &CatalogError{Status: 409, Code: "BATCH_PROCESSING", Message: "批次仍在处理中", Retryable: true}
|
||||
}
|
||||
if run.Status == model.CatalogImportFailed {
|
||||
var stored CatalogError
|
||||
if json.Unmarshal([]byte(run.ResponseBody), &stored) == nil && stored.Code != "" {
|
||||
return CatalogBatchResponse{}, &stored
|
||||
}
|
||||
return CatalogBatchResponse{}, &CatalogError{Status: 422, Code: "BATCH_FAILED", Message: run.ErrorSummary}
|
||||
}
|
||||
var resp CatalogBatchResponse
|
||||
if err = json.Unmarshal([]byte(run.ResponseBody), &resp); err != nil {
|
||||
return CatalogBatchResponse{}, fmt.Errorf("读取批次重放结果失败: %w", err)
|
||||
}
|
||||
resp.Replayed = true
|
||||
return resp, nil
|
||||
}
|
||||
func truncateCatalogError(s string) string {
|
||||
if len(s) > 1000 {
|
||||
return s[:1000]
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
|
||||
"cmautobuy/admin/repository"
|
||||
)
|
||||
|
||||
func newCatalogTestDB(t *testing.T) *sql.DB {
|
||||
t.Helper()
|
||||
db, err := sql.Open("sqlite", ":memory:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db.SetMaxOpenConns(1)
|
||||
statements := []string{
|
||||
`CREATE TABLE shopee_products(goods_id TEXT PRIMARY KEY,title TEXT NOT NULL,shopee_status TEXT,main_sku_code TEXT,source TEXT,source_observed_at TEXT,pdd_goods_url TEXT,pdd_goods_id TEXT,created_at TEXT,updated_at TEXT)`,
|
||||
`CREATE TABLE shopee_skus(sku_id TEXT PRIMARY KEY,goods_id TEXT NOT NULL,spec_raw TEXT,color TEXT,size TEXT,advice TEXT,parse_ok INTEGER,sku_code TEXT,is_manual INTEGER,source_observed_at TEXT,created_at TEXT,updated_at TEXT)`,
|
||||
`CREATE TABLE pdd_products(id INTEGER PRIMARY KEY AUTOINCREMENT,goods_id TEXT UNIQUE,url TEXT,title TEXT,shop_name TEXT,skus_json TEXT,collect_status TEXT,collect_msg TEXT,artifact_ref TEXT,collected_at TEXT,deleted_at TEXT,source TEXT,source_observed_at TEXT,created_at TEXT,updated_at TEXT)`,
|
||||
`CREATE TABLE catalog_import_runs(source TEXT,batch_id TEXT,request_hash TEXT,status TEXT,request_count INTEGER,conflict_count INTEGER,observed_at TEXT,last_request_at TEXT,last_conflict_at TEXT,shopee_created INTEGER DEFAULT 0,shopee_updated INTEGER DEFAULT 0,sku_created INTEGER DEFAULT 0,sku_updated INTEGER DEFAULT 0,pdd_created INTEGER DEFAULT 0,pdd_updated INTEGER DEFAULT 0,association_created INTEGER DEFAULT 0,association_unchanged INTEGER DEFAULT 0,failure_count INTEGER DEFAULT 0,error_summary TEXT,response_body TEXT,created_at TEXT,finished_at TEXT,PRIMARY KEY(source,batch_id))`,
|
||||
}
|
||||
for _, statement := range statements {
|
||||
if _, err := db.Exec(statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
t.Cleanup(func() { db.Close() })
|
||||
return db
|
||||
}
|
||||
|
||||
func validCatalogBatch() CatalogBatchRequest {
|
||||
price := int64(1299)
|
||||
return CatalogBatchRequest{SchemaVersion: 1, BatchID: "batch-001", ObservedAt: "2026-08-11T08:00:00+08:00",
|
||||
ShopeeProducts: []CatalogShopeeProduct{{GoodsID: "S-1", Title: "蝦皮上衣"}},
|
||||
ShopeeSKUs: []CatalogShopeeSKU{{SKUID: "SKU-1", GoodsID: "S-1", SpecRaw: "黑色,M", Color: "黑色", Size: "M", ParseOK: true}},
|
||||
PddProducts: []CatalogPddProduct{{GoodsID: "P-1", URL: "https://mobile.yangkeduo.com/goods.html?goods_id=P-1", Title: "PDD上衣", Dimensions: []CatalogPddDimension{{Key: "color", Name: "颜色"}}, SKUs: []CatalogPddSKU{{Options: map[string]string{"color": "黑色"}, PriceCent: &price, Available: true}}}},
|
||||
Associations: []CatalogAssociation{{ShopeeGoodsID: "S-1", PddGoodsID: "P-1"}}}
|
||||
}
|
||||
|
||||
func TestImportCatalogBatch_原子写入重放与冲突(t *testing.T) {
|
||||
db := newCatalogTestDB(t)
|
||||
req := validCatalogBatch()
|
||||
raw, _ := json.Marshal(req)
|
||||
got, err := ImportCatalogBatch(db, "script-a", req, raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Counts.ShopeeCreated != 1 || got.Counts.SKUCreated != 1 || got.Counts.PddCreated != 1 || got.Counts.AssociationCreated != 1 {
|
||||
t.Fatalf("统计不正确：%+v", got)
|
||||
}
|
||||
replay, err := ImportCatalogBatch(db, "script-a", req, raw)
|
||||
if err != nil || !replay.Replayed {
|
||||
t.Fatalf("相同批次应重放：resp=%+v err=%v", replay, err)
|
||||
}
|
||||
changed := append([]byte{}, raw...)
|
||||
changed = append(changed, ' ')
|
||||
_, err = ImportCatalogBatch(db, "script-a", req, changed)
|
||||
var catalogErr *CatalogError
|
||||
if !errors.As(err, &catalogErr) || catalogErr.Code != "IDEMPOTENCY_CONFLICT" {
|
||||
t.Fatalf("不同内容应冲突：%v", err)
|
||||
}
|
||||
run, err := repository.GetCatalogImportRun(db, "script-a", req.BatchID)
|
||||
if err != nil || run.RequestCount != 3 || run.ConflictCount != 1 {
|
||||
t.Fatalf("批次计数不正确：%+v err=%v", run, err)
|
||||
}
|
||||
var pddID string
|
||||
if err := db.QueryRow(`SELECT pdd_goods_id FROM shopee_products WHERE goods_id='S-1'`).Scan(&pddID); err != nil || pddID != "P-1" {
|
||||
t.Fatalf("关联未写入：%q %v", pddID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestImportCatalogBatch_关联冲突整批回滚且同请求重放错误(t *testing.T) {
|
||||
db := newCatalogTestDB(t)
|
||||
now := "2026-08-11T00:00:00.000000000Z"
|
||||
_, _ = db.Exec(`INSERT INTO shopee_products(goods_id,title,source,pdd_goods_id,created_at,updated_at) VALUES('S-1','旧标题','report','P-OLD',?,?)`, now, now)
|
||||
_, _ = db.Exec(`INSERT INTO pdd_products(goods_id,url,collect_status,created_at,updated_at) VALUES('P-OLD','old','pending',?,?)`, now, now)
|
||||
req := validCatalogBatch()
|
||||
raw, _ := json.Marshal(req)
|
||||
_, err := ImportCatalogBatch(db, "script-a", req, raw)
|
||||
var first *CatalogError
|
||||
if !errors.As(err, &first) || first.Code != "ASSOCIATION_CONFLICT" {
|
||||
t.Fatalf("应返回关联冲突：%v", err)
|
||||
}
|
||||
var title string
|
||||
_ = db.QueryRow(`SELECT title FROM shopee_products WHERE goods_id='S-1'`).Scan(&title)
|
||||
if title != "旧标题" {
|
||||
t.Fatalf("失败批次必须整体回滚，title=%q", title)
|
||||
}
|
||||
_, err = ImportCatalogBatch(db, "script-a", req, raw)
|
||||
var replay *CatalogError
|
||||
if !errors.As(err, &replay) || replay.Code != first.Code || replay.Status != first.Status {
|
||||
t.Fatalf("失败重放应保持原错误：%+v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateCatalogBatch_拒绝旧版超限和负金额(t *testing.T) {
|
||||
req := validCatalogBatch()
|
||||
req.SchemaVersion = 2
|
||||
if err := ValidateCatalogBatch(req); err == nil {
|
||||
t.Fatal("应拒绝未知 schema")
|
||||
}
|
||||
req = validCatalogBatch()
|
||||
negative := int64(-1)
|
||||
req.PddProducts[0].SKUs[0].PriceCent = &negative
|
||||
if err := ValidateCatalogBatch(req); err == nil {
|
||||
t.Fatal("应拒绝负金额")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user