feat: 重构顺运宝规格主链路 (#88)

2026-08-10 12:24:23 +08:00
parent 5bb1258462
commit d1165b9e17
28 changed files with 985 additions and 531 deletions
@@ -6,6 +6,7 @@ import (
 	"crypto/x509"
 	"database/sql"
 	"fmt"
+	"log"
 	"net"
 	"os"
 	"strings"
@@ -14,9 +15,11 @@ import (
 	"github.com/go-sql-driver/mysql"

 	"cmautobuy/admin/config"
+	"cmautobuy/admin/model"
+	"cmautobuy/admin/spec"
 )

-const mysqlSchemaVersion = 2
+const mysqlSchemaVersion = 3

 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -370,6 +373,20 @@ var mysqlSchemaV2 = []string{
 	 ON DUPLICATE KEY UPDATE id = VALUES(id)`,
 }

+const mysqlSchemaV3SpecMappings = `CREATE TABLE IF NOT EXISTS spec_mappings (
+	shopee_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+	spec_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+	pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+	pdd_option_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+	pdd_options LONGTEXT NOT NULL,
+	spec_raw TEXT NOT NULL,
+	mapped_at VARCHAR(35) NOT NULL,
+	mapped_by VARCHAR(191),
+	PRIMARY KEY (shopee_goods_id, spec_key, pdd_goods_id),
+	KEY idx_spec_mappings_goods (shopee_goods_id),
+	KEY idx_spec_mappings_pdd (pdd_goods_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`
+
 // MigrateMySQL 建立或升级 MySQL schema。生产迁移只能在这里追加新版本。
 func MigrateMySQL(db *sql.DB) error {
 	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (
@@ -406,21 +423,274 @@ func MigrateMySQL(db *sql.DB) error {
 				return fmt.Errorf("执行 MySQL schema v2 第 %d 条失败: %w", i+1, err)
 			}
 		}
-		if err := CheckMySQLSchema(db); err != nil {
+		mysqlRequiredTablesV2 := append(append([]string{}, requiredTables...), "admin_initialization_lock")
+		if err := checkMySQLSchema(db, mysqlRequiredTablesV2); err != nil {
 			return fmt.Errorf("MySQL schema v2 自检失败，未记录版本: %w", err)
 		}
 		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`,
 			2, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
 			return fmt.Errorf("记录 MySQL schema v2 失败: %w", err)
 		}
+		current = 2
+	}
+	if current < 3 {
+		if err := migrateMySQLV3(db); err != nil {
+			return fmt.Errorf("执行 MySQL schema v3 失败: %w", err)
+		}
+		if err := CheckMySQLSchema(db); err != nil {
+			return fmt.Errorf("MySQL schema v3 自检失败，未记录版本: %w", err)
+		}
+		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`,
+			3, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+			return fmt.Errorf("记录 MySQL schema v3 失败: %w", err)
+		}
 	}
 	return CheckMySQLSchema(db)
 }

-// CheckMySQLSchema 确认所有业务表和关键追加列存在。
+// migrateMySQLV3 把采购规格身份从蝦皮 SKU 改为顺运宝商品规格原文。
+// 每一步都可重放：DDL 已提交但版本尚未记录时，再启动仍会收敛。
+func migrateMySQLV3(db *sql.DB) error {
+	if err := ensureMySQLV3Columns(db); err != nil {
+		return err
+	}
+	if _, err := db.Exec(mysqlSchemaV3SpecMappings); err != nil {
+		return fmt.Errorf("建立 spec_mappings 失败: %w", err)
+	}
+	if err := backfillSybSpecKeys(db); err != nil {
+		return err
+	}
+	if err := ensureSybShopeeSkeletons(db); err != nil {
+		return err
+	}
+	return convertLegacySKUMappings(db)
+}
+
+func ensureMySQLV3Columns(db *sql.DB) error {
+	exists, err := mysqlColumnExists(db, "syb_orders", "spec_key")
+	if err != nil {
+		return err
+	}
+	if !exists {
+		if _, err := db.Exec(`ALTER TABLE syb_orders ADD COLUMN spec_key VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER product_spec`); err != nil {
+			return fmt.Errorf("增加 syb_orders.spec_key 失败: %w", err)
+		}
+	} else if _, err := db.Exec(`ALTER TABLE syb_orders MODIFY COLUMN spec_key VARCHAR(191) COLLATE utf8mb4_bin NULL`); err != nil {
+		return fmt.Errorf("校正 syb_orders.spec_key 结构失败: %w", err)
+	}
+
+	exists, err = mysqlColumnExists(db, "shopee_products", "source")
+	if err != nil {
+		return err
+	}
+	if !exists {
+		if _, err := db.Exec(`ALTER TABLE shopee_products ADD COLUMN source VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'report' AFTER main_sku_code`); err != nil {
+			return fmt.Errorf("增加 shopee_products.source 失败: %w", err)
+		}
+	} else {
+		// 兼容“列已加但还是可空”的 DDL 中断点。先回填旧行，
+		// 再收紧 NOT NULL，否则带 NULL 的存量数据会让 ALTER 失败。
+		if _, err := db.Exec(`UPDATE shopee_products SET source='report' WHERE source IS NULL OR source=''`); err != nil {
+			return fmt.Errorf("回填 shopee_products.source 失败: %w", err)
+		}
+		if _, err := db.Exec(`ALTER TABLE shopee_products MODIFY COLUMN source VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'report'`); err != nil {
+			return fmt.Errorf("校正 shopee_products.source 结构失败: %w", err)
+		}
+	}
+
+	constraintExists, err := mysqlConstraintExists(db, "shopee_products", "chk_shopee_products_source")
+	if err != nil {
+		return err
+	}
+	if !constraintExists {
+		if _, err := db.Exec(`ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_source CHECK (source IN ('report','syb'))`); err != nil {
+			return fmt.Errorf("增加 shopee_products.source CHECK 失败: %w", err)
+		}
+	}
+	return nil
+}
+
+func mysqlColumnExists(db *sql.DB, table, column string) (bool, error) {
+	var count int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns
+		WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, table, column).Scan(&count); err != nil {
+		return false, fmt.Errorf("检查 MySQL 数据表 %s.%s 失败: %w", table, column, err)
+	}
+	return count == 1, nil
+}
+
+func mysqlConstraintExists(db *sql.DB, table, constraint string) (bool, error) {
+	var count int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.table_constraints
+		WHERE constraint_schema=DATABASE() AND table_name=? AND constraint_name=?`, table, constraint).Scan(&count); err != nil {
+		return false, fmt.Errorf("检查 MySQL 约束 %s.%s 失败: %w", table, constraint, err)
+	}
+	return count == 1, nil
+}
+
+func backfillSybSpecKeys(db *sql.DB) error {
+	const batchSize = 500
+	cursor := ""
+	for {
+		rows, err := db.Query(`SELECT syb_id, product_spec FROM syb_orders
+			WHERE syb_id > ? AND spec_key IS NULL AND product_spec IS NOT NULL
+			  AND TRIM(product_spec) <> '' ORDER BY syb_id LIMIT ?`, cursor, batchSize)
+		if err != nil {
+			return fmt.Errorf("读取待回填顺运宝规格失败: %w", err)
+		}
+		type item struct{ id, raw, key string }
+		items := make([]item, 0, batchSize)
+		for rows.Next() {
+			var value item
+			if err := rows.Scan(&value.id, &value.raw); err != nil {
+				rows.Close()
+				return fmt.Errorf("读取待回填顺运宝规格失败: %w", err)
+			}
+			value.key, err = spec.SpecKey(value.raw)
+			if err != nil {
+				rows.Close()
+				return fmt.Errorf("顺运宝明细 %s 的规格不能生成身份键: %w", value.id, err)
+			}
+			items = append(items, value)
+		}
+		if err := rows.Close(); err != nil {
+			return fmt.Errorf("关闭顺运宝规格回填结果失败: %w", err)
+		}
+		if len(items) == 0 {
+			return nil
+		}
+		tx, err := db.Begin()
+		if err != nil {
+			return fmt.Errorf("开始顺运宝规格回填事务失败: %w", err)
+		}
+		for _, value := range items {
+			if _, err := tx.Exec(`UPDATE syb_orders SET spec_key=? WHERE syb_id=? AND spec_key IS NULL`, value.key, value.id); err != nil {
+				tx.Rollback()
+				return fmt.Errorf("回填顺运宝明细 %s 的规格键失败: %w", value.id, err)
+			}
+		}
+		if err := tx.Commit(); err != nil {
+			return fmt.Errorf("提交顺运宝规格回填失败: %w", err)
+		}
+		cursor = items[len(items)-1].id
+	}
+}
+
+func ensureSybShopeeSkeletons(db *sql.DB) error {
+	now := model.NowISO()
+	_, err := db.Exec(`INSERT INTO shopee_products
+		(goods_id, title, source, created_at, updated_at)
+		SELECT shopee_goods_id, MAX(COALESCE(title,'')), 'syb', ?, ?
+		  FROM syb_orders
+		 WHERE shopee_goods_id IS NOT NULL AND shopee_goods_id <> ''
+		 GROUP BY shopee_goods_id
+		ON DUPLICATE KEY UPDATE goods_id=VALUES(goods_id)`, now, now)
+	if err != nil {
+		return fmt.Errorf("按顺运宝数据补建蝦皮商品骨架失败: %w", err)
+	}
+	return nil
+}
+
+func convertLegacySKUMappings(db *sql.DB) error {
+	exists, err := mysqlTableExists(db, "sku_mappings")
+	if err != nil || !exists {
+		return err
+	}
+	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS sku_mappings_v3_backup LIKE sku_mappings`); err != nil {
+		return fmt.Errorf("建立旧规格映射备份表失败: %w", err)
+	}
+	if _, err := db.Exec(`INSERT IGNORE INTO sku_mappings_v3_backup SELECT * FROM sku_mappings`); err != nil {
+		return fmt.Errorf("备份旧规格映射失败: %w", err)
+	}
+
+	rows, err := db.Query(`SELECT m.shopee_sku_id, m.pdd_goods_id, m.pdd_option_key,
+		m.pdd_options, m.mapped_at, m.mapped_by, sk.goods_id, sk.spec_raw
+		FROM sku_mappings m LEFT JOIN shopee_skus sk ON sk.sku_id=m.shopee_sku_id`)
+	if err != nil {
+		return fmt.Errorf("读取旧规格映射失败: %w", err)
+	}
+	type legacy struct {
+		skuID, pddID, optionKey, options, mappedAt string
+		mappedBy, goodsID, raw                     sql.NullString
+	}
+	var records []legacy
+	for rows.Next() {
+		var value legacy
+		if err := rows.Scan(&value.skuID, &value.pddID, &value.optionKey, &value.options,
+			&value.mappedAt, &value.mappedBy, &value.goodsID, &value.raw); err != nil {
+			rows.Close()
+			return fmt.Errorf("读取旧规格映射失败: %w", err)
+		}
+		records = append(records, value)
+	}
+	if err := rows.Close(); err != nil {
+		return fmt.Errorf("关闭旧规格映射结果失败: %w", err)
+	}
+
+	matched, noCurrentMatch, unable := 0, 0, 0
+	for _, value := range records {
+		if !value.goodsID.Valid || !value.raw.Valid {
+			unable++
+			log.Printf("mysql_v3_mapping_unconvertible shopee_sku_id=%s pdd_goods_id=%s", value.skuID, value.pddID)
+			continue
+		}
+		key, err := spec.SpecKey(value.raw.String)
+		if err != nil {
+			unable++
+			log.Printf("mysql_v3_mapping_unconvertible shopee_sku_id=%s pdd_goods_id=%s reason=invalid_spec_key", value.skuID, value.pddID)
+			continue
+		}
+		if err := UpsertSpecMapping(db, model.SpecMapping{
+			ShopeeGoodsID: value.goodsID.String, SpecKey: key, SpecRaw: value.raw.String,
+			PddGoodsID: value.pddID, PddOptionKey: value.optionKey, PddOptions: value.options,
+			MappedAt: value.mappedAt, MappedBy: value.mappedBy.String,
+		}); err != nil {
+			return fmt.Errorf("转换旧规格映射 %s 失败: %w", value.skuID, err)
+		}
+		var hit int
+		if err := db.QueryRow(`SELECT EXISTS(SELECT 1 FROM syb_orders WHERE shopee_goods_id=? AND spec_key=?)`,
+			value.goodsID.String, key).Scan(&hit); err != nil {
+			return fmt.Errorf("验证旧规格映射 %s 命中失败: %w", value.skuID, err)
+		}
+		if hit == 1 {
+			matched++
+		} else {
+			noCurrentMatch++
+			log.Printf("mysql_v3_mapping_no_current_hit shopee_sku_id=%s pdd_goods_id=%s", value.skuID, value.pddID)
+		}
+	}
+	if matched+noCurrentMatch+unable != len(records) {
+		return fmt.Errorf("旧规格映射迁移计数不守恒: 总数=%d 命中=%d 无命中=%d 无法转换=%d",
+			len(records), matched, noCurrentMatch, unable)
+	}
+	log.Printf("mysql_v3_mapping_summary total=%d matched=%d no_current_hit=%d unconvertible=%d backup_table=sku_mappings_v3_backup",
+		len(records), matched, noCurrentMatch, unable)
+	if _, err := db.Exec(`DROP TABLE sku_mappings`); err != nil {
+		return fmt.Errorf("删除已备份的旧规格映射表失败: %w", err)
+	}
+	return nil
+}
+
+func mysqlTableExists(db *sql.DB, table string) (bool, error) {
+	var count int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.tables
+		WHERE table_schema=DATABASE() AND table_name=? AND table_type='BASE TABLE'`, table).Scan(&count); err != nil {
+		return false, fmt.Errorf("检查 MySQL 数据表 %s 失败: %w", table, err)
+	}
+	return count == 1, nil
+}
+
+// CheckMySQLSchema 确认所有业务表、关键追加列和采购身份结构完整。
 func CheckMySQLSchema(db *sql.DB) error {
-	mysqlRequiredTables := append(append([]string{}, requiredTables...), "admin_initialization_lock")
-	return checkMySQLSchema(db, mysqlRequiredTables)
+	mysqlRequiredTables := []string{
+		"shopee_products", "shopee_skus", "pdd_products", "syb_orders", "spec_mappings",
+		"tasks", "clients", "idempotency_keys", "task_claims", "syb_session", "syb_sync_state",
+		"users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock",
+	}
+	if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
+		return err
+	}
+	return checkMySQLV3Shape(db)
 }

 func checkMySQLSchema(db *sql.DB, tables []string) error {
@@ -465,3 +735,76 @@ func checkMySQLSchema(db *sql.DB, tables []string) error {
 	}
 	return nil
 }
+
+func checkMySQLV3Shape(db *sql.DB) error {
+	rows, err := db.Query(`SELECT column_name FROM information_schema.key_column_usage
+		WHERE table_schema=DATABASE() AND table_name='spec_mappings'
+		  AND constraint_name='PRIMARY' ORDER BY ordinal_position`)
+	if err != nil {
+		return fmt.Errorf("检查 spec_mappings 主键失败: %w", err)
+	}
+	var primary []string
+	for rows.Next() {
+		var column string
+		if err := rows.Scan(&column); err != nil {
+			rows.Close()
+			return fmt.Errorf("检查 spec_mappings 主键失败: %w", err)
+		}
+		primary = append(primary, column)
+	}
+	rows.Close()
+	if strings.Join(primary, ",") != "shopee_goods_id,spec_key,pdd_goods_id" {
+		return fmt.Errorf("spec_mappings 主键不正确，实际为 (%s)", strings.Join(primary, ","))
+	}
+	for _, column := range []string{"shopee_goods_id", "spec_key", "pdd_goods_id"} {
+		if err := checkMySQLVarcharColumn(db, "spec_mappings", column, 191, false, "utf8mb4_bin", ""); err != nil {
+			return err
+		}
+	}
+	if err := checkMySQLVarcharColumn(db, "syb_orders", "spec_key", 191, true, "utf8mb4_bin", ""); err != nil {
+		return err
+	}
+	if err := checkMySQLVarcharColumn(db, "shopee_products", "source", 16, false, "utf8mb4_bin", "report"); err != nil {
+		return err
+	}
+	var enforced, checkClause string
+	if err := db.QueryRow(`SELECT tc.enforced, cc.check_clause
+		FROM information_schema.table_constraints tc
+		JOIN information_schema.check_constraints cc
+		  ON cc.constraint_schema=tc.constraint_schema AND cc.constraint_name=tc.constraint_name
+		WHERE tc.constraint_schema=DATABASE() AND tc.table_name='shopee_products'
+		  AND tc.constraint_name='chk_shopee_products_source' AND tc.constraint_type='CHECK'`).Scan(&enforced, &checkClause); err != nil {
+		return fmt.Errorf("shopee_products.source CHECK 缺失或不可读: %w", err)
+	}
+	if enforced != "YES" {
+		return fmt.Errorf("shopee_products.source CHECK 未启用")
+	}
+	normalizedClause := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(checkClause))
+	if !strings.Contains(normalizedClause, "sourcein'report','syb'") {
+		return fmt.Errorf("shopee_products.source CHECK 表达式不正确")
+	}
+	return nil
+}
+
+func checkMySQLVarcharColumn(db *sql.DB, table, column string, length int64, nullable bool, collation, defaultValue string) error {
+	var dataType, isNullable string
+	var actualLength sql.NullInt64
+	var actualCollation, actualDefault sql.NullString
+	if err := db.QueryRow(`SELECT data_type, character_maximum_length, is_nullable, collation_name, column_default
+		FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`,
+		table, column).Scan(&dataType, &actualLength, &isNullable, &actualCollation, &actualDefault); err != nil {
+		return fmt.Errorf("检查 MySQL 列 %s.%s 结构失败: %w", table, column, err)
+	}
+	wantNullable := "NO"
+	if nullable {
+		wantNullable = "YES"
+	}
+	if dataType != "varchar" || !actualLength.Valid || actualLength.Int64 != length ||
+		isNullable != wantNullable || actualCollation.String != collation {
+		return fmt.Errorf("MySQL 列 %s.%s 结构不正确", table, column)
+	}
+	if defaultValue != "" && (!actualDefault.Valid || actualDefault.String != defaultValue) {
+		return fmt.Errorf("MySQL 列 %s.%s 默认值不正确", table, column)
+	}
+	return nil
+}