diff --git a/admin/handler/api/client_api.go b/admin/handler/api/client_api.go index 9d8e1e3..d7f4b38 100644 --- a/admin/handler/api/client_api.go +++ b/admin/handler/api/client_api.go @@ -123,7 +123,7 @@ func (h *Handler) Claim(c *gin.Context) { } // 2. 领取一个任务，只会拿到分配给这个客户端的 - task, err := service.ClaimNextTask(h.db, clientID, req.SupportedTypes) + task, err := service.ClaimNextTask(h.db, clientID, req.SupportedTypes, req.Capabilities.PurchaseMode) if err != nil { log.Printf("task_claim_failed client_id=%s err=%v", clientID, err) apiError(c, http.StatusInternalServerError, "TASK_CLAIM_FAILED", @@ -171,13 +171,14 @@ func taskPayload(t *model.Task) gin.H { } return gin.H{ - "id": t.TaskID, - "type": t.TaskType, - "version": t.Version, - "priority": t.Priority, - "payload": payload, - "created_at": t.CreatedAt, - "updated_at": t.UpdatedAt, + "id": t.TaskID, + "type": t.TaskType, + "execution_mode": t.ExecutionMode, + "version": t.Version, + "priority": t.Priority, + "payload": payload, + "created_at": t.CreatedAt, + "updated_at": t.UpdatedAt, } } diff --git a/admin/handler/api/client_api_test.go b/admin/handler/api/client_api_test.go new file mode 100644 index 0000000..b52781e --- /dev/null +++ b/admin/handler/api/client_api_test.go @@ -0,0 +1,14 @@ +package api + +import ( + "testing" + + "cmautobuy/admin/model" +) + +func TestTaskPayload_包含不可变执行模式(t *testing.T) { + payload := taskPayload(&model.Task{TaskID: "T-1", TaskType: model.TaskPurchase, ExecutionMode: model.TaskExecutionLive}) + if payload["execution_mode"] != model.TaskExecutionLive { + t.Fatalf("execution_mode=%v", payload["execution_mode"]) + } +} diff --git a/admin/handler/web/others.go b/admin/handler/web/others.go index 0d8f9ca..b80f259 100644 --- a/admin/handler/web/others.go +++ b/admin/handler/web/others.go @@ -630,7 +630,12 @@ func (h *Handler) SybCreateTask(c *gin.Context) { } requests = append(requests, service.PurchaseTaskRequest{SybID: sybID, MaxPriceCent: cent}) } - result, err := service.CreatePurchaseTasks(h.db, currentUser(c), requests, c.PostForm("client_id")) + result, err := service.CreatePurchaseTasksWithOptions(h.db, currentUser(c), requests, service.PurchaseTaskOptions{ + ClientID: c.PostForm("client_id"), + ExecutionMode: model.TaskExecutionMode(c.PostForm("execution_mode")), + LiveAcknowledged: c.PostForm("live_acknowledged") == "1", + LiveConfirmation: c.PostForm("live_confirmation"), + }) if err != nil { h.sybRedirect(c, "采购任务没有创建："+err.Error()) return diff --git a/admin/model/model.go b/admin/model/model.go index ca9d9c2..543bb3e 100644 --- a/admin/model/model.go +++ b/admin/model/model.go @@ -313,17 +313,27 @@ const ( TaskCancelled TaskStatus = "cancelled" // 已取消 ) +// TaskExecutionMode 区分只演练流程和会在拼多多创建未付款订单的真实流程。 +// 模式在任务创建时确定，后续取消、重派和结果回传都不得修改。 +type TaskExecutionMode string + +const ( + TaskExecutionDryRun TaskExecutionMode = "dry_run" + TaskExecutionLive TaskExecutionMode = "live" +) + // Task 是发给 Client 执行的一个任务。 // // PddGoodsURL 必填——Client 那边是 NOT NULL，空了它执行不了。 // 采购任务的 Quantity 和 MaxPriceCent 也必填，这是价格保护， // 见 docs/client/04-admin-api-contract.md §4。 type Task struct { - TaskID string - TaskType TaskType - Status TaskStatus - Version int - Priority int + TaskID string + TaskType TaskType + Status TaskStatus + ExecutionMode TaskExecutionMode + Version int + Priority int AssignedClient string ClaimedAt string @@ -344,6 +354,9 @@ type Task struct { ErrorMessage string FinishedAt string + LiveConfirmedBy string + LiveConfirmedAt string + CreatedAt string UpdatedAt string } diff --git a/admin/repository/client.go b/admin/repository/client.go index bafa683..09111ca 100644 --- a/admin/repository/client.go +++ b/admin/repository/client.go @@ -2,6 +2,7 @@ package repository import ( "database/sql" + "encoding/json" "errors" "fmt" "strings" @@ -9,6 +10,29 @@ import ( "cmautobuy/admin/model" ) +// ClientPurchaseMode 返回客户端最后一次登记时声明的采购能力。 +// 旧数据、空值或无法解析的 JSON 一律按 dry_run 处理，这是安全默认值。 +func ClientPurchaseMode(q Execer, clientID string) (string, error) { + var raw sql.NullString + if err := q.QueryRow(`SELECT capabilities FROM clients WHERE client_id = ?`, clientID).Scan(&raw); errors.Is(err, sql.ErrNoRows) { + return "", ErrClientNotFound + } else if err != nil { + return "", fmt.Errorf("读取客户端 %s 的采购能力失败: %w", clientID, err) + } + return ParseClientPurchaseMode(raw.String), nil +} + +// ParseClientPurchaseMode 把登记能力转成安全的固定值，供列表和创建校验共用。 +func ParseClientPurchaseMode(raw string) string { + var capabilities struct { + PurchaseMode string `json:"purchase_mode"` + } + if json.Unmarshal([]byte(raw), &capabilities) == nil && capabilities.PurchaseMode == string(model.TaskExecutionLive) { + return capabilities.PurchaseMode + } + return string(model.TaskExecutionDryRun) +} + var ( ErrClientNotFound = errors.New("客户端不存在") ErrPurchaserNotActive = errors.New("采购员不存在或已禁用") diff --git a/admin/repository/mysql_db.go b/admin/repository/mysql_db.go index b28b99d..e38dbcc 100644 --- a/admin/repository/mysql_db.go +++ b/admin/repository/mysql_db.go @@ -19,7 +19,7 @@ import ( "cmautobuy/admin/spec" ) -const mysqlSchemaVersion = 4 +const mysqlSchemaVersion = 5 // OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。 func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) { @@ -471,10 +471,77 @@ func MigrateMySQL(db *sql.DB) error { if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 4, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("记录 MySQL schema v4 失败: %w", err) } + current = 4 + } + if current < 5 { + if err := migrateMySQLV5(db); err != nil { + return fmt.Errorf("执行 MySQL schema v5 失败: %w", err) + } + if err := checkMySQLV5Shape(db); err != nil { + return fmt.Errorf("MySQL schema v5 自检失败，未记录版本: %w", err) + } + if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 5, time.Now().UTC().Format(time.RFC3339Nano)); err != nil { + return fmt.Errorf("记录 MySQL schema v5 失败: %w", err) + } } return CheckMySQLSchema(db) } +// migrateMySQLV5 给任务增加不可变执行模式和真实下单创建审计。 +// 每条 DDL 都先检查存在性，MySQL 在任意一步隐式提交后都可以安全重放。 +func migrateMySQLV5(db *sql.DB) error { + columns := []struct { + name string + ddl string + }{ + {"execution_mode", `ALTER TABLE tasks ADD COLUMN execution_mode VARCHAR(16) COLLATE utf8mb4_bin NOT NULL DEFAULT 'dry_run' AFTER status`}, + {"live_confirmed_by", `ALTER TABLE tasks ADD COLUMN live_confirmed_by VARCHAR(191) COLLATE utf8mb4_bin NULL AFTER finished_at`}, + {"live_confirmed_at", `ALTER TABLE tasks ADD COLUMN live_confirmed_at VARCHAR(35) NULL AFTER live_confirmed_by`}, + } + for _, column := range columns { + exists, err := mysqlColumnExists(db, "tasks", column.name) + if err != nil { + return err + } + if !exists { + if _, err := db.Exec(column.ddl); err != nil { + return fmt.Errorf("增加 tasks.%s 失败: %w", column.name, err) + } + } + } + if _, err := db.Exec(`UPDATE tasks SET execution_mode='dry_run' WHERE execution_mode IS NULL OR execution_mode=''`); err != nil { + return fmt.Errorf("回填任务执行模式失败: %w", err) + } + constraints := []struct { + name string + ddl string + }{ + {"chk_tasks_execution_mode", `ALTER TABLE tasks ADD CONSTRAINT chk_tasks_execution_mode CHECK (execution_mode IN ('dry_run','live'))`}, + {"chk_tasks_live_confirmation", `ALTER TABLE tasks ADD CONSTRAINT chk_tasks_live_confirmation CHECK ((execution_mode='dry_run' AND live_confirmed_by IS NULL AND live_confirmed_at IS NULL) OR (execution_mode='live' AND live_confirmed_by IS NOT NULL AND live_confirmed_by<>'' AND live_confirmed_at IS NOT NULL AND live_confirmed_at<>''))`}, + } + for _, constraint := range constraints { + exists, err := mysqlConstraintExists(db, "tasks", constraint.name) + if err != nil { + return err + } + if !exists { + if _, err := db.Exec(constraint.ddl); err != nil { + return fmt.Errorf("增加 tasks.%s 失败: %w", constraint.name, err) + } + } + } + indexExists, err := mysqlIndexExists(db, "tasks", "idx_tasks_claim_mode") + if err != nil { + return err + } + if !indexExists { + if _, err := db.Exec(`ALTER TABLE tasks ADD INDEX idx_tasks_claim_mode (assigned_client, status, execution_mode, priority DESC, created_at)`); err != nil { + return fmt.Errorf("增加任务模式领取索引失败: %w", err) + } + } + return nil +} + func checkMySQLSchemaV3(db *sql.DB) error { tables := []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, tables); err != nil { @@ -563,6 +630,15 @@ func mysqlConstraintExists(db *sql.DB, table, constraint string) (bool, error) { return count == 1, nil } +func mysqlIndexExists(db *sql.DB, table, index string) (bool, error) { + var count int + if err := db.QueryRow(`SELECT COUNT(DISTINCT index_name) FROM information_schema.statistics + WHERE table_schema=DATABASE() AND table_name=? AND index_name=?`, table, index).Scan(&count); err != nil { + return false, fmt.Errorf("检查 MySQL 索引 %s.%s 失败: %w", table, index, err) + } + return count == 1, nil +} + func backfillSybSpecKeys(db *sql.DB) error { const batchSize = 500 cursor := "" @@ -729,7 +805,65 @@ func CheckMySQLSchema(db *sql.DB) error { if err := checkMySQLV3Shape(db); err != nil { return err } - return checkMySQLV4Shape(db) + if err := checkMySQLV4Shape(db); err != nil { + return err + } + return checkMySQLV5Shape(db) +} + +func checkMySQLV5Shape(db *sql.DB) error { + if err := checkMySQLVarcharColumn(db, "tasks", "execution_mode", 16, false, "utf8mb4_bin", "dry_run"); err != nil { + return err + } + if err := checkMySQLVarcharColumn(db, "tasks", "live_confirmed_by", 191, true, "utf8mb4_bin", ""); err != nil { + return err + } + if err := checkMySQLNullDefault(db, "tasks", "live_confirmed_by"); err != nil { + return err + } + if err := checkMySQLVarcharColumn(db, "tasks", "live_confirmed_at", 35, true, "utf8mb4_0900_ai_ci", ""); err != nil { + return err + } + if err := checkMySQLNullDefault(db, "tasks", "live_confirmed_at"); err != nil { + return err + } + for _, check := range []struct { + name string + required []string + }{ + {"chk_tasks_execution_mode", []string{"execution_modein'dry_run','live'"}}, + {"chk_tasks_live_confirmation", []string{"execution_mode='dry_run'", "live_confirmed_byisnull", "live_confirmed_atisnull", "execution_mode='live'", "live_confirmed_byisnotnull", "live_confirmed_by<>''", "live_confirmed_atisnotnull", "live_confirmed_at<>''"}}, + } { + var enforced, clause 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='tasks' AND tc.constraint_name=? AND tc.constraint_type='CHECK'`, check.name).Scan(&enforced, &clause); err != nil { + return fmt.Errorf("任务 CHECK %s 缺失或不可读: %w", check.name, err) + } + normalized := strings.NewReplacer("`", "", " ", "", "(", "", ")", "", "_utf8mb4", "", `\`, "").Replace(strings.ToLower(clause)) + if enforced != "YES" { + return fmt.Errorf("任务 CHECK %s 未启用", check.name) + } + for _, fragment := range check.required { + if !strings.Contains(normalized, fragment) { + return fmt.Errorf("任务 CHECK %s 表达式不正确", check.name) + } + } + } + var cols string + if err := db.QueryRow(`SELECT GROUP_CONCAT(CONCAT(column_name,':',collation) ORDER BY seq_in_index) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='tasks' AND index_name='idx_tasks_claim_mode'`).Scan(&cols); err != nil || cols != "assigned_client:A,status:A,execution_mode:A,priority:D,created_at:A" { + return fmt.Errorf("任务模式领取索引不正确") + } + return nil +} + +func checkMySQLNullDefault(db *sql.DB, table, column string) error { + var defaultValue sql.NullString + if err := db.QueryRow(`SELECT column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, table, column).Scan(&defaultValue); err != nil { + return fmt.Errorf("检查 MySQL 列 %s.%s 默认值失败: %w", table, column, err) + } + if defaultValue.Valid { + return fmt.Errorf("MySQL 列 %s.%s 默认值应为 NULL", table, column) + } + return nil } func checkMySQLV4Shape(db *sql.DB) error { diff --git a/admin/repository/mysql_db_integration_test.go b/admin/repository/mysql_db_integration_test.go index 93aa5bc..7bce38b 100644 --- a/admin/repository/mysql_db_integration_test.go +++ b/admin/repository/mysql_db_integration_test.go @@ -232,6 +232,60 @@ func TestMySQLMigrate_V4形状错误不记版本(t *testing.T) { } } +func TestMySQLMigrate_V4升级V5且断点重跑(t *testing.T) { + db := openMySQLMigrationTestDB(t) + defer db.Close() + cleanMySQLTestSchema(t, db) + defer cleanMySQLTestSchema(t, db) + prepareMySQLV4(t, db) + // 模拟全部 DDL 已完成但版本尚未记录。 + if err := migrateMySQLV5(db); err != nil { + t.Fatal(err) + } + if err := MigrateMySQL(db); err != nil { + t.Fatal(err) + } + if err := MigrateMySQL(db); err != nil { + t.Fatalf("v5 重跑失败: %v", err) + } + var versions int + if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=5`).Scan(&versions); err != nil || versions != 1 { + t.Fatalf("v5=%d err=%v", versions, err) + } + now := "2026-08-10T00:00:00Z" + mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,created_at,updated_at) VALUES('OLD','collect','pending','https://example.invalid',?,?)`, now, now) + var mode string + if err := db.QueryRow(`SELECT execution_mode FROM tasks WHERE task_id='OLD'`).Scan(&mode); err != nil || mode != "dry_run" { + t.Fatalf("历史/缺省任务模式=%q err=%v", mode, err) + } + if _, err := db.Exec(`UPDATE tasks SET execution_mode='invalid' WHERE task_id='OLD'`); err == nil { + t.Fatal("执行模式 CHECK 必须拒绝非法值") + } + if _, err := db.Exec(`UPDATE tasks SET execution_mode='live' WHERE task_id='OLD'`); err == nil { + t.Fatal("缺少确认审计的 live 必须被拒绝") + } + if _, err := db.Exec(`UPDATE tasks SET live_confirmed_by='U',live_confirmed_at=? WHERE task_id='OLD'`, now); err == nil { + t.Fatal("dry_run 不得携带 live 确认审计") + } +} + +func TestMySQLMigrate_V5形状错误不记版本(t *testing.T) { + db := openMySQLMigrationTestDB(t) + defer db.Close() + cleanMySQLTestSchema(t, db) + defer cleanMySQLTestSchema(t, db) + prepareMySQLV4(t, db) + mustExec(t, db, `ALTER TABLE tasks ADD COLUMN execution_mode VARCHAR(8) NOT NULL DEFAULT 'dry_run'`) + if err := MigrateMySQL(db); err == nil { + t.Fatal("错误 execution_mode 形状必须阻止 v5") + } + var count int + db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=5`).Scan(&count) + if count != 0 { + t.Fatal("v5 自检失败不得记录版本") + } +} + func openMySQLMigrationTestDB(t *testing.T) *sql.DB { t.Helper() if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { @@ -263,6 +317,16 @@ func prepareMySQLV2(t *testing.T, db *sql.DB) { mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (1,'2026-08-10T00:00:00Z'),(2,'2026-08-10T00:00:00Z')`) } +func prepareMySQLV4(t *testing.T, db *sql.DB) { + t.Helper() + prepareMySQLV2(t, db) + if err := migrateMySQLV3(db); err != nil { + t.Fatal(err) + } + mustExec(t, db, mysqlSchemaV4Decisions) + mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z'),(4,'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/admin/repository/task.go b/admin/repository/task.go index 6486e45..d984ba9 100644 --- a/admin/repository/task.go +++ b/admin/repository/task.go @@ -30,10 +30,17 @@ import ( // // InnoDB 事务用 FOR UPDATE SKIP LOCKED 锁住一条候选任务。领取状态和 // task_claims 历史在同一个事务提交，避免只改了状态却没留下领取凭据。 -func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) { +func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string, purchaseModes ...string) (*model.Task, error) { if clientID == "" { return nil, fmt.Errorf("client_id 不能为空") } + purchaseMode := string(model.TaskExecutionDryRun) + if len(purchaseModes) > 0 { + purchaseMode = purchaseModes[0] + } + if purchaseMode != string(model.TaskExecutionDryRun) && purchaseMode != string(model.TaskExecutionLive) { + return nil, fmt.Errorf("purchase_mode 必须是 dry_run 或 live") + } ctx := context.Background() tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) if err != nil { @@ -45,6 +52,11 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model WHERE ( (assigned_client = ? AND status = 'assigned') OR (assigned_client IS NULL AND status = 'pending') )` args := []any{clientID} + // dry_run 客户端永远看不到真实采购任务。声明 live 的客户端仍可执行 + // 演练任务，避免它在没有真实任务时闲置。 + if purchaseMode != string(model.TaskExecutionLive) { + query += ` AND execution_mode = 'dry_run'` + } // 客户端只声明支持某些类型时，不要给它别的类型 if len(supportedTypes) > 0 { @@ -101,23 +113,26 @@ func GetTask(db *sql.DB, taskID string) (*model.Task, error) { var t model.Task var assigned, claimedAt, sybID, orderNo, goodsID, skuID sql.NullString var pddGoodsID, pddOptions, resultData, errCode, errMsg, finishedAt sql.NullString + var liveConfirmedBy, liveConfirmedAt sql.NullString var quantity, maxPrice sql.NullInt64 err := db.QueryRow(` - SELECT task_id, task_type, status, version, priority, + SELECT task_id, task_type, status, execution_mode, version, priority, assigned_client, claimed_at, syb_id, order_no, goods_id, shopee_sku_id, pdd_goods_url, pdd_goods_id, pdd_options, quantity, max_price_cent, result_data, error_code, error_message, finished_at, + live_confirmed_by, live_confirmed_at, created_at, updated_at FROM tasks WHERE task_id = ?`, taskID).Scan( - &t.TaskID, &t.TaskType, &t.Status, &t.Version, &t.Priority, + &t.TaskID, &t.TaskType, &t.Status, &t.ExecutionMode, &t.Version, &t.Priority, &assigned, &claimedAt, &sybID, &orderNo, &goodsID, &skuID, &t.PddGoodsURL, &pddGoodsID, &pddOptions, &quantity, &maxPrice, &resultData, &errCode, &errMsg, &finishedAt, + &liveConfirmedBy, &liveConfirmedAt, &t.CreatedAt, &t.UpdatedAt) if err == sql.ErrNoRows { return nil, nil @@ -140,6 +155,8 @@ func GetTask(db *sql.DB, taskID string) (*model.Task, error) { t.ErrorCode = errCode.String t.ErrorMessage = errMsg.String t.FinishedAt = finishedAt.String + t.LiveConfirmedBy = liveConfirmedBy.String + t.LiveConfirmedAt = liveConfirmedAt.String return &t, nil } @@ -269,6 +286,7 @@ type TaskListRow struct { TaskID string TaskType model.TaskType Status model.TaskStatus + ExecutionMode model.TaskExecutionMode AssignedClient string // 空表示无主任务 OrderNo string @@ -323,7 +341,7 @@ func taskFilterClause(filter TaskFilter) (string, []any) { func ListTasks(q Execer, filter TaskFilter, limit, offset int) ([]TaskListRow, error) { where, args := taskFilterClause(filter) sqlText := ` - SELECT t.task_id, t.task_type, t.status, t.assigned_client, + SELECT t.task_id, t.task_type, t.status, t.execution_mode, t.assigned_client, t.order_no, t.pdd_goods_id, t.pdd_options, t.quantity, t.max_price_cent, t.updated_at, p.title FROM tasks t @@ -345,7 +363,7 @@ func ListTasks(q Execer, filter TaskFilter, limit, offset int) ([]TaskListRow, e var quantity, maxPrice sql.NullInt64 if err := rows.Scan( - &r.TaskID, &r.TaskType, &r.Status, &assigned, + &r.TaskID, &r.TaskType, &r.Status, &r.ExecutionMode, &assigned, &orderNo, &pddGoodsID, &pddOptions, &quantity, &maxPrice, &r.UpdatedAt, &title, ); err != nil { @@ -465,17 +483,34 @@ func InsertPurchaseTask(q Execer, task model.Task) error { task.Quantity <= 0 || task.MaxPriceCent <= 0 { return fmt.Errorf("采购任务缺少客户端、商品、规格、数量或人民币价格上限") } + if task.ExecutionMode == "" { + task.ExecutionMode = model.TaskExecutionDryRun + } + if task.ExecutionMode != model.TaskExecutionDryRun && task.ExecutionMode != model.TaskExecutionLive { + return fmt.Errorf("采购任务执行模式无效") + } + if task.ExecutionMode == model.TaskExecutionLive && + (strings.TrimSpace(task.LiveConfirmedBy) == "" || strings.TrimSpace(task.LiveConfirmedAt) == "") { + return fmt.Errorf("真实采购任务缺少创建确认审计信息") + } + if task.ExecutionMode == model.TaskExecutionDryRun { + task.LiveConfirmedBy = "" + task.LiveConfirmedAt = "" + } now := model.NowISO() + liveConfirmedBy := sql.NullString{String: task.LiveConfirmedBy, Valid: task.LiveConfirmedBy != ""} + liveConfirmedAt := sql.NullString{String: task.LiveConfirmedAt, Valid: task.LiveConfirmedAt != ""} _, err := q.Exec(` INSERT INTO tasks - (task_id, task_type, status, assigned_client, + (task_id, task_type, status, execution_mode, assigned_client, syb_id, order_no, goods_id, shopee_sku_id, pdd_goods_url, pdd_goods_id, pdd_options, - quantity, max_price_cent, created_at, updated_at) - VALUES (?, 'purchase', 'assigned', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - task.TaskID, task.AssignedClient, task.SybID, task.OrderNo, + quantity, max_price_cent, live_confirmed_by, live_confirmed_at, + created_at, updated_at) + VALUES (?, 'purchase', 'assigned', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + task.TaskID, task.ExecutionMode, task.AssignedClient, task.SybID, task.OrderNo, task.GoodsID, task.ShopeeSKUID, task.PddGoodsURL, task.PddGoodsID, - task.PddOptions, task.Quantity, task.MaxPriceCent, now, now) + task.PddOptions, task.Quantity, task.MaxPriceCent, liveConfirmedBy, liveConfirmedAt, now, now) if err != nil { return fmt.Errorf("创建顺运宝明细 %s 的采购任务失败: %w", task.SybID, err) } diff --git a/admin/service/execution_mode_test.go b/admin/service/execution_mode_test.go new file mode 100644 index 0000000..085ca6b --- /dev/null +++ b/admin/service/execution_mode_test.go @@ -0,0 +1,49 @@ +package service + +import ( + "database/sql" + "testing" + + "cmautobuy/admin/model" +) + +func insertLiveTask(t *testing.T, db *sql.DB, taskID, clientID string) { + t.Helper() + now := model.NowISO() + _, err := db.Exec(`INSERT INTO tasks + (task_id,task_type,status,execution_mode,assigned_client,pdd_goods_url, + pdd_goods_id,pdd_options,quantity,max_price_cent,live_confirmed_by, + live_confirmed_at,created_at,updated_at) + VALUES (?,'purchase','assigned','live',?,'https://example.invalid/1','1', + '{"color":"黑色"}',1,100,'USER-1',?,?,?)`, taskID, clientID, now, now, now) + if err != nil { + t.Fatal(err) + } +} + +func TestClaimNextTask_按客户端真实采购能力隔离(t *testing.T) { + db := newTestDB(t) + insertLiveTask(t, db, "TASK-LIVE", "CLIENT-1") + if task, err := ClaimNextTask(db, "CLIENT-1", []string{"purchase"}, "dry_run"); err != nil || task != nil { + t.Fatalf("dry_run 客户端不应看见 live 任务: task=%+v err=%v", task, err) + } + task, err := ClaimNextTask(db, "CLIENT-1", []string{"purchase"}, "live") + if err != nil || task == nil || task.ExecutionMode != model.TaskExecutionLive { + t.Fatalf("live 客户端应领到 live 任务: task=%+v err=%v", task, err) + } +} + +func TestTaskExecutionMode_取消和重派不改变模式(t *testing.T) { + db := newTestDB(t) + insertLiveTask(t, db, "TASK-IMMUTABLE", "CLIENT-1") + if _, err := db.Exec(`UPDATE tasks SET status='cancelled',updated_at=? WHERE task_id='TASK-IMMUTABLE'`, model.NowISO()); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`UPDATE tasks SET status='assigned',assigned_client='CLIENT-2',updated_at=? WHERE task_id='TASK-IMMUTABLE'`, model.NowISO()); err != nil { + t.Fatal(err) + } + var mode string + if err := db.QueryRow(`SELECT execution_mode FROM tasks WHERE task_id='TASK-IMMUTABLE'`).Scan(&mode); err != nil || mode != "live" { + t.Fatalf("取消/重派后模式=%q err=%v", mode, err) + } +} diff --git a/admin/service/purchase_workflow.go b/admin/service/purchase_workflow.go index ba2a51e..68fd3c0 100644 --- a/admin/service/purchase_workflow.go +++ b/admin/service/purchase_workflow.go @@ -161,20 +161,42 @@ type PurchaseTaskRequest struct { MaxPriceCent int64 } +const LivePurchaseConfirmation = "创建未付款订单" + +// PurchaseTaskOptions 是一次批量创建共用的执行门禁。 +// ExecutionMode 留空表示 dry_run，确保旧调用和普通操作都保持安全默认值。 +type PurchaseTaskOptions struct { + ClientID string + ExecutionMode model.TaskExecutionMode + LiveAcknowledged bool + LiveConfirmation string +} + // PurchaseTaskResult 同时返回已创建数量和每条无法创建的原因。 type PurchaseTaskResult struct { Created int Failures []TaskCreateError } -// CreatePurchaseTasks 校验并创建采购任务。业务校验失败按明细返回，数据库错误整体回滚。 +// CreatePurchaseTasks 保留旧调用入口；未传执行模式时必须安全地创建演练任务。 func CreatePurchaseTasks(db *sql.DB, actor *model.User, requests []PurchaseTaskRequest, clientID string) (PurchaseTaskResult, error) { + return CreatePurchaseTasksWithOptions(db, actor, requests, PurchaseTaskOptions{ClientID: clientID}) +} + +// CreatePurchaseTasksWithOptions 校验并创建采购任务。业务校验失败按明细返回，数据库错误整体回滚。 +func CreatePurchaseTasksWithOptions(db *sql.DB, actor *model.User, requests []PurchaseTaskRequest, options PurchaseTaskOptions) (PurchaseTaskResult, error) { var result PurchaseTaskResult + if actor == nil { + return result, ErrUnauthenticated + } + if actor.Status != model.UserActive { + return result, fmt.Errorf("当前账号不是正常状态，不能创建采购任务") + } visibleUserID, err := visibleClientUserID(actor) if err != nil { return result, err } - clientID = strings.TrimSpace(clientID) + clientID := strings.TrimSpace(options.ClientID) if clientID == "" { return result, fmt.Errorf("请选择执行采购任务的客户端") } @@ -190,6 +212,27 @@ func CreatePurchaseTasks(db *sql.DB, actor *model.User, requests []PurchaseTaskR if !visible { return result, fmt.Errorf("所选客户端不存在或不在当前账号可见范围") } + executionMode := options.ExecutionMode + if executionMode == "" { + executionMode = model.TaskExecutionDryRun + } + if executionMode != model.TaskExecutionDryRun && executionMode != model.TaskExecutionLive { + return result, fmt.Errorf("执行模式无效，请选择采购演练或真实下单") + } + confirmedBy, confirmedAt := "", "" + if executionMode == model.TaskExecutionLive { + if !options.LiveAcknowledged || strings.TrimSpace(options.LiveConfirmation) != LivePurchaseConfirmation { + return result, fmt.Errorf("真实下单必须勾选风险确认并输入“%s”", LivePurchaseConfirmation) + } + purchaseMode, err := repository.ClientPurchaseMode(tx, clientID) + if err != nil { + return result, err + } + if purchaseMode != string(model.TaskExecutionLive) { + return result, fmt.Errorf("所选客户端未声明 live 能力，不能执行真实采购任务") + } + confirmedBy, confirmedAt = actor.UserID, model.NowISO() + } seen := map[string]bool{} for _, request := range requests { @@ -211,6 +254,7 @@ func CreatePurchaseTasks(db *sql.DB, actor *model.User, requests []PurchaseTaskR } if err := repository.InsertPurchaseTask(tx, model.Task{ TaskID: newPurchaseTaskID(), AssignedClient: clientID, + ExecutionMode: executionMode, LiveConfirmedBy: confirmedBy, LiveConfirmedAt: confirmedAt, SybID: context.Order.SybID, OrderNo: context.Order.OrderNo, GoodsID: context.Order.ShopeeGoodsID, PddGoodsURL: context.PddGoodsURL, PddGoodsID: context.PddGoodsID, diff --git a/admin/service/purchase_workflow_test.go b/admin/service/purchase_workflow_test.go index 030ef89..0fef5ff 100644 --- a/admin/service/purchase_workflow_test.go +++ b/admin/service/purchase_workflow_test.go @@ -3,6 +3,7 @@ package service import ( "database/sql" "errors" + "strings" "testing" "time" @@ -369,6 +370,62 @@ func TestCreatePurchaseTasks_部分业务失败不影响有效明细(t *testing. } } +func TestCreatePurchaseTasksWithOptions_正常采购员可创建真实任务(t *testing.T) { + db := newTestDB(t) + key := seedPurchasableWorkflow(t, db, "SYB-LIVE") + SaveSybMapping(db, "SYB-LIVE", key, "USR-1") + admin, buyer, _ := prepareClientAssignmentUsers(t, db) + if err := RegisterClient(db, model.Client{ClientID: "CLIENT-LIVE", Capabilities: `{"purchase_mode":"live"}`}, true); err != nil { + t.Fatal(err) + } + if _, _, err := AssignClient(db, admin, "CLIENT-LIVE", buyer.UserID, time.Now()); err != nil { + t.Fatal(err) + } + + result, err := CreatePurchaseTasksWithOptions(db, buyer, + []PurchaseTaskRequest{{SybID: "SYB-LIVE", MaxPriceCent: 4200}}, + PurchaseTaskOptions{ClientID: "CLIENT-LIVE", ExecutionMode: model.TaskExecutionLive, + LiveAcknowledged: true, LiveConfirmation: LivePurchaseConfirmation}) + if err != nil || result.Created != 1 { + t.Fatalf("正常采购员创建真实任务失败: result=%+v err=%v", result, err) + } + var mode, confirmedBy string + var confirmedAt sql.NullString + if err := db.QueryRow(`SELECT execution_mode,live_confirmed_by,live_confirmed_at FROM tasks WHERE syb_id='SYB-LIVE'`).Scan(&mode, &confirmedBy, &confirmedAt); err != nil { + t.Fatal(err) + } + if mode != "live" || confirmedBy != buyer.UserID || !confirmedAt.Valid { + t.Fatalf("真实任务审计不完整: mode=%s by=%s at=%+v", mode, confirmedBy, confirmedAt) + } +} + +func TestCreatePurchaseTasksWithOptions_真实模式安全门禁(t *testing.T) { + db := newTestDB(t) + key := seedPurchasableWorkflow(t, db, "SYB-GATE") + SaveSybMapping(db, "SYB-GATE", key, "USR-1") + admin, _, _ := prepareClientAssignmentUsers(t, db) + RegisterClient(db, model.Client{ClientID: "CLIENT-DRY", Capabilities: `{"purchase_mode":"dry_run"}`}, true) + request := []PurchaseTaskRequest{{SybID: "SYB-GATE", MaxPriceCent: 4200}} + + if _, err := CreatePurchaseTasksWithOptions(db, admin, request, PurchaseTaskOptions{ + ClientID: "CLIENT-DRY", ExecutionMode: model.TaskExecutionLive, + LiveAcknowledged: true, LiveConfirmation: LivePurchaseConfirmation, + }); err == nil || !strings.Contains(err.Error(), "未声明 live") { + t.Fatalf("dry_run 客户端必须被拒绝: %v", err) + } + if _, err := CreatePurchaseTasksWithOptions(db, admin, request, PurchaseTaskOptions{ + ClientID: "CLIENT-DRY", ExecutionMode: model.TaskExecutionLive, + LiveAcknowledged: true, LiveConfirmation: "错误短语", + }); err == nil || !strings.Contains(err.Error(), "必须勾选") { + t.Fatalf("错误确认短语必须被拒绝: %v", err) + } + disabled := *admin + disabled.Status = model.UserDisabled + if _, err := CreatePurchaseTasks(db, &disabled, request, "CLIENT-DRY"); err == nil || !strings.Contains(err.Error(), "不是正常状态") { + t.Fatalf("禁用账号必须被拒绝: %v", err) + } +} + func TestParsePriceYuanToCent_精确转换(t *testing.T) { for raw, want := range map[string]int64{"39.90": 3990, "1": 100, "0.01": 1, "12.3": 1230} { got, err := ParsePriceYuanToCent(raw) diff --git a/admin/service/service.go b/admin/service/service.go index 815cd06..5340b65 100644 --- a/admin/service/service.go +++ b/admin/service/service.go @@ -92,6 +92,7 @@ func TouchClient(db *sql.DB, clientID string) error { type ClientView struct { model.Client Status string + PurchaseMode string AssignedUserID string AssignedUsername string } @@ -136,6 +137,7 @@ func listClientViews(db *sql.DB, keyword, visibleUserID string, threshold time.D views = append(views, ClientView{ Client: c.Client, Status: c.StatusText(now, threshold), + PurchaseMode: repository.ParseClientPurchaseMode(c.Capabilities), AssignedUserID: c.AssignedUserID, AssignedUsername: c.AssignedUsername, }) @@ -185,7 +187,9 @@ func ListClientPageForUser(db *sql.DB, actor *model.User, keyword string, thresh views := make([]ClientView, 0, len(rows)) for _, client := range rows { views = append(views, ClientView{ - Client: client.Client, Status: client.StatusText(now, threshold), + Client: client.Client, + Status: client.StatusText(now, threshold), + PurchaseMode: repository.ParseClientPurchaseMode(client.Capabilities), AssignedUserID: client.AssignedUserID, AssignedUsername: client.AssignedUsername, }) } @@ -252,6 +256,6 @@ func DeleteClients(db *sql.DB, clientIDs []string) (int64, error) { // ClaimNextTask 为客户端领取一个任务，没有可领的返回 (nil, nil)。 // // 调用方拿到 nil 要返回 204 No Content，**不是 200 加空对象**。 -func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) { - return repository.ClaimNextTask(db, clientID, supportedTypes) +func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string, purchaseModes ...string) (*model.Task, error) { + return repository.ClaimNextTask(db, clientID, supportedTypes, purchaseModes...) } diff --git a/admin/service/task.go b/admin/service/task.go index a5a5006..9890fa2 100644 --- a/admin/service/task.go +++ b/admin/service/task.go @@ -58,6 +58,16 @@ func taskTypeText(t model.TaskType) string { return string(t) } +func taskExecutionModeText(taskType model.TaskType, mode model.TaskExecutionMode) string { + if taskType != model.TaskPurchase { + return "采集" + } + if mode == model.TaskExecutionLive { + return "真实下单（不支付）" + } + return "采购演练" +} + // ---------- 界面文字：状态 ---------- // taskStatusOrder 定下状态在筛选框和底部状态条里的显示顺序， @@ -222,13 +232,14 @@ func priceLimitText(cent int64) string { // TaskView 是列表页一行要显示的全部内容，全部已经是字符串。 type TaskView struct { - TaskID string - TypeText string - Target string - StatusText string - IsWarn bool // 失败 / 需人工，标黄提醒 - ClientText string - UpdatedAt string + TaskID string + TypeText string + ExecutionModeText string + Target string + StatusText string + IsWarn bool // 失败 / 需人工，标黄提醒 + ClientText string + UpdatedAt string } // TaskListResult 是列表页要的全部数据。 @@ -274,13 +285,14 @@ func ListTasksView(db *sql.DB, filter repository.TaskFilter, requestedPage int) } for _, r := range rows { result.Rows = append(result.Rows, TaskView{ - TaskID: r.TaskID, - TypeText: taskTypeText(r.TaskType), - Target: buildTarget(r), - StatusText: taskStatusText(r.Status), - IsWarn: isTaskWarn(r.Status), - ClientText: clientText(r.AssignedClient), - UpdatedAt: formatLocalTime(r.UpdatedAt), + TaskID: r.TaskID, + TypeText: taskTypeText(r.TaskType), + ExecutionModeText: taskExecutionModeText(r.TaskType, r.ExecutionMode), + Target: buildTarget(r), + StatusText: taskStatusText(r.Status), + IsWarn: isTaskWarn(r.Status), + ClientText: clientText(r.AssignedClient), + UpdatedAt: formatLocalTime(r.UpdatedAt), }) } return result, nil @@ -316,12 +328,13 @@ const resultDataLimit = 4000 // // `[必须]` 只读——本工单不做改派、重试、取消，弹窗里不应该有对应的表单。 type TaskDetailView struct { - TaskID string - TypeText string - StatusText string - ClientText string - ClaimedAt string - FinishedAt string + TaskID string + TypeText string + ExecutionModeText string + StatusText string + ClientText string + ClaimedAt string + FinishedAt string PddGoodsURL string @@ -350,14 +363,15 @@ func GetTaskDetail(db *sql.DB, taskID string) (*TaskDetailView, error) { } v := &TaskDetailView{ - TaskID: t.TaskID, - TypeText: taskTypeText(t.TaskType), - StatusText: taskStatusText(t.Status), - ClientText: clientText(t.AssignedClient), - ClaimedAt: formatLocalTime(t.ClaimedAt), - FinishedAt: formatLocalTime(t.FinishedAt), - PddGoodsURL: t.PddGoodsURL, - IsPurchase: t.TaskType == model.TaskPurchase, + TaskID: t.TaskID, + TypeText: taskTypeText(t.TaskType), + ExecutionModeText: taskExecutionModeText(t.TaskType, t.ExecutionMode), + StatusText: taskStatusText(t.Status), + ClientText: clientText(t.AssignedClient), + ClaimedAt: formatLocalTime(t.ClaimedAt), + FinishedAt: formatLocalTime(t.FinishedAt), + PddGoodsURL: t.PddGoodsURL, + IsPurchase: t.TaskType == model.TaskPurchase, } if v.IsPurchase { v.SpecText = specText(t.PddOptions) diff --git a/admin/static/css/app.css b/admin/static/css/app.css index da6c75f..42447c2 100644 --- a/admin/static/css/app.css +++ b/admin/static/css/app.css @@ -274,6 +274,27 @@ tr.empty small { color: #aaa; } border-radius: 6px; } .purchase-confirm-row label { display: grid; gap: 6px; } +.execution-mode-fieldset { + display: grid; + gap: 10px; + margin: 0; + padding: 12px; + border: 1px solid #d0d5dd; + border-radius: 6px; +} +.choice-row { display: flex; align-items: flex-start; gap: 8px; } +.choice-row span { display: grid; gap: 2px; } +.choice-row small { color: #667085; font-weight: 400; } +.live-confirm-panel { + display: grid; + gap: 10px; + padding: 12px; + border: 2px solid #b42318; + border-radius: 6px; + background: #fff5f3; +} +.live-confirm-panel[hidden] { display: none; } +.live-confirm-panel p { margin: 0; } @media (max-width: 760px) { .purchase-confirm-row { grid-template-columns: 1fr; } } diff --git a/admin/static/js/app.js b/admin/static/js/app.js index 7f05805..db3b116 100644 --- a/admin/static/js/app.js +++ b/admin/static/js/app.js @@ -140,6 +140,33 @@ var openButton = document.querySelector("[data-purchase-open]"); var form = document.querySelector("[data-purchase-form]"); if (!openButton || !form) return; + var modeInputs = form.querySelectorAll("[data-execution-mode]"); + var livePanel = form.querySelector("[data-live-confirm-panel]"); + var liveAcknowledged = form.querySelector("[data-live-acknowledged]"); + var liveConfirmation = form.querySelector("[data-live-confirmation]"); + var clientSelect = form.querySelector("select[name=client_id]"); + function updateExecutionMode() { + var selected = form.querySelector("[data-execution-mode]:checked"); + var isLive = selected && selected.value === "live"; + if (livePanel) livePanel.hidden = !isLive; + if (liveAcknowledged) liveAcknowledged.required = isLive; + if (liveConfirmation) liveConfirmation.required = isLive; + if (clientSelect) { + clientSelect.querySelectorAll("option[data-purchase-mode]").forEach(function (option) { + option.disabled = isLive && option.getAttribute("data-purchase-mode") !== "live"; + }); + if (clientSelect.selectedOptions.length && clientSelect.selectedOptions[0].disabled) { + clientSelect.value = ""; + } + } + form.setAttribute("data-confirm-submit", isLive + ? "确认创建真实的未付款订单？系统不会自动支付。" + : "确认创建采购演练任务？任务创建后将等待所选客户端领取。"); + } + modeInputs.forEach(function (input) { + input.addEventListener("change", updateExecutionMode); + }); + updateExecutionMode(); openButton.addEventListener("click", function () { var selected = {}; document.querySelectorAll("tbody input[name=ids]:checked").forEach(function (box) { @@ -156,6 +183,7 @@ }); var empty = form.querySelector("[data-purchase-empty]"); if (empty) empty.hidden = shown > 0; + updateExecutionMode(); }); } diff --git a/admin/templates/syb/list.html b/admin/templates/syb/list.html index 8ab798f..bf56af8 100644 --- a/admin/templates/syb/list.html +++ b/admin/templates/syb/list.html @@ -154,10 +154,32 @@ {{if not .AssignableClients}}当前账号没有可用客户端，请先让管理员完成客户端绑定。{{end}} +
+ 执行模式 + + +
+

价格上限单位是人民币元，默认取已映射 PDD 规格的采集价；不会使用顺运宝的台币售价。

{{range .Rows}}