feat: 增加档口入库码单表导入 (#231)

2026-08-15 08:47:22 +08:00
parent dd8a71850d
commit 5a9c0f0e6c
6 changed files with 733 additions and 2 deletions
@@ -0,0 +1,84 @@
+package repository
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+
+	"cmautobuy/admin/model"
+)
+
+// InnerCodeImportOutcome 说明幂等导入是新增还是更新。
+type InnerCodeImportOutcome string
+
+const (
+	InnerCodeImportCreated InnerCodeImportOutcome = "created"
+	InnerCodeImportUpdated InnerCodeImportOutcome = "updated"
+)
+
+// UpsertInnerCodeImportRow 按已确认业务键写入一行。
+// 必须在事务中调用；先锁定业务键，避免另一个唯一键冲突时更新错行。
+func UpsertInnerCodeImportRow(tx *sql.Tx, row model.InnerCodeImportRow, now string) (InnerCodeImportOutcome, error) {
+	var id int64
+	err := tx.QueryRow(`
+		SELECT id FROM syb_inner_code_records
+		 WHERE business_date=? AND order_number=? AND stall=? AND spec_key=?
+		 FOR UPDATE`, row.BusinessDate, row.OrderNumber, row.Stall, row.SpecKey).Scan(&id)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return "", fmt.Errorf("锁定档口入库码业务键失败: %w", err)
+	}
+	if errors.Is(err, sql.ErrNoRows) {
+		_, err = tx.Exec(`
+			INSERT INTO syb_inner_code_records (
+				business_date,source_row,print_sequence,order_number,shop_name,stall,
+				spec_raw,spec_key,inner_code,source_duplicate_count,status,
+				created_by_user_id,created_at,updated_at
+			) VALUES (?,?,?,?,?,?,?,?,?,?,'pending',?,?,?)`,
+			row.BusinessDate, row.SourceRow, nullablePositiveInt(row.PrintSequence), row.OrderNumber,
+			nullableString(row.ShopName), row.Stall, row.SpecRaw, row.SpecKey, row.InnerCode,
+			row.SourceDuplicateCount, row.CreatedByUserID, now, now)
+		if err != nil {
+			return "", fmt.Errorf("新增档口入库码记录失败: %w", err)
+		}
+		return InnerCodeImportCreated, nil
+	}
+
+	_, err = tx.Exec(`
+		UPDATE syb_inner_code_records
+		   SET source_row=?,print_sequence=?,shop_name=?,spec_raw=?,inner_code=?,
+		       source_duplicate_count=?,
+		       status=CASE WHEN status IN ('updated','already_filled','applying','needs_check')
+		                   THEN status ELSE 'pending' END,
+		       stock_id=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN stock_id ELSE NULL END,
+		       detail_id=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN detail_id ELSE NULL END,
+		       syb_spec=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN syb_spec ELSE NULL END,
+		       syb_sku=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN syb_sku ELSE NULL END,
+		       syb_variation_sku=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN syb_variation_sku ELSE NULL END,
+		       purchase_platform=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN purchase_platform ELSE NULL END,
+		       purchase_code=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN purchase_code ELSE NULL END,
+		       remote_inner_code=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN remote_inner_code ELSE NULL END,
+		       result_message=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN result_message ELSE NULL END,
+		       planned_at=CASE WHEN status IN ('updated','already_filled','applying','needs_check') THEN planned_at ELSE NULL END,
+		       updated_at=?
+		 WHERE id=?`,
+		row.SourceRow, nullablePositiveInt(row.PrintSequence), nullableString(row.ShopName), row.SpecRaw,
+		row.InnerCode, row.SourceDuplicateCount, now, id)
+	if err != nil {
+		return "", fmt.Errorf("更新档口入库码导入记录失败: %w", err)
+	}
+	return InnerCodeImportUpdated, nil
+}
+
+func nullablePositiveInt(value int) any {
+	if value <= 0 {
+		return nil
+	}
+	return value
+}
+
+func nullableString(value string) any {
+	if value == "" {
+		return nil
+	}
+	return value
+}
@@ -20,7 +20,7 @@ import (
 	"cmautobuy/admin/spec"
 )

-const mysqlSchemaVersion = 22
+const mysqlSchemaVersion = 23

 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -688,10 +688,68 @@ func MigrateMySQL(db *sql.DB) error {
 		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 22, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
 			return fmt.Errorf("记录 MySQL schema v22 失败: %w", err)
 		}
+		current = 22
+	}
+	if current < 23 {
+		if err := migrateMySQLV23(db); err != nil {
+			return fmt.Errorf("执行 MySQL schema v23 失败: %w", err)
+		}
+		if err := checkMySQLV23Shape(db); err != nil {
+			return fmt.Errorf("MySQL schema v23 自检失败，未记录版本: %w", err)
+		}
+		if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 23, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+			return fmt.Errorf("记录 MySQL schema v23 失败: %w", err)
+		}
 	}
 	return CheckMySQLSchema(db)
 }

+// migrateMySQLV23 建立档口入库码最小闭环使用的唯一业务表。
+// Excel 原文件、批次和每次尝试不会另建表保存。
+func migrateMySQLV23(db *sql.DB) error {
+	_, err := db.Exec(`CREATE TABLE IF NOT EXISTS syb_inner_code_records (
+		id BIGINT AUTO_INCREMENT PRIMARY KEY,
+		business_date DATE NOT NULL,
+		source_row INT NOT NULL,
+		print_sequence INT NULL,
+		order_number VARCHAR(64) COLLATE utf8mb4_bin NOT NULL,
+		shop_name VARCHAR(191) NULL,
+		stall VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+		spec_raw VARCHAR(500) NOT NULL,
+		spec_key VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+		inner_code VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
+		source_duplicate_count INT NOT NULL DEFAULT 1,
+		stock_id BIGINT NULL,
+		detail_id BIGINT NULL,
+		syb_spec VARCHAR(500) NULL,
+		syb_sku VARCHAR(191) NULL,
+		syb_variation_sku VARCHAR(191) NULL,
+		purchase_platform VARCHAR(191) NULL,
+		purchase_code VARCHAR(191) NULL,
+		remote_inner_code VARCHAR(128) NULL,
+		status VARCHAR(32) COLLATE utf8mb4_bin NOT NULL DEFAULT 'pending',
+		result_message VARCHAR(500) NULL,
+		created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+		applied_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NULL,
+		planned_at VARCHAR(35) NULL,
+		apply_started_at VARCHAR(35) NULL,
+		applied_at VARCHAR(35) NULL,
+		created_at VARCHAR(35) NOT NULL,
+		updated_at VARCHAR(35) NOT NULL,
+		UNIQUE KEY uq_inner_code_business (business_date,order_number,stall,spec_key),
+		UNIQUE KEY uq_inner_code_date_code (business_date,inner_code),
+		KEY idx_inner_code_list (business_date,status,id DESC),
+		KEY idx_inner_code_order (order_number,business_date,id),
+		KEY idx_inner_code_detail (detail_id,status,id),
+		CONSTRAINT fk_inner_code_created_by FOREIGN KEY (created_by_user_id) REFERENCES users(user_id),
+		CONSTRAINT fk_inner_code_applied_by FOREIGN KEY (applied_by_user_id) REFERENCES users(user_id),
+		CONSTRAINT chk_inner_code_status CHECK (status IN ('pending','ready','applying','updated','already_filled','skipped','failed','needs_check')),
+		CONSTRAINT chk_inner_code_source_row CHECK (source_row > 0),
+		CONSTRAINT chk_inner_code_duplicate_count CHECK (source_duplicate_count > 0)
+	) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`)
+	return err
+}
+
 func migrateMySQLV22(db *sql.DB) error {
 	statements := []string{
 		`CREATE TABLE IF NOT EXISTS ai_match_batches (
@@ -2071,7 +2129,42 @@ func CheckMySQLSchema(db *sql.DB) error {
 	if err := checkMySQLV21Shape(db); err != nil {
 		return err
 	}
-	return checkMySQLV22Shape(db)
+	if err := checkMySQLV22Shape(db); err != nil {
+		return err
+	}
+	return checkMySQLV23Shape(db)
+}
+
+func checkMySQLV23Shape(db *sql.DB) error {
+	if err := checkMySQLSchema(db, []string{"syb_inner_code_records"}); err != nil {
+		return err
+	}
+	for _, name := range []string{"business_date", "order_number", "stall", "spec_key", "inner_code",
+		"source_duplicate_count", "stock_id", "detail_id", "status", "created_by_user_id", "applied_by_user_id"} {
+		exists, err := mysqlColumnExists(db, "syb_inner_code_records", name)
+		if err != nil || !exists {
+			return fmt.Errorf("档口入库码字段 %s 缺失: %v", name, err)
+		}
+	}
+	for _, item := range []struct{ kind, name string }{
+		{"index", "uq_inner_code_business"},
+		{"index", "uq_inner_code_date_code"},
+		{"index", "idx_inner_code_list"},
+		{"constraint", "chk_inner_code_status"},
+		{"constraint", "fk_inner_code_created_by"},
+	} {
+		var exists bool
+		var err error
+		if item.kind == "index" {
+			exists, err = mysqlIndexExists(db, "syb_inner_code_records", item.name)
+		} else {
+			exists, err = mysqlConstraintExists(db, "syb_inner_code_records", item.name)
+		}
+		if err != nil || !exists {
+			return fmt.Errorf("档口入库码%s %s 缺失: %v", item.kind, item.name, err)
+		}
+	}
+	return nil
 }

 func checkMySQLV22Shape(db *sql.DB) error {