From 107ee5ce85cc5b0eda993af21701092f6c5c6a8f Mon Sep 17 00:00:00 2001 From: chengma Date: Tue, 11 Aug 2026 10:00:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=BB=BA=E7=AB=8B=E5=95=86=E5=93=81?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E6=8E=A5=E5=85=A5=E5=9F=BA=E7=A1=80=20(#132)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- admin/config.example.yaml | 10 ++ admin/config/config.go | 71 +++++++- admin/config/config_test.go | 26 +++ admin/handler/integration/auth.go | 60 +++++++ admin/handler/integration/auth_test.go | 47 ++++++ admin/model/catalog_import.go | 36 ++++ admin/repository/catalog_import.go | 94 +++++++++++ admin/repository/catalog_import_test.go | 52 ++++++ admin/repository/mysql_db.go | 157 +++++++++++++++++- admin/repository/mysql_db_integration_test.go | 55 ++++++ docs/admin/00-glossary.md | 2 + docs/admin/03-data-model.md | 24 ++- docs/admin/06-quality-security.md | 4 + 13 files changed, 631 insertions(+), 7 deletions(-) create mode 100644 admin/handler/integration/auth.go create mode 100644 admin/handler/integration/auth_test.go create mode 100644 admin/model/catalog_import.go create mode 100644 admin/repository/catalog_import.go create mode 100644 admin/repository/catalog_import_test.go diff --git a/admin/config.example.yaml b/admin/config.example.yaml index 4b18d71..f4f403b 100644 --- a/admin/config.example.yaml +++ b/admin/config.example.yaml @@ -34,6 +34,16 @@ database: # 运行机器上，不得提交、打包、截图或粘贴到工单和日志。 password: "你的 MySQL 密码" +integrations: + catalog: + # 第三方商品目录脚本的稳定来源名，会写入导入记录。不要放用户姓名或密码。 + source: external-catalog + + # 专用 Bearer Token，至少 32 个字符。线上优先使用 + # CMAUTOBUY_CATALOG_TOKEN；Token 不得写入日志、数据库、工单或提交。 + # 留空表示商品目录接口禁用，不影响 Admin 页面和 Client 四接口。 + token: "" + syb: # 顺运宝 ERP 地址。一般不用改，域名变了才改。 base_url: https://www.shunyunbaoerp.com diff --git a/admin/config/config.go b/admin/config/config.go index c3e25e2..bbec85e 100644 --- a/admin/config/config.go +++ b/admin/config/config.go @@ -31,6 +31,8 @@ const ( databasePasswordEnv = "CMAUTOBUY_DB_PASSWORD" databaseTLSModeEnv = "CMAUTOBUY_DB_TLS_MODE" databaseTLSCAEnv = "CMAUTOBUY_DB_TLS_CA" + catalogSourceEnv = "CMAUTOBUY_CATALOG_SOURCE" + catalogTokenEnv = "CMAUTOBUY_CATALOG_TOKEN" ) const ( @@ -52,6 +54,24 @@ type DatabaseConfig struct { TLSCA string `yaml:"tls_ca"` } +// CatalogIntegrationConfig 是第三方商品目录接口的独立身份配置。 +// Token 只允许来自未提交的 config.yaml 或环境变量，String 永远不输出明文。 +type CatalogIntegrationConfig struct { + Source string `yaml:"source"` + Token string `yaml:"token"` +} + +func (c CatalogIntegrationConfig) String() string { + token := "(未启用)" + if c.Token != "" { + token = "****" + } + return fmt.Sprintf("CatalogIntegrationConfig{Source:%s Token:%s}", c.Source, token) +} + +// Enabled 表示商品目录接口已经配置专用凭据。 +func (c CatalogIntegrationConfig) Enabled() bool { return c.Token != "" } + // String 永远隐藏密码，防止排错时用 %v 把凭据写进日志。 func (c DatabaseConfig) String() string { password := "(空)" @@ -98,6 +118,47 @@ func LoadDatabaseFromEnv() (DatabaseConfig, error) { return mergeDatabaseConfig(DatabaseConfig{}) } +// LoadCatalogIntegration 读取可选的商品目录接口配置。未配置 Token 时接口保持禁用， +// 不影响 Admin 其余页面和 Client 四接口启动。 +func LoadCatalogIntegration() (CatalogIntegrationConfig, error) { + path, err := ConfigPath() + if err != nil { + return CatalogIntegrationConfig{}, fmt.Errorf("无法确定 config.yaml 应该在的位置: %w", err) + } + var cfg CatalogIntegrationConfig + raw, readErr := os.ReadFile(path) + if readErr == nil { + parsed, parseErr := parseConfig(raw, path) + if parseErr != nil { + return CatalogIntegrationConfig{}, parseErr + } + cfg = parsed.Integrations.Catalog + } else if !os.IsNotExist(readErr) { + return CatalogIntegrationConfig{}, fmt.Errorf("读取配置文件 %s 失败: %w", path, readErr) + } + return mergeCatalogIntegration(cfg) +} + +func mergeCatalogIntegration(cfg CatalogIntegrationConfig) (CatalogIntegrationConfig, error) { + if value := strings.TrimSpace(os.Getenv(catalogSourceEnv)); value != "" { + cfg.Source = value + } + if value := os.Getenv(catalogTokenEnv); value != "" { + cfg.Token = value + } + cfg.Source = strings.TrimSpace(cfg.Source) + if cfg.Source == "" { + cfg.Source = "external" + } + if len(cfg.Source) > 64 { + return CatalogIntegrationConfig{}, fmt.Errorf("商品目录来源名称不能超过 64 个字符") + } + if cfg.Token != "" && len(cfg.Token) < 32 { + return CatalogIntegrationConfig{}, fmt.Errorf("%s 至少需要 32 个字符", catalogTokenEnv) + } + return cfg, nil +} + // mergeDatabaseConfig 把环境变量合并到文件配置上。环境变量只要非空， // 就覆盖对应 YAML 字段，避免线上误读部署目录里遗留的本地配置。 func mergeDatabaseConfig(cfg DatabaseConfig) (DatabaseConfig, error) { @@ -258,8 +319,14 @@ func (c SybConfig) String() string { // Config 是 config.yaml 的顶层结构。 type Config struct { - Database DatabaseConfig `yaml:"database"` - Syb SybConfig `yaml:"syb"` + Database DatabaseConfig `yaml:"database"` + Syb SybConfig `yaml:"syb"` + Integrations IntegrationsConfig `yaml:"integrations"` +} + +// IntegrationsConfig 收拢所有非 Client 的第三方数据接入配置。 +type IntegrationsConfig struct { + Catalog CatalogIntegrationConfig `yaml:"catalog"` } // configFileName 是 config.yaml 相对 exe（或 go run 时相对工作目录）的文件名。 diff --git a/admin/config/config_test.go b/admin/config/config_test.go index f3033de..be70a06 100644 --- a/admin/config/config_test.go +++ b/admin/config/config_test.go @@ -301,3 +301,29 @@ func TestLoad_配置文件损坏时报明确错误(t *testing.T) { t.Fatal("格式错误的 YAML 应该返回错误") } } + +func TestCatalogIntegrationConfig_环境覆盖且不泄露Token(t *testing.T) { + t.Setenv(catalogSourceEnv, "script-a") + t.Setenv(catalogTokenEnv, "0123456789abcdef0123456789abcdef") + cfg, err := mergeCatalogIntegration(CatalogIntegrationConfig{Source: "yaml", Token: "yaml-token-that-is-long-enough-123456"}) + if err != nil { + t.Fatal(err) + } + if cfg.Source != "script-a" || cfg.Token != "0123456789abcdef0123456789abcdef" { + t.Fatalf("环境变量未覆盖：%s", cfg) + } + if strings.Contains(cfg.String(), cfg.Token) || !strings.Contains(cfg.String(), "****") { + t.Fatalf("配置字符串泄露 Token：%s", cfg) + } +} + +func TestCatalogIntegrationConfig_弱Token拒绝且留空可禁用(t *testing.T) { + t.Setenv(catalogSourceEnv, "") + t.Setenv(catalogTokenEnv, "") + if cfg, err := mergeCatalogIntegration(CatalogIntegrationConfig{}); err != nil || cfg.Enabled() { + t.Fatalf("留空应安全禁用：cfg=%s err=%v", cfg, err) + } + if _, err := mergeCatalogIntegration(CatalogIntegrationConfig{Token: "too-short"}); err == nil || !strings.Contains(err.Error(), catalogTokenEnv) { + t.Fatalf("弱 Token 应被拒绝：%v", err) + } +} diff --git a/admin/handler/integration/auth.go b/admin/handler/integration/auth.go new file mode 100644 index 0000000..aa08b32 --- /dev/null +++ b/admin/handler/integration/auth.go @@ -0,0 +1,60 @@ +// Package integration 提供给受信任第三方脚本的 JSON 接口。 +// 它与网页登录和 /api/v1/client 完全隔离，不能复用 Cookie 或 X-Client-Id。 +package integration + +import ( + "crypto/subtle" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + + "cmautobuy/admin/config" +) + +const sourceContextKey = "catalog_integration_source" + +// RequireCatalogToken 校验商品目录接口的专用 Bearer Token。 +func RequireCatalogToken(cfg config.CatalogIntegrationConfig) gin.HandlerFunc { + return func(c *gin.Context) { + if !cfg.Enabled() { + integrationError(c, http.StatusServiceUnavailable, "INTEGRATION_DISABLED", + "商品目录接口尚未配置", true, nil) + c.Abort() + return + } + provided, ok := bearerToken(c.GetHeader("Authorization")) + if !ok || subtle.ConstantTimeCompare([]byte(provided), []byte(cfg.Token)) != 1 { + integrationError(c, http.StatusUnauthorized, "UNAUTHORIZED", + "接口凭据无效", false, nil) + c.Abort() + return + } + c.Set(sourceContextKey, cfg.Source) + c.Next() + } +} + +func bearerToken(header string) (string, bool) { + parts := strings.Fields(header) + if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") || parts[1] == "" { + return "", false + } + return parts[1], true +} + +func integrationSource(c *gin.Context) string { + value, _ := c.Get(sourceContextKey) + source, _ := value.(string) + return source +} + +func integrationError(c *gin.Context, status int, code, message string, retryable bool, details any) { + if details == nil { + details = gin.H{} + } + c.JSON(status, gin.H{"error": gin.H{ + "code": code, "message": message, "retryable": retryable, + "request_id": c.GetHeader("X-Request-Id"), "details": details, + }}) +} diff --git a/admin/handler/integration/auth_test.go b/admin/handler/integration/auth_test.go new file mode 100644 index 0000000..b31e95e --- /dev/null +++ b/admin/handler/integration/auth_test.go @@ -0,0 +1,47 @@ +package integration + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + + "cmautobuy/admin/config" +) + +func TestRequireCatalogToken(t *testing.T) { + gin.SetMode(gin.TestMode) + const token = "0123456789abcdef0123456789abcdef" + for _, tc := range []struct { + name, header string + cfg config.CatalogIntegrationConfig + want int + }{ + {"未启用", "", config.CatalogIntegrationConfig{}, http.StatusServiceUnavailable}, + {"缺少", "", config.CatalogIntegrationConfig{Source: "script", Token: token}, http.StatusUnauthorized}, + {"错误", "Bearer wrong", config.CatalogIntegrationConfig{Source: "script", Token: token}, http.StatusUnauthorized}, + {"正确", "Bearer " + token, config.CatalogIntegrationConfig{Source: "script", Token: token}, http.StatusNoContent}, + } { + t.Run(tc.name, func(t *testing.T) { + r := gin.New() + r.GET("/protected", RequireCatalogToken(tc.cfg), func(c *gin.Context) { + if integrationSource(c) != tc.cfg.Source { + t.Fatalf("source=%q", integrationSource(c)) + } + c.Status(http.StatusNoContent) + }) + req := httptest.NewRequest(http.MethodGet, "/protected", nil) + req.Header.Set("Authorization", tc.header) + resp := httptest.NewRecorder() + r.ServeHTTP(resp, req) + if resp.Code != tc.want { + t.Fatalf("status=%d body=%s", resp.Code, resp.Body.String()) + } + if strings.Contains(resp.Body.String(), token) || strings.Contains(resp.Body.String(), "wrong") { + t.Fatal("响应泄露接口凭据") + } + }) + } +} diff --git a/admin/model/catalog_import.go b/admin/model/catalog_import.go new file mode 100644 index 0000000..6ad784d --- /dev/null +++ b/admin/model/catalog_import.go @@ -0,0 +1,36 @@ +package model + +// CatalogImportStatus 是一次商品目录批次的稳定状态。 +type CatalogImportStatus string + +const ( + CatalogImportProcessing CatalogImportStatus = "processing" + CatalogImportSucceeded CatalogImportStatus = "succeeded" + CatalogImportFailed CatalogImportStatus = "failed" +) + +// CatalogImportRun 保存批次追踪信息，不保存 Token 或完整请求体。 +type CatalogImportRun struct { + Source string + BatchID string + RequestHash string + Status CatalogImportStatus + RequestCount int + ConflictCount int + ObservedAt string + LastRequestAt string + LastConflictAt string + ShopeeCreated int + ShopeeUpdated int + SKUCreated int + SKUUpdated int + PddCreated int + PddUpdated int + AssociationCreated int + AssociationUnchanged int + FailureCount int + ErrorSummary string + ResponseBody string + CreatedAt string + FinishedAt string +} diff --git a/admin/repository/catalog_import.go b/admin/repository/catalog_import.go new file mode 100644 index 0000000..f7f9c23 --- /dev/null +++ b/admin/repository/catalog_import.go @@ -0,0 +1,94 @@ +package repository + +import ( + "database/sql" + "errors" + "fmt" + "strings" + + "github.com/go-sql-driver/mysql" + + "cmautobuy/admin/model" +) + +var ( + // ErrCatalogImportRunExists 表示同一来源的批次号已经登记，由 service 判断是否重放。 + ErrCatalogImportRunExists = errors.New("商品目录批次已经存在") + ErrCatalogImportRunNotFound = errors.New("商品目录批次不存在") +) + +// 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, + last_request_at,created_at + ) VALUES (?,?,?,?,1,0,NULLIF(?,''),?,?)`, run.Source, run.BatchID, run.RequestHash, + run.Status, run.ObservedAt, 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,last_request_at,last_conflict_at,shopee_created,shopee_updated,sku_created, + sku_updated,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, + &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 +} + +// 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, "记录商品目录批次冲突") +} + +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 +} diff --git a/admin/repository/catalog_import_test.go b/admin/repository/catalog_import_test.go new file mode 100644 index 0000000..f729ccf --- /dev/null +++ b/admin/repository/catalog_import_test.go @@ -0,0 +1,52 @@ +package repository + +import ( + "database/sql" + "errors" + "testing" + + _ "modernc.org/sqlite" + + "cmautobuy/admin/model" +) + +func TestCatalogImportRun_登记查询重放和冲突(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatal(err) + } + 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, + 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, + 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, + PRIMARY KEY(source,batch_id))`); err != nil { + t.Fatal(err) + } + run := model.CatalogImportRun{Source: "script-a", BatchID: "batch-1", RequestHash: "abc", + Status: model.CatalogImportProcessing, ObservedAt: "2026-08-11T00:00:00Z", + LastRequestAt: "2026-08-11T00:01:00Z", CreatedAt: "2026-08-11T00:01:00Z"} + if err := InsertCatalogImportRun(db, run); err != nil { + t.Fatal(err) + } + if err := InsertCatalogImportRun(db, run); !errors.Is(err, ErrCatalogImportRunExists) { + t.Fatalf("重复批次应由唯一键拦截：%v", err) + } + if err := RecordCatalogImportReplay(db, run.Source, run.BatchID, "2026-08-11T00:02:00Z"); err != nil { + t.Fatal(err) + } + if err := RecordCatalogImportConflict(db, run.Source, run.BatchID, "2026-08-11T00:03:00Z"); err != nil { + t.Fatal(err) + } + got, err := GetCatalogImportRun(db, run.Source, run.BatchID) + if err != nil { + t.Fatal(err) + } + if got.RequestCount != 3 || got.ConflictCount != 1 || got.LastConflictAt == "" { + t.Fatalf("批次计数不正确：%+v", got) + } +} diff --git a/admin/repository/mysql_db.go b/admin/repository/mysql_db.go index 3619652..bd22cc0 100644 --- a/admin/repository/mysql_db.go +++ b/admin/repository/mysql_db.go @@ -19,7 +19,7 @@ import ( "cmautobuy/admin/spec" ) -const mysqlSchemaVersion = 6 +const mysqlSchemaVersion = 7 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) { @@ -495,10 +495,98 @@ func MigrateMySQL(db *sql.DB) error { if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 6, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v6 失败: %w", err) } + current = 6 + } + if current < 7 { + if err := migrateMySQLV7(db); err != nil { + return fmt.Errorf("执行 MySQL schema v7 失败: %w", err) + } + if err := checkMySQLV7Shape(db); err != nil { + return fmt.Errorf("MySQL schema v7 自检失败，未记录版本: %w", err) + } + 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) + } } return CheckMySQLSchema(db) } +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, + request_hash VARCHAR(64) COLLATE utf8mb4_bin NOT NULL, + status VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'processing', + request_count INT UNSIGNED NOT NULL DEFAULT 1, + conflict_count INT UNSIGNED NOT NULL DEFAULT 0, + observed_at VARCHAR(35) NULL, + last_request_at VARCHAR(35) NOT NULL, + last_conflict_at VARCHAR(35) NULL, + shopee_created INT UNSIGNED NOT NULL DEFAULT 0, + shopee_updated INT UNSIGNED NOT NULL DEFAULT 0, + sku_created INT UNSIGNED NOT NULL DEFAULT 0, + sku_updated INT UNSIGNED NOT NULL DEFAULT 0, + pdd_created INT UNSIGNED NOT NULL DEFAULT 0, + pdd_updated INT UNSIGNED NOT NULL DEFAULT 0, + association_created INT UNSIGNED NOT NULL DEFAULT 0, + association_unchanged INT UNSIGNED NOT NULL DEFAULT 0, + failure_count INT UNSIGNED NOT NULL DEFAULT 0, + error_summary VARCHAR(1000) NULL, + response_body LONGTEXT NULL, + created_at VARCHAR(35) NOT NULL, + finished_at VARCHAR(35) NULL, + PRIMARY KEY (source, batch_id), + KEY idx_catalog_runs_list (created_at DESC, source, batch_id), + KEY idx_catalog_runs_status (status, created_at DESC), + CONSTRAINT chk_catalog_runs_status CHECK (status IN ('processing','succeeded','failed')) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci` + +// migrateMySQLV7 建立第三方商品目录批次基础。所有 DDL 均逐项检查，支持隐式提交后重放。 +func migrateMySQLV7(db *sql.DB) error { + columns := []struct { + table, name, ddl string + }{ + {"shopee_products", "source_observed_at", `ALTER TABLE shopee_products ADD COLUMN source_observed_at VARCHAR(35) NULL AFTER source`}, + {"shopee_skus", "source_observed_at", `ALTER TABLE shopee_skus ADD COLUMN source_observed_at VARCHAR(35) NULL AFTER is_manual`}, + {"pdd_products", "source", `ALTER TABLE pdd_products ADD COLUMN source VARCHAR(16) COLLATE utf8mb4_bin NULL AFTER collect_status`}, + {"pdd_products", "source_observed_at", `ALTER TABLE pdd_products ADD COLUMN source_observed_at VARCHAR(35) NULL AFTER source`}, + } + 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) + } + } + } + + var clause string + err := db.QueryRow(`SELECT 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(&clause) + if err != nil && err != sql.ErrNoRows { + return fmt.Errorf("读取蝦皮来源约束失败: %w", err) + } + normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) + if !strings.Contains(normalized, "'api'") { + if err == nil { + if _, dropErr := db.Exec(`ALTER TABLE shopee_products DROP CHECK chk_shopee_products_source`); dropErr != nil { + return fmt.Errorf("移除旧蝦皮来源约束失败: %w", dropErr) + } + } + if _, addErr := db.Exec(`ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_source CHECK (source IN ('report','syb','api'))`); addErr != nil { + return fmt.Errorf("扩展蝦皮来源约束失败: %w", addErr) + } + } + if _, err := db.Exec(mysqlSchemaV7CatalogRuns); err != nil { + return fmt.Errorf("建立商品目录导入记录表失败: %w", err) + } + return nil +} + // migrateMySQLV6 给任务补充创建人。存量任务保持 NULL，明确标记为历史任务； // 新代码创建任务时写入当前网页登录用户。DDL 逐项检查，支持中断后重放。 func migrateMySQLV6(db *sql.DB) error { @@ -843,6 +931,7 @@ func CheckMySQLSchema(db *sql.DB) error { "tasks", "clients", "idempotency_keys", "task_claims", "syb_session", "syb_sync_state", "users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock", "spec_mapping_decisions", + "catalog_import_runs", } if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil { return err @@ -856,7 +945,71 @@ func CheckMySQLSchema(db *sql.DB) error { if err := checkMySQLV5Shape(db); err != nil { return err } - return checkMySQLV6Shape(db) + if err := checkMySQLV6Shape(db); err != nil { + return err + } + return checkMySQLV7Shape(db) +} + +func checkMySQLV7Shape(db *sql.DB) error { + for _, column := range []struct{ table, name string }{ + {"shopee_products", "source_observed_at"}, + {"shopee_skus", "source_observed_at"}, + {"pdd_products", "source_observed_at"}, + } { + if err := checkMySQLVarcharColumn(db, column.table, column.name, 35, true, "utf8mb4_0900_ai_ci", ""); err != nil { + return err + } + if err := checkMySQLNullDefault(db, column.table, column.name); err != nil { + return err + } + } + if err := checkMySQLVarcharColumn(db, "pdd_products", "source", 16, true, "utf8mb4_bin", ""); err != nil { + return err + } + if err := checkMySQLNullDefault(db, "pdd_products", "source"); err != nil { + return err + } + if err := checkMySQLSchema(db, []string{"catalog_import_runs"}); err != nil { + return err + } + for _, column := range []struct { + name string + length int64 + nullable bool + collation string + def string + }{ + {"source", 64, false, "utf8mb4_bin", ""}, + {"batch_id", 191, false, "utf8mb4_bin", ""}, + {"request_hash", 64, false, "utf8mb4_bin", ""}, + {"status", 16, false, "utf8mb4_bin", "processing"}, + } { + if err := checkMySQLVarcharColumn(db, "catalog_import_runs", column.name, column.length, column.nullable, column.collation, column.def); err != nil { + return err + } + } + var clause string + if err := db.QueryRow(`SELECT 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='catalog_import_runs' + AND tc.constraint_name='chk_catalog_runs_status' AND tc.constraint_type='CHECK'`).Scan(&clause); err != nil { + return fmt.Errorf("商品目录批次状态约束缺失: %w", err) + } + normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) + for _, value := range []string{"'processing'", "'succeeded'", "'failed'"} { + if !strings.Contains(normalized, value) { + return fmt.Errorf("商品目录批次状态约束不正确") + } + } + var sourceClause string + if err := db.QueryRow(`SELECT 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(&sourceClause); err != nil || !strings.Contains(strings.ToLower(sourceClause), "api") { + return fmt.Errorf("蝦皮商品来源约束未允许 api") + } + return nil } func checkMySQLV6Shape(db *sql.DB) error { diff --git a/admin/repository/mysql_db_integration_test.go b/admin/repository/mysql_db_integration_test.go index 149a212..4c172fc 100644 --- a/admin/repository/mysql_db_integration_test.go +++ b/admin/repository/mysql_db_integration_test.go @@ -340,6 +340,48 @@ func TestMySQLMigrate_V6形状错误不记版本(t *testing.T) { } } +func TestMySQLMigrate_V6升级V7且断点重跑(t *testing.T) { + db := openMySQLMigrationTestDB(t) + defer db.Close() + cleanMySQLTestSchema(t, db) + defer cleanMySQLTestSchema(t, db) + prepareMySQLV6(t, db) + + // 模拟 MySQL DDL 已经提交、版本号尚未记录的中断状态。 + if err := migrateMySQLV7(db); err != nil { + t.Fatal(err) + } + if err := MigrateMySQL(db); err != nil { + t.Fatal(err) + } + if err := MigrateMySQL(db); err != nil { + t.Fatalf("v7 重跑失败: %v", err) + } + var versions int + if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=7`).Scan(&versions); err != nil || versions != 1 { + t.Fatalf("v7=%d err=%v", versions, err) + } + mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) + VALUES('API-1','接口商品','api','2026-08-11T00:00:00Z','2026-08-11T00:00:00Z')`) +} + +func TestMySQLMigrate_V7形状错误不记版本(t *testing.T) { + db := openMySQLMigrationTestDB(t) + defer db.Close() + cleanMySQLTestSchema(t, db) + defer cleanMySQLTestSchema(t, db) + prepareMySQLV6(t, db) + mustExec(t, db, `ALTER TABLE shopee_products ADD COLUMN source_observed_at VARCHAR(10) NULL`) + if err := MigrateMySQL(db); err == nil { + t.Fatal("错误 source_observed_at 形状必须阻止 v7") + } + var count int + db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=7`).Scan(&count) + if count != 0 { + t.Fatal("v7 自检失败不得记录版本") + } +} + func openMySQLMigrationTestDB(t *testing.T) *sql.DB { t.Helper() if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { @@ -381,6 +423,19 @@ func prepareMySQLV4(t *testing.T, db *sql.DB) { mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z'),(4,'2026-08-10T00:00:00Z')`) } +func prepareMySQLV6(t *testing.T, db *sql.DB) { + t.Helper() + prepareMySQLV4(t, db) + if err := migrateMySQLV5(db); err != nil { + t.Fatal(err) + } + if err := migrateMySQLV6(db); err != nil { + t.Fatal(err) + } + mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) + VALUES (5,'2026-08-10T00:00:00Z'),(6,'2026-08-10T00:00:00Z')`) +} + func mustExec(t *testing.T, db *sql.DB, query string, args ...any) { t.Helper() if _, err := db.Exec(query, args...); err != nil { diff --git a/docs/admin/00-glossary.md b/docs/admin/00-glossary.md index de902d8..2e9a037 100644 --- a/docs/admin/00-glossary.md +++ b/docs/admin/00-glossary.md @@ -34,6 +34,8 @@ | excelize | Go 读写 Excel 的库 | 固定用它读蝦皮报表 | | CSRF | 攻击者诱导你在已登录状态下发出非本意的请求 | 所有写操作都要防，见 [06](06-quality-security.md) §4 | | 参数化查询 | SQL 里用 `?` 占位、值单独传，而不是拼字符串 | 防 SQL 注入的唯一正确做法 | +| 商品目录批次 | 第三方脚本一次提交的一组蝦皮、PDD 和关联数据 | 用 `source + batch_id` 做幂等；系统只保存处理摘要，不保存完整请求体 | +| Bearer Token | 第三方脚本放在 HTTP `Authorization` 头里的专用凭据 | 只用于商品目录接口，不复用网页登录 Cookie 或 Client 接口身份 | ## 3. 为什么导入必须是 upsert diff --git a/docs/admin/03-data-model.md b/docs/admin/03-data-model.md index 882311a..3165250 100644 --- a/docs/admin/03-data-model.md +++ b/docs/admin/03-data-model.md @@ -142,6 +142,11 @@ MySQL 迁移 v6（工单 #127）为 `tasks` 追加可空的 `created_by_user_id` 索引。存量记录不回填，NULL 明确表示“历史任务”；新建采集和采购任务必须写当前 网页登录用户。迁移与自检可重放，历史 SQLite migrations 保持冻结。 +MySQL 迁移 v7（工单 #132）新增 `catalog_import_runs` 批次摘要表，并为蝦皮商品、 +蝦皮 SKU 和 PDD 商品补充来源观测时间。`shopee_products.source` 新增 `api` 取值， +`pdd_products` 新增可空来源字段。迁移不改写既有业务数据，每条 DDL 可重放，所有 +字段和约束自检通过后才记录 v7。 + ## 3. 蝦皮数据 蝦皮报表**一个文件里混了两层数据**，所以拆成两张表。 @@ -154,7 +159,8 @@ CREATE TABLE shopee_products ( title TEXT NOT NULL, -- 蝦皮「商品名稱」 shopee_status TEXT, -- 蝦皮「商品當前狀態」 main_sku_code TEXT, -- 蝦皮「主商品貨號」 - source VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'report', -- report / syb + source VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'report', -- report / syb / api + source_observed_at VARCHAR(35), -- 外部来源观测时间，UTC ISO 8601 -- 下面两个是我们自己维护的，报表里没有，导入时绝不能覆盖 pdd_goods_url TEXT, -- ★ 人工填写的 PDD 链接原文 @@ -165,14 +171,15 @@ CREATE TABLE shopee_products ( ); ALTER TABLE shopee_products ADD CONSTRAINT chk_shopee_products_source - CHECK (source IN ('report', 'syb')); + CHECK (source IN ('report', 'syb', 'api')); CREATE INDEX idx_shopee_products_pdd ON shopee_products(pdd_goods_id); ``` `[必须]` **采集结果和采集状态不在这张表里**，它们属于 PDD 商品，见 §4。 -`source='syb'` 表示顺运宝先到、蝦皮报表尚未导入时创建的最小商品骨架。 +`source='syb'` 表示顺运宝先到、蝦皮报表尚未导入时创建的最小商品骨架； +`source='api'` 表示由商品目录批量接口写入。 后续 Excel 导入同一 `goods_id` 时必须补全商品信息并把来源提升为 `report`； 导入不得覆盖人工维护的 PDD 关联。 @@ -192,6 +199,7 @@ CREATE TABLE shopee_skus ( parse_ok INTEGER NOT NULL DEFAULT 0, -- 0=解析失败，界面上要标出来 sku_code TEXT, -- 蝦皮「商品選項貨號」 is_manual INTEGER NOT NULL DEFAULT 0, -- 1=人工新增的，导入不得删 + source_observed_at VARCHAR(35), -- 外部来源观测时间，UTC ISO 8601 created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (goods_id) REFERENCES shopee_products(goods_id) ON DELETE CASCADE @@ -875,3 +883,13 @@ CREATE UNIQUE INDEX idx_client_assignment_current - `client_id` 故意不设指向 `clients` 的外键：客户端清单允许删除后由同一稳定编号 重新登记，归属和审计历史不能随临时清单记录丢失。 - 归属记录不参与 Client API 的登记和领取判断，也不更新 `tasks.assigned_client`。 + +## 14. `catalog_import_runs` 商品目录批次摘要（MySQL v7） + +`catalog_import_runs` 以 `(source, batch_id)` 为主键，保存请求哈希、处理状态、 +新增/更新/冲突计数和错误摘要。它不保存 Bearer Token，也不保存完整请求体。 + +- 相同来源、批次号和请求哈希只处理一次；重复提交返回首次结果。 +- 相同来源和批次号携带不同请求哈希时记录冲突并返回 HTTP 409。 +- `response_body` 只保存可安全重放的结果摘要，不得包含凭据或原始商品数据。 +- `source_observed_at` 用于拒绝较旧数据覆盖较新数据；缺席记录绝不代表删除。 diff --git a/docs/admin/06-quality-security.md b/docs/admin/06-quality-security.md index 17cf41d..5f7660e 100644 --- a/docs/admin/06-quality-security.md +++ b/docs/admin/06-quality-security.md @@ -192,6 +192,10 @@ IP 主机名匹配；CA 轮换时必须同步更新客户端公开证书。 ## 7. 性能 +商品目录第三方接口使用独立 Bearer Token；未配置时接口必须返回 503 并保持禁用。 +Token 只能放在未提交的 `config.yaml` 或环境变量中，禁止写入数据库、日志、页面、 +工单和归档。鉴权失败只返回统一错误，不提示哪一部分凭据不正确。 + 规模很小（个位数操作员、集中部署），不要提前优化。但下面几条是基本功： - `[必须]` 搜索、排序、分页走**数据库查询**，不要一次查全量再在内存里过滤。