feat: 增加管理员 AI 模型配置 (#200)

2026-08-14 09:49:11 +08:00
parent ad44f83ea6
commit 018da3f732
24 changed files with 1546 additions and 8 deletions
@@ -0,0 +1,167 @@
+package repository
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+
+	"cmautobuy/admin/model"
+	"github.com/go-sql-driver/mysql"
+)
+
+var ErrAIProviderNameExists = errors.New("AI 服务商名称已经存在")
+
+func ListAIProviders(q Execer) ([]model.AIProviderConfig, error) {
+	rows, err := q.Query(`SELECT provider_id,name,base_url,model,timeout_seconds,max_concurrency,
+		confidence_threshold_bps,enabled,last_test_status,last_test_message,last_tested_at,
+		last_test_fingerprint,created_by_user_id,updated_by_user_id,created_at,updated_at
+		FROM ai_provider_configs ORDER BY enabled DESC,name,provider_id`)
+	if err != nil {
+		return nil, fmt.Errorf("查询 AI 服务商配置失败: %w", err)
+	}
+	defer rows.Close()
+	var result []model.AIProviderConfig
+	for rows.Next() {
+		item, err := scanAIProvider(rows.Scan)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, item)
+	}
+	return result, rows.Err()
+}
+
+func GetAIProvider(q Execer, providerID string) (*model.AIProviderConfig, error) {
+	item, err := scanAIProvider(func(dest ...any) error {
+		return q.QueryRow(`SELECT provider_id,name,base_url,model,timeout_seconds,max_concurrency,
+			confidence_threshold_bps,enabled,last_test_status,last_test_message,last_tested_at,
+			last_test_fingerprint,created_by_user_id,updated_by_user_id,created_at,updated_at
+			FROM ai_provider_configs WHERE provider_id=?`, providerID).Scan(dest...)
+	})
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("查询 AI 服务商配置失败: %w", err)
+	}
+	return &item, nil
+}
+
+type scanValues func(dest ...any) error
+
+func scanAIProvider(scan scanValues) (model.AIProviderConfig, error) {
+	var item model.AIProviderConfig
+	var enabled int
+	var lastMessage, lastTestedAt, fingerprint sql.NullString
+	err := scan(&item.ProviderID, &item.Name, &item.BaseURL, &item.Model, &item.TimeoutSeconds,
+		&item.MaxConcurrency, &item.ConfidenceThresholdBPS, &enabled, &item.LastTestStatus,
+		&lastMessage, &lastTestedAt, &fingerprint, &item.CreatedByUserID, &item.UpdatedByUserID,
+		&item.CreatedAt, &item.UpdatedAt)
+	item.Enabled = enabled == 1
+	item.LastTestMessage = lastMessage.String
+	item.LastTestedAt = lastTestedAt.String
+	item.LastTestFingerprint = fingerprint.String
+	return item, err
+}
+
+func InsertAIProvider(q Execer, item model.AIProviderConfig) error {
+	_, err := q.Exec(`INSERT INTO ai_provider_configs
+		(provider_id,name,base_url,model,timeout_seconds,max_concurrency,confidence_threshold_bps,
+		enabled,last_test_status,created_by_user_id,updated_by_user_id,created_at,updated_at)
+		VALUES(?,?,?,?,?,?,?,0,'pending',?,?,?,?,?)`, item.ProviderID, item.Name, item.BaseURL,
+		item.Model, item.TimeoutSeconds, item.MaxConcurrency, item.ConfidenceThresholdBPS,
+		item.CreatedByUserID, item.UpdatedByUserID, item.CreatedAt, item.UpdatedAt)
+	return aiProviderWriteError("新增 AI 服务商配置", err)
+}
+
+func UpdateAIProvider(q Execer, item model.AIProviderConfig) (bool, error) {
+	result, err := q.Exec(`UPDATE ai_provider_configs SET name=?,base_url=?,model=?,timeout_seconds=?,
+		max_concurrency=?,confidence_threshold_bps=?,last_test_status='pending',last_test_message=NULL,
+		last_tested_at=NULL,last_test_fingerprint=NULL,enabled=0,updated_by_user_id=?,updated_at=? WHERE provider_id=?`,
+		item.Name, item.BaseURL, item.Model, item.TimeoutSeconds, item.MaxConcurrency,
+		item.ConfidenceThresholdBPS, item.UpdatedByUserID, item.UpdatedAt, item.ProviderID)
+	if err != nil {
+		return false, aiProviderWriteError("更新 AI 服务商配置", err)
+	}
+	n, err := result.RowsAffected()
+	return n == 1, err
+}
+
+func RecordAIProviderTest(q Execer, providerID, status, message, testedAt, fingerprint, actorID string) (bool, error) {
+	result, err := q.Exec(`UPDATE ai_provider_configs SET last_test_status=?,last_test_message=?,
+		last_tested_at=?,last_test_fingerprint=?,updated_by_user_id=?,updated_at=? WHERE provider_id=?`,
+		status, nullableText(message), testedAt, nullableText(fingerprint), actorID, testedAt, providerID)
+	if err != nil {
+		return false, fmt.Errorf("记录 AI 连接测试失败: %w", err)
+	}
+	n, err := result.RowsAffected()
+	return n == 1, err
+}
+
+func InvalidateAIProviderTest(q Execer, providerID, actorID, updatedAt string) (bool, error) {
+	result, err := q.Exec(`UPDATE ai_provider_configs SET last_test_status='pending',last_test_message=NULL,
+		last_tested_at=NULL,last_test_fingerprint=NULL,enabled=0,updated_by_user_id=?,updated_at=? WHERE provider_id=?`,
+		actorID, updatedAt, providerID)
+	if err != nil {
+		return false, fmt.Errorf("重置 AI 连接测试状态失败: %w", err)
+	}
+	n, err := result.RowsAffected()
+	return n == 1, err
+}
+
+func LockAIProviders(q Execer) error {
+	rows, err := q.Query(`SELECT provider_id FROM ai_provider_configs FOR UPDATE`)
+	if err != nil {
+		return fmt.Errorf("锁定 AI 服务商配置失败: %w", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var providerID string
+		if err := rows.Scan(&providerID); err != nil {
+			return err
+		}
+	}
+	return rows.Err()
+}
+
+func EnableAIProvider(q Execer, providerID, actorID, updatedAt string) (bool, error) {
+	if _, err := q.Exec(`UPDATE ai_provider_configs SET enabled=0,updated_by_user_id=?,updated_at=? WHERE enabled=1`, actorID, updatedAt); err != nil {
+		return false, fmt.Errorf("停用旧 AI 服务商失败: %w", err)
+	}
+	result, err := q.Exec(`UPDATE ai_provider_configs SET enabled=1,updated_by_user_id=?,updated_at=? WHERE provider_id=?`, actorID, updatedAt, providerID)
+	if err != nil {
+		return false, fmt.Errorf("启用 AI 服务商失败: %w", err)
+	}
+	n, err := result.RowsAffected()
+	return n == 1, err
+}
+
+func DisableAIProvider(q Execer, providerID, actorID, updatedAt string) (bool, error) {
+	result, err := q.Exec(`UPDATE ai_provider_configs SET enabled=0,updated_by_user_id=?,updated_at=? WHERE provider_id=?`, actorID, updatedAt, providerID)
+	if err != nil {
+		return false, fmt.Errorf("停用 AI 服务商失败: %w", err)
+	}
+	n, err := result.RowsAffected()
+	return n == 1, err
+}
+
+func InsertAIProviderAudit(q Execer, audit model.AIProviderAudit) error {
+	_, err := q.Exec(`INSERT INTO ai_provider_audits(provider_id,action,details_json,actor_user_id,created_at)
+		VALUES(?,?,?,?,?)`, nullableText(audit.ProviderID), audit.Action, audit.DetailsJSON,
+		audit.ActorUserID, audit.CreatedAt)
+	if err != nil {
+		return fmt.Errorf("写入 AI 配置审计失败: %w", err)
+	}
+	return nil
+}
+
+func aiProviderWriteError(action string, err error) error {
+	if err == nil {
+		return nil
+	}
+	var mysqlErr *mysql.MySQLError
+	if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
+		return ErrAIProviderNameExists
+	}
+	return fmt.Errorf("%s失败: %w", action, err)
+}
@@ -20,7 +20,7 @@ import (
 	"cmautobuy/admin/spec"
 )

-const mysqlSchemaVersion = 19
+const mysqlSchemaVersion = 20

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

+// migrateMySQLV20 保存 AI 服务商的非敏感配置和配置审计。
+// API Key 只存在独立密钥文件中，不在这两张表里留列。
+func migrateMySQLV20(db *sql.DB) error {
+	statements := []string{
+		`CREATE TABLE IF NOT EXISTS ai_provider_configs (
+			provider_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY,
+			name VARCHAR(191) NOT NULL,
+			base_url VARCHAR(2048) NOT NULL,
+			model VARCHAR(191) NOT NULL,
+			timeout_seconds INT NOT NULL,
+			max_concurrency INT NOT NULL,
+			confidence_threshold_bps INT NOT NULL,
+			enabled TINYINT NOT NULL DEFAULT 0,
+			active_slot TINYINT GENERATED ALWAYS AS (IF(enabled=1,1,NULL)) STORED,
+			last_test_status VARCHAR(16) NOT NULL DEFAULT 'pending',
+			last_test_message VARCHAR(500),
+			last_tested_at VARCHAR(35),
+			last_test_fingerprint CHAR(64) COLLATE utf8mb4_bin,
+			created_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+			updated_by_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+			created_at VARCHAR(35) NOT NULL,
+			updated_at VARCHAR(35) NOT NULL,
+			UNIQUE KEY uq_ai_provider_name (name),
+			UNIQUE KEY uq_ai_provider_active (active_slot),
+			KEY idx_ai_provider_updated (updated_at DESC,provider_id),
+			CONSTRAINT fk_ai_provider_created_by FOREIGN KEY (created_by_user_id) REFERENCES users(user_id),
+			CONSTRAINT fk_ai_provider_updated_by FOREIGN KEY (updated_by_user_id) REFERENCES users(user_id),
+			CONSTRAINT chk_ai_provider_enabled CHECK (enabled IN (0,1)),
+			CONSTRAINT chk_ai_provider_timeout CHECK (timeout_seconds BETWEEN 1 AND 120),
+			CONSTRAINT chk_ai_provider_concurrency CHECK (max_concurrency BETWEEN 1 AND 16),
+			CONSTRAINT chk_ai_provider_confidence CHECK (confidence_threshold_bps BETWEEN 0 AND 10000),
+			CONSTRAINT chk_ai_provider_test_status CHECK (last_test_status IN ('pending','succeeded','failed'))
+		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
+		`CREATE TABLE IF NOT EXISTS ai_provider_audits (
+			id BIGINT AUTO_INCREMENT PRIMARY KEY,
+			provider_id VARCHAR(191) COLLATE utf8mb4_bin,
+			action VARCHAR(32) NOT NULL,
+			details_json LONGTEXT NOT NULL,
+			actor_user_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+			created_at VARCHAR(35) NOT NULL,
+			KEY idx_ai_provider_audit (provider_id,created_at DESC,id DESC),
+			CONSTRAINT fk_ai_audit_provider FOREIGN KEY (provider_id) REFERENCES ai_provider_configs(provider_id) ON DELETE SET NULL,
+			CONSTRAINT fk_ai_audit_actor FOREIGN KEY (actor_user_id) REFERENCES users(user_id)
+		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
+	}
+	for _, statement := range statements {
+		if _, err := db.Exec(statement); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // migrateMySQLV15 以低优先级补全历史 SYB 商品的店铺和图片。
 func migrateMySQLV15(db *sql.DB) error {
 	return backfillSybProductMetadata(db)
@@ -1798,6 +1863,7 @@ func CheckMySQLSchema(db *sql.DB) error {
 		"task_syb_sources",
 		"task_sequences",
 		"syb_allowed_shops",
+		"ai_provider_configs", "ai_provider_audits",
 	}
 	if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
 		return err
@@ -1850,7 +1916,42 @@ func CheckMySQLSchema(db *sql.DB) error {
 	if err := checkMySQLV18Shape(db); err != nil {
 		return err
 	}
-	return checkMySQLV19Shape(db)
+	if err := checkMySQLV19Shape(db); err != nil {
+		return err
+	}
+	return checkMySQLV20Shape(db)
+}
+
+func checkMySQLV20Shape(db *sql.DB) error {
+	if err := checkMySQLSchema(db, []string{"ai_provider_configs", "ai_provider_audits"}); err != nil {
+		return err
+	}
+	for _, name := range []string{"provider_id", "base_url", "model", "timeout_seconds", "max_concurrency",
+		"confidence_threshold_bps", "enabled", "active_slot", "last_test_status", "last_test_fingerprint"} {
+		exists, err := mysqlColumnExists(db, "ai_provider_configs", name)
+		if err != nil || !exists {
+			return fmt.Errorf("AI 服务商配置字段 %s 缺失: %v", name, err)
+		}
+	}
+	for _, item := range []struct{ kind, table, name string }{
+		{"index", "ai_provider_configs", "uq_ai_provider_name"},
+		{"index", "ai_provider_configs", "uq_ai_provider_active"},
+		{"constraint", "ai_provider_configs", "chk_ai_provider_test_status"},
+		{"constraint", "ai_provider_configs", "fk_ai_provider_created_by"},
+		{"constraint", "ai_provider_audits", "fk_ai_audit_actor"},
+	} {
+		var exists bool
+		var err error
+		if item.kind == "index" {
+			exists, err = mysqlIndexExists(db, item.table, item.name)
+		} else {
+			exists, err = mysqlConstraintExists(db, item.table, item.name)
+		}
+		if err != nil || !exists {
+			return fmt.Errorf("AI 配置%s %s.%s 缺失: %v", item.kind, item.table, item.name, err)
+		}
+	}
+	return nil
 }

 func checkMySQLV15Shape(db *sql.DB) error {
@@ -953,6 +953,38 @@ func openMySQLMigrationTestDB(t *testing.T) *sql.DB {
 	return db
 }

+func TestMySQLV20_AI配置首建重放与唯一启用(t *testing.T) {
+	db := openMySQLMigrationTestDB(t)
+	defer db.Close()
+	cleanMySQLTestSchema(t, db)
+	defer cleanMySQLTestSchema(t, db)
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := MigrateMySQL(db); err != nil {
+		t.Fatalf("v20 重放失败: %v", err)
+	}
+	if err := checkMySQLV20Shape(db); err != nil {
+		t.Fatal(err)
+	}
+	now := "2026-08-14T00:00:00Z"
+	mustExec(t, db, `INSERT INTO users(user_id,username,password_hash,role,status,password_changed_at,created_at,updated_at)
+		VALUES('AI-ADMIN','ai-admin','x','admin','active',?,?,?)`, now, now, now)
+	mustExec(t, db, `INSERT INTO ai_provider_configs(provider_id,name,base_url,model,timeout_seconds,
+		max_concurrency,confidence_threshold_bps,enabled,last_test_status,created_by_user_id,updated_by_user_id,created_at,updated_at)
+		VALUES('AI-1','一号','https://one.example/v1','m1',30,2,8500,1,'pending','AI-ADMIN','AI-ADMIN',?,?)`, now, now)
+	if _, err := db.Exec(`INSERT INTO ai_provider_configs(provider_id,name,base_url,model,timeout_seconds,
+		max_concurrency,confidence_threshold_bps,enabled,last_test_status,created_by_user_id,updated_by_user_id,created_at,updated_at)
+		VALUES('AI-2','二号','https://two.example/v1','m2',30,2,8500,1,'pending','AI-ADMIN','AI-ADMIN',?,?)`, now, now); err == nil {
+		t.Fatal("数据库必须拒绝同时启用两个 AI 服务商")
+	}
+	var secretColumns int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE()
+		AND table_name='ai_provider_configs' AND column_name IN ('api_key','secret','token')`).Scan(&secretColumns); err != nil || secretColumns != 0 {
+		t.Fatalf("AI 配置表不得含密钥列: count=%d err=%v", secretColumns, err)
+	}
+}
+
 func prepareMySQLV2(t *testing.T, db *sql.DB) {
 	t.Helper()
 	mustExec(t, db, `CREATE TABLE schema_migrations (version INT PRIMARY KEY, applied_at VARCHAR(35) NOT NULL) ENGINE=InnoDB`)