diff --git a/admin/handler/integration/auth_test.go b/admin/handler/integration/auth_test.go index b31e95e..bc55ead 100644 --- a/admin/handler/integration/auth_test.go +++ b/admin/handler/integration/auth_test.go @@ -1,6 +1,7 @@ package integration import ( + "bytes" "net/http" "net/http/httptest" "strings" @@ -45,3 +46,33 @@ func TestRequireCatalogToken(t *testing.T) { }) } } + +func TestCatalogBatchAPI_鉴权JSON与体积限制(t *testing.T) { + gin.SetMode(gin.TestMode) + const token = "0123456789abcdef0123456789abcdef" + r := gin.New() + Register(r, nil, config.CatalogIntegrationConfig{Source: "script", Token: token}) + + for _, tc := range []struct { + name, auth string + body []byte + want int + }{ + {"无鉴权", "", []byte(`{}`), http.StatusUnauthorized}, + {"坏JSON", "Bearer " + token, []byte(`{"schema_version":`), http.StatusBadRequest}, + {"超限", "Bearer " + token, bytes.Repeat([]byte("x"), catalogMaxBody+1), http.StatusRequestEntityTooLarge}, + } { + t.Run(tc.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/api/v1/integrations/catalog/batches", bytes.NewReader(tc.body)) + req.Header.Set("Authorization", tc.auth) + resp := httptest.NewRecorder() + r.ServeHTTP(resp, req) + if resp.Code != tc.want { + t.Fatalf("status=%d body=%s", resp.Code, resp.Body.String()) + } + if !strings.Contains(resp.Header().Get("Content-Type"), "application/json") { + t.Fatalf("错误响应必须是 JSON：%s", resp.Header()) + } + }) + } +} diff --git a/admin/handler/integration/catalog_api.go b/admin/handler/integration/catalog_api.go new file mode 100644 index 0000000..da41510 --- /dev/null +++ b/admin/handler/integration/catalog_api.go @@ -0,0 +1,61 @@ +package integration + +import ( + "bytes" + "database/sql" + "encoding/json" + "errors" + "io" + "net/http" + + "github.com/gin-gonic/gin" + + "cmautobuy/admin/config" + "cmautobuy/admin/service" +) + +const catalogMaxBody = 5 << 20 + +type Handler struct{ db *sql.DB } + +// Register 只挂载第三方目录接口，不复用 Client 或网页登录中间件。 +func Register(r *gin.Engine, db *sql.DB, cfg config.CatalogIntegrationConfig) { + h := &Handler{db: db} + g := r.Group("/api/v1/integrations/catalog", RequireCatalogToken(cfg)) + g.POST("/batches", h.CreateBatch) +} + +func (h *Handler) CreateBatch(c *gin.Context) { + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, catalogMaxBody) + raw, err := io.ReadAll(c.Request.Body) + if err != nil { + if _, ok := err.(*http.MaxBytesError); ok { + integrationError(c, 413, "PAYLOAD_TOO_LARGE", "请求体不能超过 5 MB", false, nil) + return + } + integrationError(c, 400, "INVALID_BODY", "读取请求失败", false, nil) + return + } + var req service.CatalogBatchRequest + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + if err = dec.Decode(&req); err != nil { + integrationError(c, 400, "INVALID_JSON", "请求体不是合法的商品目录 JSON", false, nil) + return + } + if dec.Decode(&struct{}{}) != io.EOF { + integrationError(c, 400, "INVALID_JSON", "请求体只能包含一个 JSON 对象", false, nil) + return + } + resp, err := service.ImportCatalogBatch(h.db, integrationSource(c), req, raw) + if err != nil { + var catalogErr *service.CatalogError + if errors.As(err, &catalogErr) { + integrationError(c, catalogErr.Status, catalogErr.Code, catalogErr.Message, catalogErr.Retryable, catalogErr.Details) + return + } + integrationError(c, 500, "INTERNAL_ERROR", "商品目录批次处理失败", true, nil) + return + } + c.JSON(http.StatusOK, resp) +} diff --git a/admin/main.go b/admin/main.go index 2dea8d5..5f377ce 100644 --- a/admin/main.go +++ b/admin/main.go @@ -22,6 +22,7 @@ import ( "cmautobuy/admin/config" "cmautobuy/admin/handler/api" + "cmautobuy/admin/handler/integration" "cmautobuy/admin/handler/web" "cmautobuy/admin/repository" "cmautobuy/admin/service" @@ -108,5 +109,10 @@ func newRouter(db *sql.DB) (*gin.Engine, error) { // 4. 路由 web.Register(r, db, config.OnlineThreshold) // 给浏览器的页面 api.Register(r, db) // 给 Client 的接口 + catalogConfig, err := config.LoadCatalogIntegration() + if err != nil { + return nil, fmt.Errorf("读取商品目录接口配置失败: %w", err) + } + integration.Register(r, db, catalogConfig) // 给受信任第三方脚本的接口 return r, nil } diff --git a/admin/repository/catalog_import.go b/admin/repository/catalog_import.go index f7f9c23..6d348fa 100644 --- a/admin/repository/catalog_import.go +++ b/admin/repository/catalog_import.go @@ -79,6 +79,121 @@ func RecordCatalogImportConflict(q Execer, source, batchID, requestedAt string) return catalogRunUpdateResult(result, err, "记录商品目录批次冲突") } +// CompleteCatalogImportRun 把处理结果摘要和安全的响应 JSON 固化，供幂等重放。 +func CompleteCatalogImportRun(q Execer, run model.CatalogImportRun) error { + result, err := q.Exec(`UPDATE catalog_import_runs SET status=?,shopee_created=?,shopee_updated=?, + sku_created=?,sku_updated=?,pdd_created=?,pdd_updated=?,association_created=?, + association_unchanged=?,failure_count=?,error_summary=NULLIF(?,''),response_body=NULLIF(?,''), + finished_at=? WHERE source=? AND batch_id=?`, run.Status, run.ShopeeCreated, run.ShopeeUpdated, + run.SKUCreated, run.SKUUpdated, run.PddCreated, run.PddUpdated, run.AssociationCreated, + run.AssociationUnchanged, run.FailureCount, run.ErrorSummary, run.ResponseBody, + run.FinishedAt, run.Source, run.BatchID) + return catalogRunUpdateResult(result, err, "完成商品目录批次") +} + +// UpsertCatalogShopeeProduct 按上游观测时间更新来源字段，绝不覆盖人工 PDD 关联。 +func UpsertCatalogShopeeProduct(q Execer, goodsID, title, status, mainSKU, observedAt, now string) (created, updated bool, err error) { + var oldObserved sql.NullString + err = q.QueryRow(`SELECT source_observed_at FROM shopee_products WHERE goods_id=?`, goodsID).Scan(&oldObserved) + if errors.Is(err, sql.ErrNoRows) { + _, err = q.Exec(`INSERT INTO shopee_products(goods_id,title,shopee_status,main_sku_code,source, + source_observed_at,created_at,updated_at) VALUES(?,?,NULLIF(?,''),NULLIF(?,''),'api',?,?,?)`, + goodsID, title, status, mainSKU, observedAt, now, now) + return err == nil, false, err + } + if err != nil { + return false, false, err + } + if oldObserved.Valid && oldObserved.String > observedAt { + return false, false, nil + } + _, err = q.Exec(`UPDATE shopee_products SET title=?,shopee_status=NULLIF(?,''),main_sku_code=NULLIF(?,''), + source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`, title, status, mainSKU, observedAt, now, goodsID) + return false, err == nil, err +} + +// UpsertCatalogShopeeSKU 拒绝把同一 SKU 静默挪到另一商品，并保留 is_manual。 +func UpsertCatalogShopeeSKU(q Execer, skuID, goodsID, specRaw, color, size, advice string, parseOK bool, skuCode, observedAt, now string) (created, updated bool, err error) { + var oldGoods string + var oldObserved sql.NullString + err = q.QueryRow(`SELECT goods_id,source_observed_at FROM shopee_skus WHERE sku_id=?`, skuID).Scan(&oldGoods, &oldObserved) + parse := 0 + if parseOK { + parse = 1 + } + if errors.Is(err, sql.ErrNoRows) { + _, err = q.Exec(`INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,color,size,advice,parse_ok,sku_code, + is_manual,source_observed_at,created_at,updated_at) VALUES(?,?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?,NULLIF(?,''),0,?,?,?)`, + skuID, goodsID, specRaw, color, size, advice, parse, skuCode, observedAt, now, now) + return err == nil, false, err + } + if err != nil { + return false, false, err + } + if oldGoods != goodsID { + return false, false, fmt.Errorf("SKU %s 已属于蝦皮商品 %s", skuID, oldGoods) + } + if oldObserved.Valid && oldObserved.String > observedAt { + return false, false, nil + } + _, err = q.Exec(`UPDATE shopee_skus SET spec_raw=?,color=NULLIF(?,''),size=NULLIF(?,''),advice=NULLIF(?,''), + parse_ok=?,sku_code=NULLIF(?,''),source_observed_at=?,updated_at=? WHERE sku_id=?`, + specRaw, color, size, advice, parse, skuCode, observedAt, now, skuID) + return false, err == nil, err +} + +// UpsertCatalogPddProduct 写入结构化 PDD 数据转换后的规范 JSON；空规格不清除既有采集结果。 +func UpsertCatalogPddProduct(q Execer, goodsID, url, title, shopName, skusJSON, observedAt, now string) (created, updated bool, err error) { + var oldObserved, deletedAt sql.NullString + err = q.QueryRow(`SELECT source_observed_at,deleted_at FROM pdd_products WHERE goods_id=?`, goodsID).Scan(&oldObserved, &deletedAt) + status := "pending" + if skusJSON != "" { + status = "collected" + } + if errors.Is(err, sql.ErrNoRows) { + _, err = q.Exec(`INSERT INTO pdd_products(goods_id,url,title,shop_name,skus_json,collect_status,collected_at, + source,source_observed_at,created_at,updated_at) VALUES(?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?, + CASE WHEN ?='collected' THEN ? ELSE NULL END,'api',?,?,?)`, goodsID, url, title, shopName, skusJSON, status, status, now, observedAt, now, now) + return err == nil, false, err + } + if err != nil { + return false, false, err + } + if oldObserved.Valid && oldObserved.String > observedAt { + return false, false, nil + } + if skusJSON == "" { + _, err = q.Exec(`UPDATE pdd_products SET url=?,title=CASE WHEN TRIM(?)='' THEN title ELSE ? END, + shop_name=CASE WHEN TRIM(?)='' THEN shop_name ELSE ? END,deleted_at=NULL,source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`, + url, title, title, shopName, shopName, observedAt, now, goodsID) + } else { + _, err = q.Exec(`UPDATE pdd_products SET url=?,title=NULLIF(?,''),shop_name=NULLIF(?,''),skus_json=?, + collect_status='collected',collect_msg=NULL,artifact_ref=NULL,collected_at=?,deleted_at=NULL, + source='api',source_observed_at=?,updated_at=? WHERE goods_id=?`, url, title, shopName, skusJSON, now, observedAt, now, goodsID) + } + return false, err == nil, err +} + +// ApplyCatalogAssociation 只允许空关联建立或相同关联重放。 +func ApplyCatalogAssociation(q Execer, shopeeGoodsID, pddGoodsID, now string) (created, unchanged bool, err error) { + var current sql.NullString + if err = q.QueryRow(`SELECT pdd_goods_id FROM shopee_products WHERE goods_id=?`, shopeeGoodsID).Scan(¤t); err != nil { + return + } + var url string + if err = q.QueryRow(`SELECT url FROM pdd_products WHERE goods_id=? AND deleted_at IS NULL`, pddGoodsID).Scan(&url); err != nil { + return + } + if current.Valid && current.String != "" { + if current.String == pddGoodsID { + return false, true, nil + } + return false, false, fmt.Errorf("蝦皮商品 %s 已关联 PDD 商品 %s", shopeeGoodsID, current.String) + } + _, err = q.Exec(`UPDATE shopee_products SET pdd_goods_id=?,pdd_goods_url=?,updated_at=? WHERE goods_id=?`, pddGoodsID, url, now, shopeeGoodsID) + return err == nil, false, err +} + func catalogRunUpdateResult(result sql.Result, err error, action string) error { if err != nil { return fmt.Errorf("%s失败: %w", action, err) diff --git a/admin/service/catalog_import.go b/admin/service/catalog_import.go new file mode 100644 index 0000000..d8befde --- /dev/null +++ b/admin/service/catalog_import.go @@ -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 +} diff --git a/admin/service/catalog_import_test.go b/admin/service/catalog_import_test.go new file mode 100644 index 0000000..e51ae10 --- /dev/null +++ b/admin/service/catalog_import_test.go @@ -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("应拒绝负金额") + } +} diff --git a/docs/README.md b/docs/README.md index 79d452e..5cff028 100644 --- a/docs/README.md +++ b/docs/README.md @@ -50,6 +50,7 @@ | 加字段、改表、写 SQL | [03 数据模型](admin/03-data-model.md) | [02 架构](admin/02-architecture.md) §2 分层 | | 改 Excel 导入 | [03 数据模型](admin/03-data-model.md) §3.3 | [00 术语表](admin/00-glossary.md) §3 upsert | | 改顺运宝同步 | [08 顺运宝接口](admin/08-顺运宝接口.md) | [03 数据模型](admin/03-data-model.md) | +| 对接第三方商品目录脚本 | [10 商品目录接入接口](admin/10-商品目录接入接口.md) | [03 数据模型](admin/03-data-model.md) | | 把旧 Admin 数据迁移到 MySQL | [09 SQLite 单向迁移](admin/09-sqlite迁移到mysql.md) | [06 质量与安全](admin/06-quality-security.md) | | 改给 Client 的接口 | [04 Client 接口实现](admin/04-client-api.md) | [Client 侧契约](client/04-admin-api-contract.md) | | 和 Admin 联调、登记新设备 | [07 设备登记联调手册](admin/07-设备登记联调手册.md) | [Client 侧契约](client/04-admin-api-contract.md) §5 | @@ -92,6 +93,7 @@ | [07 设备登记联调手册](admin/07-设备登记联调手册.md) | **给 Client 开发者**：怎么让新设备登记成功 | | [08 顺运宝接口](admin/08-顺运宝接口.md) | 从抓包还原的外部 ERP 契约：登录、会话、货运单列表与明细 | | [09 SQLite 单向迁移](admin/09-sqlite迁移到mysql.md) | 只读演练、一次性导入、逐表核对和失败处理 | +| [10 商品目录接入接口](admin/10-商品目录接入接口.md) | 第三方批量提交蝦皮、PDD 和商品关联的 JSON 契约 | ## 文档标注说明 diff --git a/docs/admin/10-商品目录接入接口.md b/docs/admin/10-商品目录接入接口.md new file mode 100644 index 0000000..13d8e12 --- /dev/null +++ b/docs/admin/10-商品目录接入接口.md @@ -0,0 +1,72 @@ +# 10 商品目录批量接入接口 + +第三方脚本先把不同 Excel 归一化，再用本接口一次提交蝦皮商品、蝦皮 SKU、PDD +商品及商品关联。接口只做 upsert，本批次没出现的数据不会被删除。 + +## 1. 地址与鉴权 + +- `POST /api/v1/integrations/catalog/batches` +- 请求头：`Authorization: Bearer <专用 Token>` +- 请求体上限 5 MB；每批商品总数最多 500，蝦皮 SKU 最多 5000。 + +Token 通过 `CMAUTOBUY_CATALOG_TOKEN` 或未提交的 `config.yaml` 配置。未配置时接口 +返回 503；凭据错误返回 401。不要把 Token 放进 URL、日志或导入文件。 + +## 2. 请求示例 + +```json +{ + "schema_version": 1, + "batch_id": "source-file-20260811-001", + "observed_at": "2026-08-11T10:00:00+08:00", + "shopee_products": [ + {"goods_id": "S-1", "title": "蝦皮上衣", "status": "NORMAL", "main_sku_code": "A01"} + ], + "shopee_skus": [ + {"sku_id": "SKU-1", "goods_id": "S-1", "spec_raw": "黑色,M", "color": "黑色", "size": "M", "parse_ok": true, "sku_code": "A01-B-M"} + ], + "pdd_products": [ + { + "goods_id": "P-1", + "url": "https://mobile.yangkeduo.com/goods.html?goods_id=P-1", + "title": "PDD 上衣", + "shop_name": "示例店铺", + "dimensions": [{"key": "color", "name": "颜色"}, {"key": "size", "name": "尺码"}], + "skus": [{"options": {"color": "黑色", "size": "M"}, "price_cent": 1299, "list_price_cent": 1599, "available": true}] + } + ], + "associations": [{"shopee_goods_id": "S-1", "pdd_goods_id": "P-1"}] +} +``` + +金额单位固定为人民币分，不接受小数金额。`spec_raw` 必须原样提交。PDD 规格必须用 +结构化 `dimensions` 和 `skus`，第三方不能提交数据库内部的 `skus_json`。 + +## 3. 写入与幂等规则 + +- 整个批次在一个事务中写入：实体、SKU、关联任一步失败就全部回滚。 +- `(source, batch_id)` 是幂等键。请求字节相同则重放首次结果；同键不同内容返回 409。 +- 较旧的 `observed_at` 不覆盖较新的接口数据。 +- 蝦皮 upsert 不覆盖采购员维护的 PDD 链接；SKU 的 `is_manual` 不被接口改写。 +- 空关联可以建立、相同关联不重复写；已有不同 PDD 关联返回 409，绝不静默替换。 +- 批次缺少某条记录不表示删除，接口没有“全量覆盖”语义。 + +成功响应包含批次状态和各类新增、更新、关联计数。相同请求重放时 +`replayed=true`。 + +## 4. 错误格式 + +```json +{ + "error": { + "code": "ASSOCIATION_CONFLICT", + "message": "蝦皮商品已经关联其他 PDD 商品", + "retryable": false, + "request_id": "调用方可选请求号", + "details": {} + } +} +``` + +调用方应按 `code` 处理，不要解析中文 `message`。校验错误通常是 400/422，业务冲突 +是 409，临时服务错误是 500/503。响应和导入记录都不会回显 Token 或完整请求体。