feat: 支持无SKU目录导入策略 (#141)
This commit is contained in:
@@ -2,6 +2,7 @@ package repository
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
@@ -21,9 +22,9 @@ var (
|
||||
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,
|
||||
last_request_at,created_at
|
||||
) VALUES (?,?,?,?,1,0,NULLIF(?,''),?,?)`, run.Source, run.BatchID, run.RequestHash,
|
||||
run.Status, run.ObservedAt, run.LastRequestAt, run.CreatedAt)
|
||||
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
|
||||
}
|
||||
@@ -42,13 +43,15 @@ func GetCatalogImportRun(q Execer, source, batchID string) (model.CatalogImportR
|
||||
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,last_request_at,last_conflict_at,shopee_created,shopee_updated,sku_created,
|
||||
sku_updated,pdd_created,pdd_updated,association_created,association_unchanged,failure_count,
|
||||
observed_at,update_policy,last_request_at,last_conflict_at,shopee_created,shopee_updated,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.LastRequestAt, &lastConflictAt, &run.ShopeeCreated, &run.ShopeeUpdated,
|
||||
&run.SKUCreated, &run.SKUUpdated, &run.PddCreated, &run.PddUpdated, &run.AssociationCreated,
|
||||
&observedAt, &run.UpdatePolicy, &run.LastRequestAt, &lastConflictAt, &run.ShopeeCreated, &run.ShopeeUpdated,
|
||||
&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
|
||||
@@ -147,10 +150,12 @@ func RecordCatalogImportConflict(q Execer, source, batchID, requestedAt string)
|
||||
// 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=?,
|
||||
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.SKUCreated, run.SKUUpdated, run.PddCreated, run.PddUpdated, run.AssociationCreated,
|
||||
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, "完成商品目录批次")
|
||||
@@ -177,34 +182,194 @@ func UpsertCatalogShopeeProduct(q Execer, goodsID, title, status, mainSKU, obser
|
||||
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
|
||||
}
|
||||
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) {
|
||||
_, 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
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, false, err
|
||||
return nil, err
|
||||
}
|
||||
if oldGoods != goodsID {
|
||||
return false, false, fmt.Errorf("SKU %s 已属于蝦皮商品 %s", skuID, oldGoods)
|
||||
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 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))
|
||||
}
|
||||
|
||||
// UpsertCatalogShopeeSKU 用真实 ID 或商品规格身份定位同一内部记录,人工行永不覆盖。
|
||||
func UpsertCatalogShopeeSKU(q Execer, in CatalogShopeeSKUInput) (CatalogSKUOutcome, error) {
|
||||
bySpec, err := findCatalogSKUBySpec(q, in.GoodsID, in.SpecKey)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if oldObserved.Valid && oldObserved.String > observedAt {
|
||||
return false, false, nil
|
||||
var byID *catalogSKUStored
|
||||
if in.ShopeeSKUID != "" {
|
||||
byID, err = findCatalogSKUByExternalID(q, in.ShopeeSKUID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
_, 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
|
||||
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 "", err
|
||||
}
|
||||
return CatalogSKUCreated, nil
|
||||
}
|
||||
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
|
||||
if color == "" && strings.TrimSpace(in.Color) != "" {
|
||||
color = in.Color
|
||||
sources["color"], times["color"] = in.Source, in.ObservedAt
|
||||
changed = true
|
||||
}
|
||||
if size == "" && strings.TrimSpace(in.Size) != "" {
|
||||
size = in.Size
|
||||
sources["size"], times["size"] = in.Source, in.ObservedAt
|
||||
changed = true
|
||||
}
|
||||
if advice == "" && strings.TrimSpace(in.Advice) != "" {
|
||||
advice = in.Advice
|
||||
sources["advice"], times["advice"] = in.Source, in.ObservedAt
|
||||
changed = true
|
||||
}
|
||||
if skuCode == "" && strings.TrimSpace(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=?,updated_at=? WHERE sku_id=?`, external, color, size, advice, skuCode, parse, string(sourcesJSON), string(timesJSON), 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;空规格不清除既有采集结果。
|
||||
|
||||
@@ -19,9 +19,9 @@ func TestCatalogImportRun_登记查询重放和冲突(t *testing.T) {
|
||||
defer db.Close()
|
||||
if _, err := db.Exec(`CREATE TABLE catalog_import_runs (
|
||||
source TEXT NOT NULL,batch_id TEXT NOT NULL,request_hash TEXT NOT NULL,status TEXT NOT NULL,
|
||||
request_count INTEGER NOT NULL,conflict_count INTEGER NOT NULL,observed_at TEXT,last_request_at TEXT NOT NULL,
|
||||
request_count INTEGER NOT NULL,conflict_count INTEGER NOT NULL,observed_at TEXT,update_policy TEXT NOT NULL DEFAULT 'fill_missing',last_request_at TEXT NOT NULL,
|
||||
last_conflict_at TEXT,shopee_created INTEGER NOT NULL DEFAULT 0,shopee_updated INTEGER NOT NULL DEFAULT 0,
|
||||
sku_created INTEGER NOT NULL DEFAULT 0,sku_updated INTEGER NOT NULL DEFAULT 0,pdd_created INTEGER NOT NULL DEFAULT 0,
|
||||
sku_created INTEGER NOT NULL DEFAULT 0,sku_updated INTEGER NOT NULL DEFAULT 0,sku_filled INTEGER NOT NULL DEFAULT 0,sku_same_source_updated INTEGER NOT NULL DEFAULT 0,sku_skipped INTEGER NOT NULL DEFAULT 0,sku_manual_skipped INTEGER NOT NULL DEFAULT 0,sku_stale_skipped INTEGER NOT NULL DEFAULT 0,pdd_created INTEGER NOT NULL DEFAULT 0,
|
||||
pdd_updated INTEGER NOT NULL DEFAULT 0,association_created INTEGER NOT NULL DEFAULT 0,
|
||||
association_unchanged INTEGER NOT NULL DEFAULT 0,failure_count INTEGER NOT NULL DEFAULT 0,
|
||||
error_summary TEXT,response_body TEXT,created_at TEXT NOT NULL,finished_at TEXT,
|
||||
|
||||
@@ -19,7 +19,7 @@ import (
|
||||
"cmautobuy/admin/spec"
|
||||
)
|
||||
|
||||
const mysqlSchemaVersion = 7
|
||||
const mysqlSchemaVersion = 8
|
||||
|
||||
// OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
|
||||
func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
|
||||
@@ -507,10 +507,115 @@ func MigrateMySQL(db *sql.DB) error {
|
||||
if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 7, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
|
||||
return fmt.Errorf("记录 MySQL schema v7 失败: %w", err)
|
||||
}
|
||||
current = 7
|
||||
}
|
||||
if current < 8 {
|
||||
if err := migrateMySQLV8(db); err != nil {
|
||||
return fmt.Errorf("执行 MySQL schema v8 失败: %w", err)
|
||||
}
|
||||
if err := checkMySQLV8Shape(db); err != nil {
|
||||
return fmt.Errorf("MySQL schema v8 自检失败,未记录版本: %w", err)
|
||||
}
|
||||
if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 8, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
|
||||
return fmt.Errorf("记录 MySQL schema v8 失败: %w", err)
|
||||
}
|
||||
}
|
||||
return CheckMySQLSchema(db)
|
||||
}
|
||||
|
||||
// migrateMySQLV8 保留 sku_id 作为内部主键,并增加可空的真实蝦皮 SKU ID。
|
||||
// 每个 DDL 都先检查形状,支持 MySQL 隐式提交后的中断重放。
|
||||
func migrateMySQLV8(db *sql.DB) error {
|
||||
columns := []struct{ table, name, ddl string }{
|
||||
{"shopee_skus", "shopee_sku_id", `ALTER TABLE shopee_skus ADD COLUMN shopee_sku_id VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER sku_id`},
|
||||
{"shopee_skus", "spec_key", `ALTER TABLE shopee_skus ADD COLUMN spec_key VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER spec_raw`},
|
||||
{"shopee_skus", "source", `ALTER TABLE shopee_skus ADD COLUMN source VARCHAR(64) COLLATE utf8mb4_bin NULL AFTER is_manual`},
|
||||
{"shopee_skus", "field_sources", `ALTER TABLE shopee_skus ADD COLUMN field_sources JSON NULL AFTER source`},
|
||||
{"shopee_skus", "field_observed_at", `ALTER TABLE shopee_skus ADD COLUMN field_observed_at JSON NULL AFTER field_sources`},
|
||||
{"catalog_import_runs", "update_policy", `ALTER TABLE catalog_import_runs ADD COLUMN update_policy VARCHAR(32) COLLATE utf8mb4_bin NOT NULL DEFAULT 'fill_missing' AFTER observed_at`},
|
||||
{"catalog_import_runs", "sku_filled", `ALTER TABLE catalog_import_runs ADD COLUMN sku_filled INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_updated`},
|
||||
{"catalog_import_runs", "sku_same_source_updated", `ALTER TABLE catalog_import_runs ADD COLUMN sku_same_source_updated INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_filled`},
|
||||
{"catalog_import_runs", "sku_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN sku_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_same_source_updated`},
|
||||
{"catalog_import_runs", "sku_manual_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN sku_manual_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_skipped`},
|
||||
{"catalog_import_runs", "sku_stale_skipped", `ALTER TABLE catalog_import_runs ADD COLUMN sku_stale_skipped INT UNSIGNED NOT NULL DEFAULT 0 AFTER sku_manual_skipped`},
|
||||
}
|
||||
for _, column := range columns {
|
||||
exists, err := mysqlColumnExists(db, column.table, column.name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !exists {
|
||||
if _, err := db.Exec(column.ddl); err != nil {
|
||||
return fmt.Errorf("增加 %s.%s 失败: %w", column.table, column.name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := db.Exec(`UPDATE shopee_skus SET shopee_sku_id=sku_id WHERE shopee_sku_id IS NULL AND sku_id NOT LIKE 'catalog:%'`); err != nil {
|
||||
return fmt.Errorf("回填真实蝦皮 SKU ID 失败: %w", err)
|
||||
}
|
||||
if _, err := db.Exec(`UPDATE shopee_skus SET source='report' WHERE source IS NULL OR source=''`); err != nil {
|
||||
return fmt.Errorf("回填 SKU 来源失败: %w", err)
|
||||
}
|
||||
if _, err := db.Exec(`UPDATE shopee_skus SET field_sources=JSON_OBJECT('color',IF(color IS NULL OR color='','',COALESCE(source,'report')),'size',IF(size IS NULL OR size='','',COALESCE(source,'report')),'advice',IF(advice IS NULL OR advice='','',COALESCE(source,'report')),'sku_code',IF(sku_code IS NULL OR sku_code='','',COALESCE(source,'report'))) WHERE field_sources IS NULL`); err != nil {
|
||||
return fmt.Errorf("回填 SKU 字段来源失败: %w", err)
|
||||
}
|
||||
if _, err := db.Exec(`UPDATE shopee_skus SET field_observed_at=JSON_OBJECT('color',IF(color IS NULL OR color='','',COALESCE(source_observed_at,updated_at)),'size',IF(size IS NULL OR size='','',COALESCE(source_observed_at,updated_at)),'advice',IF(advice IS NULL OR advice='','',COALESCE(source_observed_at,updated_at)),'sku_code',IF(sku_code IS NULL OR sku_code='','',COALESCE(source_observed_at,updated_at))) WHERE field_observed_at IS NULL`); err != nil {
|
||||
return fmt.Errorf("回填 SKU 字段观测时间失败: %w", err)
|
||||
}
|
||||
rows, err := db.Query(`SELECT sku_id,spec_raw FROM shopee_skus WHERE spec_key IS NULL OR spec_key=''`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取待回填规格键失败: %w", err)
|
||||
}
|
||||
type pendingKey struct{ id, raw string }
|
||||
var pending []pendingKey
|
||||
for rows.Next() {
|
||||
var row pendingKey
|
||||
if err := rows.Scan(&row.id, &row.raw); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
pending = append(pending, row)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, row := range pending {
|
||||
key, keyErr := spec.SpecKey(row.raw)
|
||||
if keyErr != nil {
|
||||
return fmt.Errorf("SKU 内部记录 %s 无法生成规格键: %w", row.id, keyErr)
|
||||
}
|
||||
if len([]rune(key)) > 191 {
|
||||
return fmt.Errorf("SKU 内部记录 %s 的规格键超过 191 个字符", row.id)
|
||||
}
|
||||
if _, err := db.Exec(`UPDATE shopee_skus SET spec_key=? WHERE sku_id=?`, key, row.id); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
var duplicate string
|
||||
err = db.QueryRow(`SELECT CONCAT(goods_id,' / ',spec_key) FROM shopee_skus GROUP BY goods_id,spec_key HAVING COUNT(*)>1 LIMIT 1`).Scan(&duplicate)
|
||||
if err != nil && err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
if err == nil {
|
||||
return fmt.Errorf("存在同商品重复规格 %s,请人工确认真实 SKU 后再迁移", duplicate)
|
||||
}
|
||||
for _, index := range []struct{ name, ddl string }{
|
||||
{"uq_shopee_skus_external", `ALTER TABLE shopee_skus ADD UNIQUE KEY uq_shopee_skus_external (shopee_sku_id)`},
|
||||
{"uq_shopee_skus_spec", `ALTER TABLE shopee_skus ADD UNIQUE KEY uq_shopee_skus_spec (goods_id,spec_key)`},
|
||||
} {
|
||||
exists, err := mysqlIndexExists(db, "shopee_skus", index.name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !exists {
|
||||
if _, err := db.Exec(index.ddl); err != nil {
|
||||
return fmt.Errorf("增加索引 %s 失败: %w", index.name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
const mysqlSchemaV7CatalogRuns = `CREATE TABLE IF NOT EXISTS catalog_import_runs (
|
||||
source VARCHAR(64) COLLATE utf8mb4_bin NOT NULL,
|
||||
batch_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
|
||||
@@ -948,7 +1053,49 @@ func CheckMySQLSchema(db *sql.DB) error {
|
||||
if err := checkMySQLV6Shape(db); err != nil {
|
||||
return err
|
||||
}
|
||||
return checkMySQLV7Shape(db)
|
||||
if err := checkMySQLV7Shape(db); err != nil {
|
||||
return err
|
||||
}
|
||||
return checkMySQLV8Shape(db)
|
||||
}
|
||||
|
||||
func checkMySQLV8Shape(db *sql.DB) error {
|
||||
for _, column := range []struct {
|
||||
name string
|
||||
length int64
|
||||
nullable bool
|
||||
collation, def string
|
||||
}{
|
||||
{"shopee_sku_id", 191, true, "utf8mb4_bin", ""},
|
||||
{"spec_key", 191, true, "utf8mb4_bin", ""},
|
||||
{"source", 64, true, "utf8mb4_bin", ""},
|
||||
} {
|
||||
if err := checkMySQLVarcharColumn(db, "shopee_skus", column.name, column.length, column.nullable, column.collation, column.def); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, name := range []string{"field_sources", "field_observed_at"} {
|
||||
exists, err := mysqlColumnExists(db, "shopee_skus", name)
|
||||
if err != nil || !exists {
|
||||
return fmt.Errorf("蝦皮 SKU 字段 %s 缺失", name)
|
||||
}
|
||||
}
|
||||
if err := checkMySQLVarcharColumn(db, "catalog_import_runs", "update_policy", 32, false, "utf8mb4_bin", "fill_missing"); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, name := range []string{"sku_filled", "sku_same_source_updated", "sku_skipped", "sku_manual_skipped", "sku_stale_skipped"} {
|
||||
exists, err := mysqlColumnExists(db, "catalog_import_runs", name)
|
||||
if err != nil || !exists {
|
||||
return fmt.Errorf("商品目录导入记录列 %s 缺失", name)
|
||||
}
|
||||
}
|
||||
for _, name := range []string{"uq_shopee_skus_external", "uq_shopee_skus_spec"} {
|
||||
exists, err := mysqlIndexExists(db, "shopee_skus", name)
|
||||
if err != nil || !exists {
|
||||
return fmt.Errorf("蝦皮 SKU 唯一索引 %s 缺失", name)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkMySQLV7Shape(db *sql.DB) error {
|
||||
|
||||
@@ -382,6 +382,53 @@ func TestMySQLMigrate_V7形状错误不记版本(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMySQLMigrate_V7升级V8保留内部引用且可重放(t *testing.T) {
|
||||
db := openMySQLMigrationTestDB(t)
|
||||
defer db.Close()
|
||||
cleanMySQLTestSchema(t, db)
|
||||
defer cleanMySQLTestSchema(t, db)
|
||||
prepareMySQLV7(t, db)
|
||||
now := "2026-08-11T00:00:00Z"
|
||||
mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES('S-1','商品','report',?,?)`, now, now)
|
||||
mustExec(t, db, `INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,created_at,updated_at) VALUES('REAL-1','S-1',' 黑色, M ',?,?)`, now, now)
|
||||
// 模拟全部 DDL/回填已提交,但 schema_migrations 尚未记录 v8。
|
||||
if err := migrateMySQLV8(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := MigrateMySQL(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := MigrateMySQL(db); err != nil {
|
||||
t.Fatalf("v8 重跑失败: %v", err)
|
||||
}
|
||||
var internalID, externalID, key string
|
||||
if err := db.QueryRow(`SELECT sku_id,shopee_sku_id,spec_key FROM shopee_skus`).Scan(&internalID, &externalID, &key); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if internalID != "REAL-1" || externalID != "REAL-1" || key != "黑色, M" {
|
||||
t.Fatalf("历史身份未保留:internal=%q external=%q key=%q", internalID, externalID, key)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMySQLMigrate_V8重复规格停止且不记版本(t *testing.T) {
|
||||
db := openMySQLMigrationTestDB(t)
|
||||
defer db.Close()
|
||||
cleanMySQLTestSchema(t, db)
|
||||
defer cleanMySQLTestSchema(t, db)
|
||||
prepareMySQLV7(t, db)
|
||||
now := "2026-08-11T00:00:00Z"
|
||||
mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES('S-1','商品','report',?,?)`, now, now)
|
||||
mustExec(t, db, `INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,created_at,updated_at) VALUES('A','S-1','黑色,M',?,?),('B','S-1','黑色,M',?,?)`, now, now, now, now)
|
||||
if err := MigrateMySQL(db); err == nil || !strings.Contains(err.Error(), "重复规格") {
|
||||
t.Fatalf("应明确阻止冲突迁移:%v", err)
|
||||
}
|
||||
var count int
|
||||
db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=8`).Scan(&count)
|
||||
if count != 0 {
|
||||
t.Fatal("冲突迁移不得记录 v8")
|
||||
}
|
||||
}
|
||||
|
||||
func openMySQLMigrationTestDB(t *testing.T) *sql.DB {
|
||||
t.Helper()
|
||||
if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" {
|
||||
@@ -436,6 +483,15 @@ func prepareMySQLV6(t *testing.T, db *sql.DB) {
|
||||
VALUES (5,'2026-08-10T00:00:00Z'),(6,'2026-08-10T00:00:00Z')`)
|
||||
}
|
||||
|
||||
func prepareMySQLV7(t *testing.T, db *sql.DB) {
|
||||
t.Helper()
|
||||
prepareMySQLV6(t, db)
|
||||
if err := migrateMySQLV7(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (7,'2026-08-11T00:00:00Z')`)
|
||||
}
|
||||
|
||||
func mustExec(t *testing.T, db *sql.DB, query string, args ...any) {
|
||||
t.Helper()
|
||||
if _, err := db.Exec(query, args...); err != nil {
|
||||
|
||||
+33
-18
@@ -2,11 +2,13 @@ package repository
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"cmautobuy/admin/model"
|
||||
"cmautobuy/admin/spec"
|
||||
)
|
||||
|
||||
// UpsertShopeeProduct 写入或更新一条蝦皮商品汇总行。
|
||||
@@ -57,22 +59,35 @@ func UpsertShopeeSKU(q Execer, skuID, goodsID, specRaw, color, size, advice stri
|
||||
parseOKInt = 1
|
||||
}
|
||||
now := model.NowISO()
|
||||
_, err := q.Exec(`
|
||||
specKey, err := spec.SpecKey(specRaw)
|
||||
if err != nil {
|
||||
return fmt.Errorf("规格原文无效: %w", err)
|
||||
}
|
||||
fieldSources, fieldTimes := map[string]string{}, map[string]string{}
|
||||
for name, value := range map[string]string{"color": color, "size": size, "advice": advice, "sku_code": skuCode} {
|
||||
if strings.TrimSpace(value) != "" {
|
||||
fieldSources[name], fieldTimes[name] = "report", now
|
||||
}
|
||||
}
|
||||
fieldSourcesJSON, _ := json.Marshal(fieldSources)
|
||||
fieldTimesJSON, _ := json.Marshal(fieldTimes)
|
||||
_, err = q.Exec(`
|
||||
INSERT INTO shopee_skus
|
||||
(sku_id, goods_id, spec_raw, color, size, advice, parse_ok, sku_code,
|
||||
is_manual, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)
|
||||
(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, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 'report', ?, ?, ?, ?)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
goods_id = VALUES(goods_id),
|
||||
spec_raw = VALUES(spec_raw),
|
||||
color = VALUES(color),
|
||||
size = VALUES(size),
|
||||
advice = VALUES(advice),
|
||||
parse_ok = VALUES(parse_ok),
|
||||
sku_code = VALUES(sku_code),
|
||||
updated_at = VALUES(updated_at)`,
|
||||
// is_manual 不在上面的 SET 列表里,SQLite 对没提到的列保持原值不变。
|
||||
skuID, goodsID, specRaw, color, size, advice, parseOKInt, skuCode, now, now)
|
||||
goods_id = CASE WHEN is_manual=1 THEN goods_id ELSE VALUES(goods_id) END,
|
||||
color = CASE WHEN is_manual=1 THEN color ELSE VALUES(color) END,
|
||||
size = CASE WHEN is_manual=1 THEN size ELSE VALUES(size) END,
|
||||
advice = CASE WHEN is_manual=1 THEN advice ELSE VALUES(advice) END,
|
||||
parse_ok = CASE WHEN is_manual=1 THEN parse_ok ELSE VALUES(parse_ok) END,
|
||||
sku_code = CASE WHEN is_manual=1 THEN sku_code ELSE VALUES(sku_code) END,
|
||||
field_sources = CASE WHEN is_manual=1 THEN field_sources ELSE VALUES(field_sources) END,
|
||||
field_observed_at = CASE WHEN is_manual=1 THEN field_observed_at ELSE VALUES(field_observed_at) END,
|
||||
updated_at = CASE WHEN is_manual=1 THEN updated_at ELSE VALUES(updated_at) END`,
|
||||
// 人工行的业务字段和更新时间均由 CASE 保留。
|
||||
skuID, skuID, goodsID, specRaw, specKey, color, size, advice, parseOKInt, skuCode, string(fieldSourcesJSON), string(fieldTimesJSON), now, now)
|
||||
if err != nil {
|
||||
return fmt.Errorf("写入蝦皮 SKU %s 失败: %w", skuID, err)
|
||||
}
|
||||
@@ -293,7 +308,7 @@ func GetShopeeCollectStatus(q Execer, pddGoodsID string) (status, msg sql.NullSt
|
||||
// 见 #41「弹窗里待补的行必须显示 spec_raw 原文」。
|
||||
func ListShopeeSKUsByGoodsID(q Execer, goodsID string) ([]model.ShopeeSKU, error) {
|
||||
rows, err := q.Query(`
|
||||
SELECT sku_id, goods_id, spec_raw, color, size, advice, parse_ok, sku_code,
|
||||
SELECT sku_id, COALESCE(shopee_sku_id,''), goods_id, spec_raw, color, size, advice, parse_ok, sku_code,
|
||||
is_manual, created_at, updated_at
|
||||
FROM shopee_skus
|
||||
WHERE goods_id = ?
|
||||
@@ -309,7 +324,7 @@ func ListShopeeSKUsByGoodsID(q Execer, goodsID string) ([]model.ShopeeSKU, error
|
||||
var color, size, advice, skuCode sql.NullString
|
||||
var parseOK, isManual int
|
||||
if err := rows.Scan(
|
||||
&sk.SKUID, &sk.GoodsID, &sk.SpecRaw, &color, &size, &advice,
|
||||
&sk.RecordID, &sk.SKUID, &sk.GoodsID, &sk.SpecRaw, &color, &size, &advice,
|
||||
&parseOK, &skuCode, &isManual, &sk.CreatedAt, &sk.UpdatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("读取蝦皮商品 %s 的规格失败: %w", goodsID, err)
|
||||
@@ -331,10 +346,10 @@ func GetShopeeSKUByID(q Execer, skuID string) (*model.ShopeeSKU, error) {
|
||||
var color, size, advice, skuCode sql.NullString
|
||||
var parseOK, isManual int
|
||||
err := q.QueryRow(`
|
||||
SELECT sku_id, goods_id, spec_raw, color, size, advice, parse_ok, sku_code,
|
||||
SELECT sku_id, COALESCE(shopee_sku_id,''), goods_id, spec_raw, color, size, advice, parse_ok, sku_code,
|
||||
is_manual, created_at, updated_at
|
||||
FROM shopee_skus WHERE sku_id = ?`, skuID).Scan(
|
||||
&sk.SKUID, &sk.GoodsID, &sk.SpecRaw, &color, &size, &advice,
|
||||
&sk.RecordID, &sk.SKUID, &sk.GoodsID, &sk.SpecRaw, &color, &size, &advice,
|
||||
&parseOK, &skuCode, &isManual, &sk.CreatedAt, &sk.UpdatedAt,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
|
||||
@@ -5,11 +5,14 @@ import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"cmautobuy/admin/spec"
|
||||
)
|
||||
|
||||
// SQLiteMigrationSummary 只包含表名和数量,可以安全写入终端或工单。
|
||||
@@ -117,6 +120,9 @@ func MigrateSQLiteToMySQL(source, target *sql.DB, dryRun bool) (*SQLiteMigration
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := normalizeMigratedShopeeSKUs(tx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return nil, &SQLiteMigrationError{Stage: "无法提交 MySQL 导入事务", Cause: err}
|
||||
}
|
||||
@@ -126,6 +132,52 @@ func MigrateSQLiteToMySQL(source, target *sql.DB, dryRun bool) (*SQLiteMigration
|
||||
return summary, nil
|
||||
}
|
||||
|
||||
// normalizeMigratedShopeeSKUs 补齐 MySQL v8 新增、历史 SQLite 没有的规格身份列。
|
||||
func normalizeMigratedShopeeSKUs(tx *sql.Tx) error {
|
||||
rows, err := tx.Query(`SELECT sku_id,spec_raw,color,size,advice,sku_code,updated_at FROM shopee_skus`)
|
||||
if err != nil {
|
||||
return &SQLiteMigrationError{Stage: "无法读取待转换 SKU", Table: "shopee_skus", Cause: err}
|
||||
}
|
||||
type item struct {
|
||||
id, raw, updated string
|
||||
color, size, advice, code sql.NullString
|
||||
}
|
||||
var list []item
|
||||
for rows.Next() {
|
||||
var row item
|
||||
if err := rows.Scan(&row.id, &row.raw, &row.color, &row.size, &row.advice, &row.code, &row.updated); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
list = append(list, row)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, row := range list {
|
||||
key, err := spec.SpecKey(row.raw)
|
||||
if err != nil {
|
||||
return &SQLiteMigrationError{Stage: "历史 SKU 规格原文无效", Table: "shopee_skus", Cause: err}
|
||||
}
|
||||
if len([]rune(key)) > 191 {
|
||||
return &SQLiteMigrationError{Stage: "历史 SKU 规格键过长", Table: "shopee_skus"}
|
||||
}
|
||||
sources, times := map[string]string{}, map[string]string{}
|
||||
for name, value := range map[string]string{"color": row.color.String, "size": row.size.String, "advice": row.advice.String, "sku_code": row.code.String} {
|
||||
if value != "" {
|
||||
sources[name] = "report"
|
||||
times[name] = row.updated
|
||||
}
|
||||
}
|
||||
sourcesJSON, _ := json.Marshal(sources)
|
||||
timesJSON, _ := json.Marshal(times)
|
||||
if _, err := tx.Exec(`UPDATE shopee_skus SET shopee_sku_id=sku_id,spec_key=?,source='report',field_sources=?,field_observed_at=? WHERE sku_id=?`, key, string(sourcesJSON), string(timesJSON), row.id); err != nil {
|
||||
return &SQLiteMigrationError{Stage: "无法补齐 MySQL v8 SKU 身份", Table: "shopee_skus", Cause: err}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// VerifySQLiteToMySQL 比较逐表数量和全部迁移列的确定性摘要,不输出业务内容。
|
||||
func VerifySQLiteToMySQL(source, target *sql.DB) (*SQLiteMigrationSummary, error) {
|
||||
if err := validateLegacySQLite(source); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user