296 lines
10 KiB
Go
296 lines
10 KiB
Go
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
|
|
}
|