585 lines
27 KiB
Go
585 lines
27 KiB
Go
package repository
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/go-sql-driver/mysql"
|
|
|
|
"cmautobuy/admin/model"
|
|
)
|
|
|
|
var (
|
|
// ErrCatalogImportRunExists 表示同一来源的批次号已经登记，由 service 判断是否重放。
|
|
ErrCatalogImportRunExists = errors.New("商品目录批次已经存在")
|
|
ErrCatalogImportRunNotFound = errors.New("商品目录批次不存在")
|
|
)
|
|
|
|
// CatalogEntityExists 仅用于目录导入预检，不会修改任何记录。
|
|
func CatalogEntityExists(q Execer, table, goodsID string) (bool, error) {
|
|
var found int
|
|
var query string
|
|
switch table {
|
|
case "shopee_products":
|
|
query = `SELECT 1 FROM shopee_products WHERE goods_id=? AND deleted_at IS NULL`
|
|
case "pdd_products":
|
|
query = `SELECT 1 FROM pdd_products WHERE goods_id=? AND deleted_at IS NULL`
|
|
default:
|
|
return false, fmt.Errorf("不支持的目录表: %s", table)
|
|
}
|
|
err := q.QueryRow(query, goodsID).Scan(&found)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return false, nil
|
|
}
|
|
return err == nil, err
|
|
}
|
|
|
|
// CatalogSKUExists 按与实际写入一致的商品+规格键判断 SKU 是否已存在。
|
|
func CatalogSKUExists(q Execer, goodsID, specKey string) (bool, error) {
|
|
var found int
|
|
err := q.QueryRow(`SELECT 1 FROM shopee_skus WHERE goods_id=? AND spec_key=?`, goodsID, specKey).Scan(&found)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return false, nil
|
|
}
|
|
return err == nil, err
|
|
}
|
|
|
|
// CatalogAssociationCurrentPDD 返回当前关联；空字符串表示该蝦皮商品没有 PDD 关联。
|
|
func CatalogAssociationCurrentPDD(q Execer, goodsID string) (string, bool, error) {
|
|
var current sql.NullString
|
|
err := q.QueryRow(`SELECT pdd_goods_id FROM shopee_products WHERE goods_id=? AND deleted_at IS NULL`, goodsID).Scan(¤t)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
return current.String, true, nil
|
|
}
|
|
|
|
// InsertCatalogImportRun 先登记 processing 批次；表的复合主键是并发幂等的最终防线。
|
|
func InsertCatalogImportRun(q Execer, run model.CatalogImportRun) error {
|
|
_, err := q.Exec(`INSERT INTO catalog_import_runs (
|
|
source,batch_id,request_hash,status,request_count,conflict_count,observed_at,
|
|
update_policy,last_request_at,created_at
|
|
) VALUES (?,?,?,?,1,0,NULLIF(?,''),?,?,?)`, run.Source, run.BatchID, run.RequestHash,
|
|
run.Status, run.ObservedAt, run.UpdatePolicy, run.LastRequestAt, run.CreatedAt)
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
var mysqlErr *mysql.MySQLError
|
|
if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
|
|
return ErrCatalogImportRunExists
|
|
}
|
|
if strings.Contains(strings.ToLower(err.Error()), "unique constraint failed") {
|
|
return ErrCatalogImportRunExists
|
|
}
|
|
return fmt.Errorf("登记商品目录批次失败: %w", err)
|
|
}
|
|
|
|
// GetCatalogImportRun 查询一个批次的摘要，不读取原始请求体。
|
|
func GetCatalogImportRun(q Execer, source, batchID string) (model.CatalogImportRun, error) {
|
|
var run model.CatalogImportRun
|
|
var observedAt, lastConflictAt, errorSummary, responseBody, finishedAt sql.NullString
|
|
err := q.QueryRow(`SELECT source,batch_id,request_hash,status,request_count,conflict_count,
|
|
observed_at,update_policy,last_request_at,last_conflict_at,shopee_created,shopee_updated,shopee_fields_filled,shopee_fields_same_source_updated,shopee_fields_manual_skipped,shopee_fields_stale_skipped,sku_created,
|
|
sku_updated,sku_filled,sku_same_source_updated,sku_skipped,sku_manual_skipped,sku_stale_skipped,
|
|
pdd_created,pdd_updated,association_created,association_unchanged,failure_count,
|
|
error_summary,response_body,created_at,finished_at
|
|
FROM catalog_import_runs WHERE source=? AND batch_id=?`, source, batchID).Scan(
|
|
&run.Source, &run.BatchID, &run.RequestHash, &run.Status, &run.RequestCount, &run.ConflictCount,
|
|
&observedAt, &run.UpdatePolicy, &run.LastRequestAt, &lastConflictAt, &run.ShopeeCreated, &run.ShopeeUpdated, &run.ShopeeFieldsFilled, &run.ShopeeFieldsSameSourceUpdated, &run.ShopeeFieldsManualSkipped, &run.ShopeeFieldsStaleSkipped,
|
|
&run.SKUCreated, &run.SKUUpdated, &run.SKUFilled, &run.SKUSameSourceUpdated, &run.SKUSkipped,
|
|
&run.SKUManualSkipped, &run.SKUStaleSkipped, &run.PddCreated, &run.PddUpdated, &run.AssociationCreated,
|
|
&run.AssociationUnchanged, &run.FailureCount, &errorSummary, &responseBody, &run.CreatedAt, &finishedAt)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return model.CatalogImportRun{}, ErrCatalogImportRunNotFound
|
|
}
|
|
if err != nil {
|
|
return model.CatalogImportRun{}, fmt.Errorf("查询商品目录批次失败: %w", err)
|
|
}
|
|
run.ObservedAt = observedAt.String
|
|
run.LastConflictAt = lastConflictAt.String
|
|
run.ErrorSummary = errorSummary.String
|
|
run.ResponseBody = responseBody.String
|
|
run.FinishedAt = finishedAt.String
|
|
return run, nil
|
|
}
|
|
|
|
// ListCatalogImportRuns 在数据库中按来源、状态筛选并倒序分页。
|
|
func ListCatalogImportRuns(q Execer, source string, status model.CatalogImportStatus, limit, offset int) ([]model.CatalogImportRun, int, error) {
|
|
where := " WHERE 1=1"
|
|
args := make([]any, 0, 4)
|
|
if source != "" {
|
|
where += " AND source=?"
|
|
args = append(args, source)
|
|
}
|
|
if status != "" {
|
|
where += " AND status=?"
|
|
args = append(args, status)
|
|
}
|
|
var total int
|
|
if err := q.QueryRow(`SELECT COUNT(*) FROM catalog_import_runs`+where, args...).Scan(&total); err != nil {
|
|
return nil, 0, fmt.Errorf("统计商品目录批次失败: %w", err)
|
|
}
|
|
listArgs := append(append([]any{}, args...), limit, offset)
|
|
rows, err := q.Query(`SELECT source,batch_id FROM catalog_import_runs`+where+` ORDER BY created_at DESC,source,batch_id LIMIT ? OFFSET ?`, listArgs...)
|
|
if err != nil {
|
|
return nil, 0, fmt.Errorf("查询商品目录批次列表失败: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
keys := make([][2]string, 0, limit)
|
|
for rows.Next() {
|
|
var key [2]string
|
|
if err := rows.Scan(&key[0], &key[1]); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
keys = append(keys, key)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
list := make([]model.CatalogImportRun, 0, len(keys))
|
|
for _, key := range keys {
|
|
run, err := GetCatalogImportRun(q, key[0], key[1])
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
list = append(list, run)
|
|
}
|
|
return list, total, nil
|
|
}
|
|
|
|
// ListCatalogImportSources 返回筛选下拉框所需的稳定来源名。
|
|
func ListCatalogImportSources(q Execer) ([]string, error) {
|
|
rows, err := q.Query(`SELECT DISTINCT source FROM catalog_import_runs ORDER BY source`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var list []string
|
|
for rows.Next() {
|
|
var source string
|
|
if err := rows.Scan(&source); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, source)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
// RecordCatalogImportReplay 记录相同请求的重复提交，业务数据不会再次写入。
|
|
func RecordCatalogImportReplay(q Execer, source, batchID, requestedAt string) error {
|
|
result, err := q.Exec(`UPDATE catalog_import_runs SET request_count=request_count+1,last_request_at=?
|
|
WHERE source=? AND batch_id=?`, requestedAt, source, batchID)
|
|
return catalogRunUpdateResult(result, err, "记录商品目录批次重放")
|
|
}
|
|
|
|
// RecordCatalogImportConflict 记录同一批次号携带不同内容的冲突。
|
|
func RecordCatalogImportConflict(q Execer, source, batchID, requestedAt string) error {
|
|
result, err := q.Exec(`UPDATE catalog_import_runs SET request_count=request_count+1,
|
|
conflict_count=conflict_count+1,last_request_at=?,last_conflict_at=?
|
|
WHERE source=? AND batch_id=?`, requestedAt, requestedAt, source, batchID)
|
|
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=?,shopee_fields_filled=?,shopee_fields_same_source_updated=?,shopee_fields_manual_skipped=?,shopee_fields_stale_skipped=?,
|
|
sku_created=?,sku_updated=?,sku_filled=?,sku_same_source_updated=?,sku_skipped=?,
|
|
sku_manual_skipped=?,sku_stale_skipped=?,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.ShopeeFieldsFilled, run.ShopeeFieldsSameSourceUpdated, run.ShopeeFieldsManualSkipped, run.ShopeeFieldsStaleSkipped, run.SKUCreated, run.SKUUpdated, run.SKUFilled, run.SKUSameSourceUpdated, run.SKUSkipped,
|
|
run.SKUManualSkipped, run.SKUStaleSkipped, run.PddCreated, run.PddUpdated, run.AssociationCreated,
|
|
run.AssociationUnchanged, run.FailureCount, run.ErrorSummary, run.ResponseBody,
|
|
run.FinishedAt, run.Source, run.BatchID)
|
|
return catalogRunUpdateResult(result, err, "完成商品目录批次")
|
|
}
|
|
|
|
type CatalogShopeeProductInput struct{ GoodsID, Title, Status, MainSKU, ImageURL, ShopName, Source, ObservedAt, Now, UpdatePolicy string }
|
|
type CatalogProductOutcome struct {
|
|
Created, Updated bool
|
|
FieldsFilled, FieldsSameSourceUpdated, FieldsManualSkipped, FieldsStaleSkipped int
|
|
}
|
|
|
|
// UpsertCatalogShopeeProduct 独立保护主图和店铺字段，绝不覆盖人工 PDD 关联。
|
|
func UpsertCatalogShopeeProduct(q Execer, in CatalogShopeeProductInput) (out CatalogProductOutcome, err error) {
|
|
var oldObserved, oldShopID, title, status, mainSKU, imageURL, shopName, imageSource, imageObserved, shopSource, shopObserved sql.NullString
|
|
var imageManual, shopManual int
|
|
shopID, err := FindShopIDByName(q, in.ShopName)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
err = q.QueryRow(`SELECT source_observed_at,shop_id,title,shopee_status,main_sku_code,image_url,shopee_shop_name,image_source,image_observed_at,image_is_manual,shop_name_source,shop_name_observed_at,shop_name_is_manual FROM shopee_products WHERE goods_id=?`, in.GoodsID).Scan(&oldObserved, &oldShopID, &title, &status, &mainSKU, &imageURL, &shopName, &imageSource, &imageObserved, &imageManual, &shopSource, &shopObserved, &shopManual)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
_, err = q.Exec(`INSERT INTO shopee_products(goods_id,shop_id,title,shopee_status,main_sku_code,image_url,shopee_shop_name,image_source,image_observed_at,image_is_manual,shop_name_source,shop_name_observed_at,shop_name_is_manual,source,source_observed_at,created_at,updated_at) VALUES(?,NULLIF(?,''),?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,0,CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,CASE WHEN TRIM(?)='' THEN NULL ELSE ? END,0,'api',?,?,?)`, in.GoodsID, shopID, in.Title, in.Status, in.MainSKU, in.ImageURL, in.ShopName, in.ImageURL, in.Source, in.ImageURL, in.ObservedAt, in.ShopName, in.Source, in.ShopName, in.ObservedAt, in.ObservedAt, in.Now, in.Now)
|
|
out.Created = err == nil
|
|
return out, err
|
|
}
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
baseUpdate := in.UpdatePolicy == "overwrite_same_source" && (!oldObserved.Valid || oldObserved.String <= in.ObservedAt)
|
|
fillTitle := in.UpdatePolicy == "fill_missing" && strings.TrimSpace(in.Title) != "" && strings.TrimSpace(title.String) == ""
|
|
fillStatus := in.UpdatePolicy == "fill_missing" && strings.TrimSpace(in.Status) != "" && strings.TrimSpace(status.String) == ""
|
|
fillMainSKU := in.UpdatePolicy == "fill_missing" && strings.TrimSpace(in.MainSKU) != "" && strings.TrimSpace(mainSKU.String) == ""
|
|
newImage, newShop := imageURL.String, shopName.String
|
|
imageChanged, shopChanged := false, false
|
|
apply := func(incoming string, current *string, source, observed sql.NullString, manual int) (changed bool) {
|
|
if strings.TrimSpace(incoming) == "" || in.UpdatePolicy == "insert_only" {
|
|
return false
|
|
}
|
|
if manual != 0 {
|
|
out.FieldsManualSkipped++
|
|
return false
|
|
}
|
|
// 商品目录是权威来源，可以替换顺运宝为了预览而填入的低优先级值。
|
|
// 这不是任意跨来源覆盖：只有明确标记为 syb 的值享受升级规则。
|
|
if source.Valid && source.String == "syb" {
|
|
if *current != incoming {
|
|
*current = incoming
|
|
out.FieldsFilled++
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
if in.UpdatePolicy == "fill_missing" {
|
|
if *current == "" {
|
|
*current = incoming
|
|
out.FieldsFilled++
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
if !source.Valid || source.String != in.Source {
|
|
return false
|
|
}
|
|
if observed.Valid && observed.String > in.ObservedAt {
|
|
out.FieldsStaleSkipped++
|
|
return false
|
|
}
|
|
if *current != incoming {
|
|
*current = incoming
|
|
out.FieldsSameSourceUpdated++
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
imageChanged = apply(in.ImageURL, &newImage, imageSource, imageObserved, imageManual)
|
|
shopChanged = apply(in.ShopName, &newShop, shopSource, shopObserved, shopManual)
|
|
resolvedShopID := oldShopID.String
|
|
shopAssociationChanged := false
|
|
if shopChanged {
|
|
resolvedShopID, err = FindShopIDByName(q, newShop)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
shopAssociationChanged = resolvedShopID != oldShopID.String
|
|
} else if resolvedShopID == "" && shopID != "" && strings.TrimSpace(newShop) == strings.TrimSpace(in.ShopName) {
|
|
resolvedShopID = shopID
|
|
shopAssociationChanged = true
|
|
}
|
|
if !baseUpdate && !fillTitle && !fillStatus && !fillMainSKU && !imageChanged && !shopChanged && !shopAssociationChanged {
|
|
return out, nil
|
|
}
|
|
_, err = q.Exec(`UPDATE shopee_products SET title=CASE WHEN ? OR ? THEN ? ELSE title END,shopee_status=CASE WHEN ? OR ? THEN NULLIF(?,'') ELSE shopee_status END,main_sku_code=CASE WHEN ? OR ? THEN NULLIF(?,'') ELSE main_sku_code END,source=CASE WHEN ? THEN 'api' ELSE source END,source_observed_at=CASE WHEN ? THEN ? ELSE source_observed_at END,shop_id=CASE WHEN ? THEN NULLIF(?,'') ELSE shop_id END,image_url=NULLIF(?,''),shopee_shop_name=NULLIF(?,''),image_source=CASE WHEN ? THEN ? ELSE image_source END,image_observed_at=CASE WHEN ? THEN ? ELSE image_observed_at END,shop_name_source=CASE WHEN ? THEN ? ELSE shop_name_source END,shop_name_observed_at=CASE WHEN ? THEN ? ELSE shop_name_observed_at END,updated_at=? WHERE goods_id=?`, baseUpdate, fillTitle, in.Title, baseUpdate, fillStatus, in.Status, baseUpdate, fillMainSKU, in.MainSKU, baseUpdate, baseUpdate, in.ObservedAt, shopAssociationChanged, resolvedShopID, newImage, newShop, imageChanged, in.Source, imageChanged, in.ObservedAt, shopChanged, in.Source, shopChanged, in.ObservedAt, in.Now, in.GoodsID)
|
|
out.Updated = err == nil
|
|
return out, err
|
|
}
|
|
|
|
type CatalogSKUOutcome string
|
|
|
|
const (
|
|
CatalogSKUCreated CatalogSKUOutcome = "created"
|
|
CatalogSKUFilled CatalogSKUOutcome = "filled"
|
|
CatalogSKUSameSourceUpdated CatalogSKUOutcome = "same_source_updated"
|
|
CatalogSKUSkipped CatalogSKUOutcome = "skipped"
|
|
CatalogSKUManualSkipped CatalogSKUOutcome = "manual_skipped"
|
|
CatalogSKUStaleSkipped CatalogSKUOutcome = "stale_skipped"
|
|
)
|
|
|
|
type CatalogShopeeSKUInput struct {
|
|
RecordID, ShopeeSKUID, GoodsID, SpecRaw, SpecKey string
|
|
Color, Size, Advice, SKUCode string
|
|
ParseOK bool
|
|
Source, ObservedAt, Now, UpdatePolicy string
|
|
}
|
|
|
|
type catalogSKUStored struct {
|
|
RecordID, GoodsID, SpecRaw, SpecKey string
|
|
ShopeeSKUID, Color, Size, Advice, SKUCode sql.NullString
|
|
ParseOK, IsManual int
|
|
Source, ObservedAt sql.NullString
|
|
FieldSources, FieldObservedAt sql.NullString
|
|
}
|
|
|
|
func scanCatalogSKU(dbRow *sql.Row) (*catalogSKUStored, error) {
|
|
var row catalogSKUStored
|
|
err := dbRow.Scan(&row.RecordID, &row.ShopeeSKUID, &row.GoodsID, &row.SpecRaw, &row.SpecKey, &row.Color, &row.Size, &row.Advice, &row.ParseOK, &row.SKUCode, &row.IsManual, &row.Source, &row.ObservedAt, &row.FieldSources, &row.FieldObservedAt)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &row, nil
|
|
}
|
|
|
|
func findCatalogSKUBySpec(q Execer, goodsID, specKey string) (*catalogSKUStored, error) {
|
|
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE goods_id=? AND spec_key=?`, goodsID, specKey))
|
|
}
|
|
|
|
func findCatalogSKUBySpecForUpdate(q Execer, goodsID, specKey string) (*catalogSKUStored, error) {
|
|
// 规格唯一键冲突后必须使用当前读。普通 SELECT 在 MySQL 默认的
|
|
// REPEATABLE READ 事务中会继续使用第一次查询的快照，看不到竞争方
|
|
// 刚刚提交的 SKU，导致无法恢复本来可幂等的并发写入。
|
|
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE goods_id=? AND spec_key=? FOR UPDATE`, goodsID, specKey))
|
|
}
|
|
|
|
func findCatalogSKUByExternalID(q Execer, skuID string) (*catalogSKUStored, error) {
|
|
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE shopee_sku_id=?`, skuID))
|
|
}
|
|
|
|
func isShopeeSKUSpecUniqueConflict(err error) bool {
|
|
var mysqlErr *mysql.MySQLError
|
|
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 &&
|
|
strings.Contains(mysqlErr.Message, "uq_shopee_skus_spec")
|
|
}
|
|
|
|
// UpsertCatalogShopeeSKU 用真实 ID 或商品规格身份定位同一内部记录，人工行永不覆盖。
|
|
func UpsertCatalogShopeeSKU(q Execer, in CatalogShopeeSKUInput) (CatalogSKUOutcome, error) {
|
|
bySpec, err := findCatalogSKUBySpec(q, in.GoodsID, in.SpecKey)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var byID *catalogSKUStored
|
|
if in.ShopeeSKUID != "" {
|
|
byID, err = findCatalogSKUByExternalID(q, in.ShopeeSKUID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
if byID != nil && (byID.GoodsID != in.GoodsID || byID.SpecKey != in.SpecKey) {
|
|
return "", fmt.Errorf("真实蝦皮 SKU %s 已属于其他商品或规格", in.ShopeeSKUID)
|
|
}
|
|
if byID != nil && bySpec != nil && byID.RecordID != bySpec.RecordID {
|
|
return "", fmt.Errorf("商品 %s 的规格 %s 已对应另一个真实蝦皮 SKU", in.GoodsID, in.SpecRaw)
|
|
}
|
|
current := byID
|
|
if current == nil {
|
|
current = bySpec
|
|
}
|
|
parse := 0
|
|
if in.ParseOK {
|
|
parse = 1
|
|
}
|
|
if current == nil {
|
|
fieldSources, fieldTimes := newCatalogFieldProvenance(in)
|
|
sourcesJSON, _ := json.Marshal(fieldSources)
|
|
timesJSON, _ := json.Marshal(fieldTimes)
|
|
_, err = q.Exec(`INSERT INTO shopee_skus(sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,field_sources,field_observed_at,source_observed_at,created_at,updated_at) VALUES(?,NULLIF(?,''),?,?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?,NULLIF(?,''),0,?,?,?,?,?,?)`, in.RecordID, in.ShopeeSKUID, in.GoodsID, in.SpecRaw, in.SpecKey, in.Color, in.Size, in.Advice, parse, in.SKUCode, in.Source, string(sourcesJSON), string(timesJSON), in.ObservedAt, in.Now, in.Now)
|
|
if err == nil {
|
|
return CatalogSKUCreated, nil
|
|
}
|
|
if !isShopeeSKUSpecUniqueConflict(err) {
|
|
return "", err
|
|
}
|
|
|
|
// 目录导入和 SYB 同步可以同时写同一个规格。唯一键已经证明竞争方
|
|
// 已提交该业务身份；重新读取后沿用下面原有的来源优先级规则，不能
|
|
// 把它当成同步失败，也不能用 INSERT IGNORE 静默跳过字段补全。
|
|
current, err = findCatalogSKUBySpecForUpdate(q, in.GoodsID, in.SpecKey)
|
|
if err != nil {
|
|
return "", fmt.Errorf("规格唯一键冲突后重读商品 %s 的 SKU 失败: %w", in.GoodsID, err)
|
|
}
|
|
if current == nil {
|
|
return "", fmt.Errorf("规格唯一键冲突后未找到商品 %s 的规格 %s", in.GoodsID, in.SpecKey)
|
|
}
|
|
}
|
|
if current.ShopeeSKUID.Valid && in.ShopeeSKUID != "" && current.ShopeeSKUID.String != in.ShopeeSKUID {
|
|
return "", fmt.Errorf("商品 %s 的规格 %s 已对应真实蝦皮 SKU %s", in.GoodsID, in.SpecRaw, current.ShopeeSKUID.String)
|
|
}
|
|
if in.UpdatePolicy == "insert_only" {
|
|
return CatalogSKUSkipped, nil
|
|
}
|
|
if current.IsManual != 0 {
|
|
return CatalogSKUManualSkipped, nil
|
|
}
|
|
external := current.ShopeeSKUID.String
|
|
externalAdded := external == "" && in.ShopeeSKUID != ""
|
|
if external == "" {
|
|
external = in.ShopeeSKUID
|
|
}
|
|
if in.UpdatePolicy == "fill_missing" {
|
|
color, size, advice, skuCode := current.Color.String, current.Size.String, current.Advice.String, current.SKUCode.String
|
|
sources, times := catalogFieldProvenance(current)
|
|
changed := external != current.ShopeeSKUID.String
|
|
canFill := func(field, current, incoming string) bool {
|
|
return sources[field] != "manual" && strings.TrimSpace(incoming) != "" && current == ""
|
|
}
|
|
if canFill("color", color, in.Color) {
|
|
color = in.Color
|
|
sources["color"], times["color"] = in.Source, in.ObservedAt
|
|
changed = true
|
|
}
|
|
if canFill("size", size, in.Size) {
|
|
size = in.Size
|
|
sources["size"], times["size"] = in.Source, in.ObservedAt
|
|
changed = true
|
|
}
|
|
if canFill("advice", advice, in.Advice) {
|
|
advice = in.Advice
|
|
sources["advice"], times["advice"] = in.Source, in.ObservedAt
|
|
changed = true
|
|
}
|
|
if canFill("sku_code", skuCode, in.SKUCode) {
|
|
skuCode = in.SKUCode
|
|
sources["sku_code"], times["sku_code"] = in.Source, in.ObservedAt
|
|
changed = true
|
|
}
|
|
if !changed {
|
|
return CatalogSKUSkipped, nil
|
|
}
|
|
sourcesJSON, _ := json.Marshal(sources)
|
|
timesJSON, _ := json.Marshal(times)
|
|
_, err = q.Exec(`UPDATE shopee_skus SET shopee_sku_id=NULLIF(?,''),color=NULLIF(?,''),size=NULLIF(?,''),advice=NULLIF(?,''),sku_code=NULLIF(?,''),parse_ok=CASE WHEN parse_ok=1 THEN 1 ELSE ? END,field_sources=?,field_observed_at=?,source_observed_at=CASE WHEN source='syb' AND ?<>'syb' THEN ? ELSE source_observed_at END,source=CASE WHEN source='syb' AND ?<>'syb' THEN ? ELSE source END,updated_at=? WHERE sku_id=?`, external, color, size, advice, skuCode, parse, string(sourcesJSON), string(timesJSON), in.Source, in.ObservedAt, in.Source, in.Source, in.Now, current.RecordID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return CatalogSKUFilled, nil
|
|
}
|
|
sources, times := catalogFieldProvenance(current)
|
|
color, size, advice, skuCode := current.Color.String, current.Size.String, current.Advice.String, current.SKUCode.String
|
|
changed, stale := false, false
|
|
for _, field := range []struct {
|
|
name string
|
|
incoming string
|
|
current *string
|
|
}{{"color", in.Color, &color}, {"size", in.Size, &size}, {"advice", in.Advice, &advice}, {"sku_code", in.SKUCode, &skuCode}} {
|
|
if sources[field.name] != in.Source {
|
|
continue
|
|
}
|
|
if times[field.name] > in.ObservedAt {
|
|
stale = true
|
|
continue
|
|
}
|
|
*field.current = field.incoming
|
|
times[field.name] = in.ObservedAt
|
|
changed = true
|
|
}
|
|
if !changed {
|
|
if externalAdded {
|
|
_, err = q.Exec(`UPDATE shopee_skus SET shopee_sku_id=?,updated_at=? WHERE sku_id=?`, external, in.Now, current.RecordID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return CatalogSKUFilled, nil
|
|
}
|
|
if stale {
|
|
return CatalogSKUStaleSkipped, nil
|
|
}
|
|
return CatalogSKUSkipped, nil
|
|
}
|
|
sourcesJSON, _ := json.Marshal(sources)
|
|
timesJSON, _ := json.Marshal(times)
|
|
_, err = q.Exec(`UPDATE shopee_skus SET shopee_sku_id=NULLIF(?,''),color=NULLIF(?,''),size=NULLIF(?,''),advice=NULLIF(?,''),parse_ok=?,sku_code=NULLIF(?,''),field_sources=?,field_observed_at=?,source_observed_at=?,updated_at=? WHERE sku_id=?`, external, color, size, advice, parse, skuCode, string(sourcesJSON), string(timesJSON), in.ObservedAt, in.Now, current.RecordID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return CatalogSKUSameSourceUpdated, nil
|
|
}
|
|
|
|
func newCatalogFieldProvenance(in CatalogShopeeSKUInput) (map[string]string, map[string]string) {
|
|
sources, times := map[string]string{}, map[string]string{}
|
|
for name, value := range map[string]string{"color": in.Color, "size": in.Size, "advice": in.Advice, "sku_code": in.SKUCode} {
|
|
if strings.TrimSpace(value) != "" {
|
|
sources[name], times[name] = in.Source, in.ObservedAt
|
|
}
|
|
}
|
|
return sources, times
|
|
}
|
|
|
|
func catalogFieldProvenance(row *catalogSKUStored) (map[string]string, map[string]string) {
|
|
sources, times := map[string]string{}, map[string]string{}
|
|
_ = json.Unmarshal([]byte(row.FieldSources.String), &sources)
|
|
_ = json.Unmarshal([]byte(row.FieldObservedAt.String), ×)
|
|
return sources, times
|
|
}
|
|
|
|
// UpsertCatalogPddProduct 写入结构化 PDD 数据转换后的规范 JSON；空规格不清除既有采集结果。
|
|
func UpsertCatalogPddProduct(q Execer, goodsID, url, title, shopName, skusJSON, observedAt, now, updatePolicy 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 updatePolicy == "insert_only" || (oldObserved.Valid && oldObserved.String > observedAt && updatePolicy != "fill_missing") {
|
|
return false, false, nil
|
|
}
|
|
if skusJSON == "" || updatePolicy == "fill_missing" {
|
|
if updatePolicy == "fill_missing" {
|
|
_, err = q.Exec(`UPDATE pdd_products SET url=CASE WHEN TRIM(COALESCE(url,''))='' THEN ? ELSE url END,title=CASE WHEN TRIM(COALESCE(title,''))='' THEN NULLIF(?,'') ELSE title END,shop_name=CASE WHEN TRIM(COALESCE(shop_name,''))='' THEN NULLIF(?,'') ELSE shop_name END,updated_at=? WHERE goods_id=?`, url, title, shopName, now, goodsID)
|
|
} else {
|
|
_, 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=? AND deleted_at IS NULL`, 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=? AND deleted_at IS NULL`, 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)
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return fmt.Errorf("读取%s结果失败: %w", action, err)
|
|
}
|
|
if affected == 0 {
|
|
return ErrCatalogImportRunNotFound
|
|
}
|
|
return nil
|
|
}
|