diff --git a/admin/repository/mysql_db.go b/admin/repository/mysql_db.go
index df50e61..40826b1 100644
--- a/admin/repository/mysql_db.go
+++ b/admin/repository/mysql_db.go
@@ -19,7 +19,7 @@ import (
"cmautobuy/admin/spec"
)
-const mysqlSchemaVersion = 11
+const mysqlSchemaVersion = 12
// OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -555,10 +555,102 @@ func MigrateMySQL(db *sql.DB) error {
if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 11, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
return fmt.Errorf("记录 MySQL schema v11 失败: %w", err)
}
+ current = 11
+ }
+ if current < 12 {
+ if err := migrateMySQLV12(db); err != nil {
+ return fmt.Errorf("执行 MySQL schema v12 失败: %w", err)
+ }
+ if err := checkMySQLV12Shape(db); err != nil {
+ return fmt.Errorf("MySQL schema v12 自检失败,未记录版本: %w", err)
+ }
+ if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 12, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
+ return fmt.Errorf("记录 MySQL schema v12 失败: %w", err)
+ }
}
return CheckMySQLSchema(db)
}
+// migrateMySQLV12 保存采集任务的顺运宝来源,并回收没有有效任务的孤儿采集中状态。
+// CREATE TABLE IF NOT EXISTS 和带条件的 UPDATE 都可重放,适合 MySQL DDL 隐式提交后的重启恢复。
+func migrateMySQLV12(db *sql.DB) error {
+ if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS task_syb_sources (
+ task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+ syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+ created_at VARCHAR(35) NOT NULL,
+ PRIMARY KEY (task_id, syb_id),
+ KEY idx_task_syb_sources_syb (syb_id, task_id),
+ CONSTRAINT fk_task_syb_sources_task FOREIGN KEY (task_id)
+ REFERENCES tasks(task_id) ON DELETE CASCADE,
+ CONSTRAINT fk_task_syb_sources_syb FOREIGN KEY (syb_id)
+ REFERENCES syb_orders(syb_id) ON DELETE CASCADE
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`); err != nil {
+ return fmt.Errorf("建立采集任务顺运宝来源表失败: %w", err)
+ }
+ now := model.NowISO()
+ if _, err := db.Exec(`UPDATE pdd_products AS pp
+ SET collect_status='pending', collect_msg=NULL, updated_at=?
+ WHERE pp.collect_status='collecting' AND pp.deleted_at IS NULL
+ AND NOT EXISTS (
+ SELECT 1 FROM tasks t
+ WHERE t.task_type='collect' AND t.pdd_goods_id=pp.goods_id
+ AND t.status IN ('pending','assigned','claimed')
+ )`, now); err != nil {
+ return fmt.Errorf("回收无有效任务的 PDD 采集中状态失败: %w", err)
+ }
+ return nil
+}
+
+func checkMySQLV12Shape(db *sql.DB) error {
+ exists, err := mysqlTableExists(db, "task_syb_sources")
+ if err != nil || !exists {
+ return fmt.Errorf("采集任务顺运宝来源表缺失")
+ }
+ for _, column := range []struct {
+ name string
+ length int64
+ collation string
+ }{
+ {"task_id", 191, "utf8mb4_bin"},
+ {"syb_id", 191, "utf8mb4_bin"},
+ {"created_at", 35, "utf8mb4_0900_ai_ci"},
+ } {
+ if err := checkMySQLVarcharColumn(db, "task_syb_sources", column.name, column.length, false, column.collation, ""); err != nil {
+ return err
+ }
+ }
+ for _, index := range []struct{ name, columns string }{
+ {"PRIMARY", "task_id,syb_id"},
+ {"idx_task_syb_sources_syb", "syb_id,task_id"},
+ } {
+ var columns string
+ if err := db.QueryRow(`SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index)
+ FROM information_schema.statistics
+ WHERE table_schema=DATABASE() AND table_name='task_syb_sources' AND index_name=?`, index.name).Scan(&columns); err != nil || columns != index.columns {
+ return fmt.Errorf("采集任务顺运宝来源索引 %s 不正确", index.name)
+ }
+ }
+ for _, foreignKey := range []struct{ name, column, table, target string }{
+ {"fk_task_syb_sources_task", "task_id", "tasks", "task_id"},
+ {"fk_task_syb_sources_syb", "syb_id", "syb_orders", "syb_id"},
+ } {
+ var column, table, target string
+ err := db.QueryRow(`SELECT column_name,referenced_table_name,referenced_column_name
+ FROM information_schema.key_column_usage
+ WHERE constraint_schema=DATABASE() AND table_name='task_syb_sources' AND constraint_name=?`, foreignKey.name).
+ Scan(&column, &table, &target)
+ if err != nil || column != foreignKey.column || table != foreignKey.table || target != foreignKey.target {
+ return fmt.Errorf("采集任务顺运宝来源约束 %s 不正确", foreignKey.name)
+ }
+ var deleteRule string
+ if err := db.QueryRow(`SELECT delete_rule FROM information_schema.referential_constraints
+ WHERE constraint_schema=DATABASE() AND table_name='task_syb_sources' AND constraint_name=?`, foreignKey.name).Scan(&deleteRule); err != nil || deleteRule != "CASCADE" {
+ return fmt.Errorf("采集任务顺运宝来源约束 %s 删除规则不正确", foreignKey.name)
+ }
+ }
+ return nil
+}
+
// migrateMySQLV11 增加顺运宝店铺列,并从保留的原始 JSON 回填历史数据。
// DDL 和 UPDATE 都可重放:中断后再次启动只补缺列和仍为空的记录。
func migrateMySQLV11(db *sql.DB) error {
@@ -1167,6 +1259,7 @@ func CheckMySQLSchema(db *sql.DB) error {
"users", "web_sessions", "client_user_assignments", "syb_sync_runs", "admin_initialization_lock",
"spec_mapping_decisions",
"catalog_import_runs",
+ "task_syb_sources",
}
if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
return err
@@ -1189,7 +1282,16 @@ func CheckMySQLSchema(db *sql.DB) error {
if err := checkMySQLV8Shape(db); err != nil {
return err
}
- return checkMySQLV9Shape(db)
+ if err := checkMySQLV9Shape(db); err != nil {
+ return err
+ }
+ if err := checkMySQLV10Shape(db); err != nil {
+ return err
+ }
+ if err := checkMySQLV11Shape(db); err != nil {
+ return err
+ }
+ return checkMySQLV12Shape(db)
}
func checkMySQLV9Shape(db *sql.DB) error {
diff --git a/admin/repository/mysql_db_integration_test.go b/admin/repository/mysql_db_integration_test.go
index e30dcb2..e90cd1d 100644
--- a/admin/repository/mysql_db_integration_test.go
+++ b/admin/repository/mysql_db_integration_test.go
@@ -585,6 +585,71 @@ func TestMySQLMigrate_V11回填顺运宝店铺且可重放(t *testing.T) {
}
}
+func TestMySQLMigrate_V11升级V12并修复孤儿采集中状态(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)
+ }
+ // 模拟已在 v11 的生产库:去掉 v12 版本及其附加表,保留全部 v1-v11 结构。
+ mustExec(t, db, `DELETE FROM schema_migrations WHERE version=12`)
+ mustExec(t, db, `DROP TABLE task_syb_sources`)
+ now := model.NowISO()
+ mustExec(t, db, `INSERT INTO pdd_products(goods_id,url,collect_status,created_at,updated_at) VALUES
+ ('ORPHAN','https://mobile.yangkeduo.com/goods.html?goods_id=1','collecting',?,?),
+ ('ACTIVE','https://mobile.yangkeduo.com/goods.html?goods_id=2','collecting',?,?)`, now, now, now, now)
+ mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,pdd_goods_id,created_at,updated_at)
+ VALUES('COL-ACTIVE','collect','pending','https://mobile.yangkeduo.com/goods.html?goods_id=2','ACTIVE',?,?)`, now, now)
+
+ if err := MigrateMySQL(db); err != nil {
+ t.Fatalf("v11 升级 v12 失败: %v", err)
+ }
+ if err := MigrateMySQL(db); err != nil {
+ t.Fatalf("v12 重放失败: %v", err)
+ }
+ var orphanStatus, activeStatus string
+ if err := db.QueryRow(`SELECT collect_status FROM pdd_products WHERE goods_id='ORPHAN'`).Scan(&orphanStatus); err != nil {
+ t.Fatal(err)
+ }
+ if err := db.QueryRow(`SELECT collect_status FROM pdd_products WHERE goods_id='ACTIVE'`).Scan(&activeStatus); err != nil {
+ t.Fatal(err)
+ }
+ if orphanStatus != "pending" || activeStatus != "collecting" {
+ t.Fatalf("v12 状态修复错误:orphan=%s active=%s", orphanStatus, activeStatus)
+ }
+ var versionCount int
+ if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=12`).Scan(&versionCount); err != nil || versionCount != 1 {
+ t.Fatalf("v12 版本记录错误: count=%d err=%v", versionCount, err)
+ }
+}
+
+func TestMySQLMigrate_V12形状错误不记录版本(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)
+ }
+ mustExec(t, db, `DELETE FROM schema_migrations WHERE version=12`)
+ mustExec(t, db, `DROP TABLE task_syb_sources`)
+ mustExec(t, db, `CREATE TABLE task_syb_sources (
+ task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+ syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+ created_at VARCHAR(35) NOT NULL,
+ PRIMARY KEY(syb_id,task_id)
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`)
+ if err := MigrateMySQL(db); err == nil {
+ t.Fatal("错误主键和缺失外键必须阻止 v12")
+ }
+ var versionCount int
+ if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=12`).Scan(&versionCount); err != nil || versionCount != 0 {
+ t.Fatalf("v12 自检失败时不得记录版本:count=%d err=%v", versionCount, err)
+ }
+}
+
func openMySQLMigrationTestDB(t *testing.T) *sql.DB {
t.Helper()
if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" {
diff --git a/admin/repository/pdd.go b/admin/repository/pdd.go
index 2979d9b..9296d1e 100644
--- a/admin/repository/pdd.go
+++ b/admin/repository/pdd.go
@@ -387,7 +387,8 @@ func SetCollectFailed(q Execer, pddGoodsID, msg, artifactRef string) error {
//
// 允许发起采集的状态:
// - pending / failed —— 没有任务在跑;
-// - collecting **且已经超时**(`updated_at` 早于 `now - model.CollectStaleAfter`)
+// - collecting **且已经超时**(`updated_at` 早于 `now - model.CollectStaleAfter`),
+// 或数据库里已经没有有效采集任务;
// —— 客户端离线、崩溃或任务被删都会让一个 collecting 卡住不动,
// 这些是常态不是异常,界面上必须有出口,见 #24。
//
@@ -414,11 +415,17 @@ func MarkCollecting(q Execer, pddGoodsID string) (bool, error) {
staleBefore := now.Add(-model.CollectStaleAfter).UTC().Format(model.TimeLayout)
res, err := q.Exec(`
- UPDATE pdd_products
+ UPDATE pdd_products AS pp
SET collect_status = 'collecting', updated_at = ?
- WHERE goods_id = ? AND deleted_at IS NULL
+ WHERE pp.goods_id = ? AND deleted_at IS NULL
AND ( collect_status IN ('pending', 'failed')
- OR (collect_status = 'collecting' AND updated_at < ?) )`,
+ OR (collect_status = 'collecting' AND (
+ updated_at < ? OR NOT EXISTS (
+ SELECT 1 FROM tasks t
+ WHERE t.task_type='collect' AND t.pdd_goods_id=pp.goods_id
+ AND t.status IN ('pending','assigned','claimed')
+ )
+ )) )`,
nowISO, pddGoodsID, staleBefore)
if err != nil {
return false, fmt.Errorf("标记 PDD 商品 %s 采集中失败: %w", pddGoodsID, err)
@@ -442,11 +449,17 @@ func MarkRecollecting(q Execer, pddGoodsID string) (bool, error) {
staleBefore := now.Add(-model.CollectStaleAfter).UTC().Format(model.TimeLayout)
res, err := q.Exec(`
- UPDATE pdd_products
+ UPDATE pdd_products AS pp
SET collect_status = 'collecting', updated_at = ?
- WHERE goods_id = ? AND deleted_at IS NULL
+ WHERE pp.goods_id = ? AND deleted_at IS NULL
AND ( collect_status IN ('pending', 'failed', 'collected')
- OR (collect_status = 'collecting' AND updated_at < ?) )`,
+ OR (collect_status = 'collecting' AND (
+ updated_at < ? OR NOT EXISTS (
+ SELECT 1 FROM tasks t
+ WHERE t.task_type='collect' AND t.pdd_goods_id=pp.goods_id
+ AND t.status IN ('pending','assigned','claimed')
+ )
+ )) )`,
nowISO, pddGoodsID, staleBefore)
if err != nil {
return false, fmt.Errorf("标记 PDD 商品 %s 重新采集中失败: %w", pddGoodsID, err)
diff --git a/admin/repository/syb.go b/admin/repository/syb.go
index 62f6687..5a9f90f 100644
--- a/admin/repository/syb.go
+++ b/admin/repository/syb.go
@@ -396,17 +396,18 @@ const sybOrderContextFrom = `
// SybOrderContext 是顺运宝明细及其当前蝦皮/PDD 处理上下文。
// 处理阶段由 service 计算,Repository 只提供数据库事实。
type SybOrderContext struct {
- Order model.SybOrder
- ShopeeExists bool
- PddGoodsID string
- PddGoodsURL string
- PddCollectStatus string
- PddCollectMsg string
- PddSkusJSON string
- PddUpdatedAt string
- MappingOptionKey string
- MappingOptions string
- HasActiveTask bool
+ Order model.SybOrder
+ ShopeeExists bool
+ PddGoodsID string
+ PddGoodsURL string
+ PddCollectStatus string
+ PddCollectMsg string
+ PddSkusJSON string
+ PddUpdatedAt string
+ MappingOptionKey string
+ MappingOptions string
+ HasActiveTask bool
+ HasActiveCollectTask bool
}
func scanSybOrderContext(s rowScanner) (SybOrderContext, error) {
@@ -416,13 +417,13 @@ func scanSybOrderContext(s rowScanner) (SybOrderContext, error) {
var shopeeExists int
var pddGoodsID, pddGoodsURL, collectStatus, collectMsg, skusJSON, pddUpdatedAt sql.NullString
var mappingKey, mappingOptions sql.NullString
- var hasActiveTask int
+ var hasActiveTask, hasActiveCollectTask int
err := s.Scan(
&c.Order.SybID, &c.Order.OrderNo, &shopName, &title, &productSpec, &specKey, &shopeeGoodsID,
&c.Order.Quantity, &priceCent, &imageURL, &c.Order.SybData,
&c.Order.CreatedAt, &c.Order.UpdatedAt, &shopeeExists,
&pddGoodsID, &pddGoodsURL, &collectStatus, &collectMsg, &skusJSON, &pddUpdatedAt,
- &mappingKey, &mappingOptions, &hasActiveTask,
+ &mappingKey, &mappingOptions, &hasActiveTask, &hasActiveCollectTask,
)
c.Order.ShopName = shopName.String
c.Order.Title = title.String
@@ -441,6 +442,7 @@ func scanSybOrderContext(s rowScanner) (SybOrderContext, error) {
c.MappingOptionKey = mappingKey.String
c.MappingOptions = mappingOptions.String
c.HasActiveTask = hasActiveTask != 0
+ c.HasActiveCollectTask = hasActiveCollectTask != 0
return c, err
}
@@ -456,6 +458,9 @@ func ListSybOrderContexts(q Execer, filter SybOrderFilter, limit, offset int) ([
pp.skus_json, pp.updated_at, sm.pdd_option_key, sm.pdd_options,
EXISTS(SELECT 1 FROM tasks t WHERE t.task_type = 'purchase'
AND t.syb_id = so.syb_id
+ AND t.status IN ('pending', 'assigned', 'claimed')),
+ EXISTS(SELECT 1 FROM tasks t WHERE t.task_type = 'collect'
+ AND t.pdd_goods_id = pp.goods_id
AND t.status IN ('pending', 'assigned', 'claimed'))` +
sybOrderContextFrom + where + `
ORDER BY so.updated_at DESC, so.syb_id DESC`
@@ -490,6 +495,9 @@ func GetSybOrderContext(q Execer, sybID string) (*SybOrderContext, error) {
pp.skus_json, pp.updated_at, sm.pdd_option_key, sm.pdd_options,
EXISTS(SELECT 1 FROM tasks t WHERE t.task_type = 'purchase'
AND t.syb_id = so.syb_id
+ AND t.status IN ('pending', 'assigned', 'claimed')),
+ EXISTS(SELECT 1 FROM tasks t WHERE t.task_type = 'collect'
+ AND t.pdd_goods_id = pp.goods_id
AND t.status IN ('pending', 'assigned', 'claimed'))`+
sybOrderContextFrom+` WHERE so.syb_id = ?`, sybID))
if errors.Is(err, sql.ErrNoRows) {
diff --git a/admin/repository/task.go b/admin/repository/task.go
index ed41c12..78a05a7 100644
--- a/admin/repository/task.go
+++ b/admin/repository/task.go
@@ -379,8 +379,14 @@ func taskFilterClause(filter TaskFilter) (string, []any) {
if kw := strings.TrimSpace(filter.Keyword); kw != "" {
pattern := "%" + escapeLike(kw) + "%"
clauses = append(clauses,
- "(t.task_id LIKE ? ESCAPE '!' OR t.order_no LIKE ? ESCAPE '!' OR t.pdd_goods_id LIKE ? ESCAPE '!')")
- args = append(args, pattern, pattern, pattern)
+ `(t.task_id LIKE ? ESCAPE '!' OR t.order_no LIKE ? ESCAPE '!' OR t.pdd_goods_id LIKE ? ESCAPE '!'
+ OR EXISTS (
+ SELECT 1 FROM task_syb_sources tss
+ JOIN syb_orders so ON so.syb_id=tss.syb_id
+ WHERE tss.task_id=t.task_id
+ AND (tss.syb_id LIKE ? ESCAPE '!' OR so.order_no LIKE ? ESCAPE '!')
+ ))`)
+ args = append(args, pattern, pattern, pattern, pattern, pattern)
}
if filter.VisibleUserID != "" {
clauses = append(clauses, "t.created_by_user_id = ?")
@@ -529,6 +535,59 @@ func DeleteTasksInScope(q Execer, taskIDs []string, visibleUserID string) (int64
return res.RowsAffected()
}
+// ListCollectTaskGoodsIDsInScope 在删除前找出受影响的 PDD 商品。
+// 调用方必须和删除使用同一事务、同一可见范围,避免越权数据影响状态回收。
+func ListCollectTaskGoodsIDsInScope(q Execer, taskIDs []string, visibleUserID string) ([]string, error) {
+ if len(taskIDs) == 0 {
+ return nil, nil
+ }
+ placeholders := strings.TrimSuffix(strings.Repeat("?,", len(taskIDs)), ",")
+ args := make([]any, 0, len(taskIDs)+1)
+ for _, id := range taskIDs {
+ args = append(args, id)
+ }
+ where := `task_id IN (` + placeholders + `) AND task_type='collect' AND pdd_goods_id IS NOT NULL`
+ if visibleUserID != "" {
+ where += ` AND created_by_user_id = ?`
+ args = append(args, visibleUserID)
+ }
+ rows, err := q.Query(`SELECT DISTINCT pdd_goods_id FROM tasks WHERE `+where, args...)
+ if err != nil {
+ return nil, fmt.Errorf("查询待删除采集任务的 PDD 商品失败: %w", err)
+ }
+ defer rows.Close()
+ var goodsIDs []string
+ for rows.Next() {
+ var goodsID string
+ if err := rows.Scan(&goodsID); err != nil {
+ return nil, fmt.Errorf("读取待删除采集任务的 PDD 商品失败: %w", err)
+ }
+ goodsIDs = append(goodsIDs, goodsID)
+ }
+ return goodsIDs, rows.Err()
+}
+
+// ResetCollectingIfNoActiveTask 把没有有效采集任务的孤儿状态回收到待采集。
+// 条件判断和更新在一条 SQL 内完成,避免并发建任务时误覆盖 collecting。
+func ResetCollectingIfNoActiveTask(q Execer, goodsID string) (bool, error) {
+ result, err := q.Exec(`UPDATE pdd_products AS pp
+ SET collect_status='pending', collect_msg=NULL, updated_at=?
+ WHERE pp.goods_id=? AND pp.collect_status='collecting'
+ AND NOT EXISTS (
+ SELECT 1 FROM tasks t
+ WHERE t.task_type='collect' AND t.pdd_goods_id=pp.goods_id
+ AND t.status IN ('pending','assigned','claimed')
+ )`, model.NowISO(), goodsID)
+ if err != nil {
+ return false, fmt.Errorf("回收商品 %s 的孤立采集中状态失败: %w", goodsID, err)
+ }
+ n, err := result.RowsAffected()
+ if err != nil {
+ return false, fmt.Errorf("确认商品 %s 的采集状态回收结果失败: %w", goodsID, err)
+ }
+ return n == 1, nil
+}
+
// InsertCollectTask 建一条不指定客户端的采集任务。
// 保留这个入口给蝦皮、顺运宝等现有流程使用,避免它们被 PDD 页的新选项影响。
func InsertCollectTask(q Execer, taskID, goodsID, goodsURL string) error {
@@ -567,6 +626,31 @@ func InsertCollectTaskForClientAndUser(q Execer, taskID, goodsID, goodsURL, assi
return nil
}
+// InsertCollectTaskSybSources 保存一条采集任务对应的全部顺运宝明细来源。
+// 同一个 PDD 商品可能由多条明细共同发起,来源不能压成 tasks 上的单列。
+func InsertCollectTaskSybSources(q Execer, taskID string, sybIDs []string) error {
+ taskID = strings.TrimSpace(taskID)
+ if taskID == "" {
+ return fmt.Errorf("采集任务编号不能为空")
+ }
+ seen := make(map[string]struct{}, len(sybIDs))
+ for _, raw := range sybIDs {
+ sybID := strings.TrimSpace(raw)
+ if sybID == "" {
+ return fmt.Errorf("顺运宝明细编号不能为空")
+ }
+ if _, ok := seen[sybID]; ok {
+ continue
+ }
+ seen[sybID] = struct{}{}
+ if _, err := q.Exec(`INSERT INTO task_syb_sources(task_id,syb_id,created_at)
+ VALUES(?,?,?) ON DUPLICATE KEY UPDATE created_at=created_at`, taskID, sybID, model.NowISO()); err != nil {
+ return fmt.Errorf("保存采集任务 %s 的顺运宝来源 %s 失败: %w", taskID, sybID, err)
+ }
+ }
+ return nil
+}
+
// HasActivePurchaseTask 判断顺运宝明细是否已有尚未结束的采购任务。
func HasActivePurchaseTask(q Execer, sybID string) (bool, error) {
var count int
diff --git a/admin/service/pdd.go b/admin/service/pdd.go
index 38152a9..11ce6d0 100644
--- a/admin/service/pdd.go
+++ b/admin/service/pdd.go
@@ -576,7 +576,7 @@ func (r CollectTaskResult) Skipped() int {
// 返回的跳过分类必须显示给操作员。静默跳过的话,
// 操作员会以为任务建好了,等半天没动静也不知道为什么。
func CreatePddCollectTasks(db *sql.DB, goodsIDs []string) (CollectTaskResult, error) {
- return createPddCollectTasks(db, goodsIDs, "", "", "")
+ return createPddCollectTasks(db, goodsIDs, "", "", "", nil)
}
// CreatePddCollectTasksForUser 为 PDD 批量页面创建可选客户端的采集任务。
@@ -586,7 +586,7 @@ func CreatePddCollectTasksForUser(db *sql.DB, actor *model.User, goodsIDs []stri
if err != nil {
return CollectTaskResult{}, err
}
- return createPddCollectTasks(db, goodsIDs, strings.TrimSpace(clientID), visibleUserID, actor.UserID)
+ return createPddCollectTasks(db, goodsIDs, strings.TrimSpace(clientID), visibleUserID, actor.UserID, nil)
}
// RecollectPddProductForUser 为详情页一次明确的重新采集创建单商品任务。
@@ -658,7 +658,8 @@ func RecollectPddProductForUser(db *sql.DB, actor *model.User, goodsID, clientID
return result, nil
}
-func createPddCollectTasks(db *sql.DB, goodsIDs []string, clientID, visibleUserID, createdByUserID string) (CollectTaskResult, error) {
+func createPddCollectTasks(db *sql.DB, goodsIDs []string, clientID, visibleUserID, createdByUserID string,
+ sybSourcesByGoodsID map[string][]string) (CollectTaskResult, error) {
var result CollectTaskResult
goodsIDs = dedupe(goodsIDs)
@@ -725,8 +726,12 @@ func createPddCollectTasks(db *sql.DB, goodsIDs []string, clientID, visibleUserI
continue
}
+ taskID := newCollectTaskID()
if err := repository.InsertCollectTaskForClientAndUser(
- tx, newCollectTaskID(), p.GoodsID, p.URL, clientID, createdByUserID); err != nil {
+ tx, taskID, p.GoodsID, p.URL, clientID, createdByUserID); err != nil {
+ return CollectTaskResult{}, err
+ }
+ if err := repository.InsertCollectTaskSybSources(tx, taskID, sybSourcesByGoodsID[p.GoodsID]); err != nil {
return CollectTaskResult{}, err
}
result.Created++
diff --git a/admin/service/pdd_page_test.go b/admin/service/pdd_page_test.go
index 2ac5134..598a3f5 100644
--- a/admin/service/pdd_page_test.go
+++ b/admin/service/pdd_page_test.go
@@ -856,6 +856,23 @@ func TestCreatePddCollectTasks_采集中未超时的跳过并报数(t *testing.T
}
}
+func TestCreatePddCollectTasks_采集中但任务已不存在可立即重建(t *testing.T) {
+ db := newTestDB(t)
+ createProduct(t, db, "100000000003")
+ if _, err := db.Exec(`UPDATE pdd_products SET collect_status='collecting',updated_at=? WHERE goods_id=?`,
+ model.NowISO(), "100000000003"); err != nil {
+ t.Fatal(err)
+ }
+
+ result, err := CreatePddCollectTasks(db, []string{"100000000003"})
+ if err != nil {
+ t.Fatalf("孤儿采集中状态重建失败: %v", err)
+ }
+ if result.Created != 1 || result.SkippedCollecting != 0 {
+ t.Fatalf("没有有效任务时应立即重建,实际 %+v", result)
+ }
+}
+
// 采集失败的可以重新采集,不能一直卡在失败状态。
func TestCreatePddCollectTasks_失败的可以重新采集(t *testing.T) {
db := newTestDB(t)
diff --git a/admin/service/shopee_pdd.go b/admin/service/shopee_pdd.go
index 6f7669f..0f7dfa1 100644
--- a/admin/service/shopee_pdd.go
+++ b/admin/service/shopee_pdd.go
@@ -143,19 +143,16 @@ func CreateSybPddCollectTask(db *sql.DB, sybID string) (CollectTaskResult, error
if context == nil {
return CollectTaskResult{}, fmt.Errorf("顺运宝明细不存在")
}
- return CreateShopeePddCollectTask(db, context.Order.ShopeeGoodsID)
+ if context.PddGoodsID == "" {
+ return CollectTaskResult{}, fmt.Errorf("请先关联 PDD 商品")
+ }
+ return createPddCollectTasks(db, []string{context.PddGoodsID}, "", "", "",
+ map[string][]string{context.PddGoodsID: {context.Order.SybID}})
}
// CreateSybPddCollectTaskForUser 为网页登录入口记录任务创建人。
func CreateSybPddCollectTaskForUser(db *sql.DB, actor *model.User, sybID string) (CollectTaskResult, error) {
- context, err := repository.GetSybOrderContext(db, strings.TrimSpace(sybID))
- if err != nil {
- return CollectTaskResult{}, err
- }
- if context == nil {
- return CollectTaskResult{}, fmt.Errorf("顺运宝明细不存在")
- }
- return CreateShopeePddCollectTaskForUser(db, actor, context.Order.ShopeeGoodsID)
+ return CreateSybPddCollectTasksForUser(db, actor, []string{sybID}, "")
}
// CreateSybPddCollectTasksForUser 把勾选的顺运宝明细转换成 PDD 商品后批量建采集任务。
@@ -168,6 +165,7 @@ func CreateSybPddCollectTasksForUser(db *sql.DB, actor *model.User, sybIDs []str
}
goodsIDs := make([]string, 0, len(sybIDs))
+ sourcesByGoodsID := make(map[string][]string)
for _, sybID := range sybIDs {
context, err := repository.GetSybOrderContext(db, sybID)
if err != nil {
@@ -177,11 +175,16 @@ func CreateSybPddCollectTasksForUser(db *sql.DB, actor *model.User, sybIDs []str
return CollectTaskResult{}, invalidInput("顺运宝明细 %s 不存在,请刷新页面后重新选择", sybID)
}
stage, _, _, _ := sybStageFor(*context)
- if stage != SybStagePddPending && stage != SybStagePddFailed {
- return CollectTaskResult{}, invalidInput("顺运宝明细 %s 已不处于 PDD 待采集或采集失败状态,请刷新页面后重新选择", sybID)
+ if stage != SybStagePddPending && stage != SybStagePddFailed && stage != SybStagePddCollectingStale {
+ return CollectTaskResult{}, invalidInput("顺运宝明细 %s 已不处于 PDD 待采集、采集失败或采集超时状态,请刷新页面后重新选择", sybID)
}
goodsIDs = append(goodsIDs, context.PddGoodsID)
+ sourcesByGoodsID[context.PddGoodsID] = append(sourcesByGoodsID[context.PddGoodsID], sybID)
}
-
- return CreatePddCollectTasksForUser(db, actor, dedupe(goodsIDs), clientID)
+ visibleUserID, err := taskActorScope(actor)
+ if err != nil {
+ return CollectTaskResult{}, err
+ }
+ return createPddCollectTasks(db, dedupe(goodsIDs), strings.TrimSpace(clientID),
+ visibleUserID, actor.UserID, sourcesByGoodsID)
}
diff --git a/admin/service/shopee_pdd_test.go b/admin/service/shopee_pdd_test.go
index c736c00..ab85875 100644
--- a/admin/service/shopee_pdd_test.go
+++ b/admin/service/shopee_pdd_test.go
@@ -143,6 +143,42 @@ func TestCreateSybPddCollectTasksForUser_同一PDD商品去重(t *testing.T) {
if err := db.QueryRow(`SELECT COUNT(*) FROM tasks WHERE task_type='collect'`).Scan(&taskCount); err != nil || taskCount != 1 {
t.Fatalf("同一 PDD 商品应只建一条任务,count=%d err=%v", taskCount, err)
}
+ var sourceCount int
+ if err := db.QueryRow(`SELECT COUNT(*) FROM task_syb_sources`).Scan(&sourceCount); err != nil || sourceCount != 2 {
+ t.Fatalf("去重任务应保留两条顺运宝来源,count=%d err=%v", sourceCount, err)
+ }
+ for _, keyword := range []string{"ORDER-SYB-1", "ORDER-SYB-2", "SYB-1", "SYB-2"} {
+ list, err := ListTasksView(db, repository.TaskFilter{Keyword: keyword}, 1)
+ if err != nil || list.Total != 1 || len(list.Rows) != 1 {
+ t.Fatalf("按顺运宝来源 %q 搜索任务失败: total=%d rows=%d err=%v", keyword, list.Total, len(list.Rows), err)
+ }
+ }
+ if _, err := CreateSybPddCollectTasksForUser(db, admin, []string{"SYB-1"}, ""); !IsValidationError(err) {
+ t.Fatalf("未超时且有有效任务时应拒绝重复创建,实际 %v", err)
+ }
+}
+
+func TestCreateSybPddCollectTasksForUser_无任务的采集中状态可恢复(t *testing.T) {
+ db := newTestDB(t)
+ admin, _, _ := insertTaskUsers(t, db)
+ seedShopeeProduct(t, db, "1001", "商品一")
+ if _, err := AssociateShopeePdd(db, "1001", pddURLA, false); err != nil {
+ t.Fatal(err)
+ }
+ seedSybCollectOrder(t, db, "SYB-ORPHAN", "1001")
+ if _, err := db.Exec(`UPDATE pdd_products SET collect_status='collecting',updated_at=? WHERE goods_id=?`,
+ model.NowISO(), "737116531267"); err != nil {
+ t.Fatal(err)
+ }
+
+ result, err := CreateSybPddCollectTasksForUser(db, admin, []string{"SYB-ORPHAN"}, "")
+ if err != nil || result.Created != 1 {
+ t.Fatalf("孤儿采集中状态应能重建任务:result=%+v err=%v", result, err)
+ }
+ var sources int
+ if err := db.QueryRow(`SELECT COUNT(*) FROM task_syb_sources WHERE syb_id='SYB-ORPHAN'`).Scan(&sources); err != nil || sources != 1 {
+ t.Fatalf("重建任务应保存顺运宝来源:count=%d err=%v", sources, err)
+ }
}
func TestCreateSybPddCollectTasksForUser_不可采集明细整批拒绝(t *testing.T) {
diff --git a/admin/service/syb.go b/admin/service/syb.go
index b4ed88e..bdf6f5e 100644
--- a/admin/service/syb.go
+++ b/admin/service/syb.go
@@ -950,15 +950,16 @@ type SybOrderView struct {
}
const (
- SybStageSpecMissing = "spec_missing"
- SybStagePddMissing = "pdd_missing"
- SybStagePddPending = "pdd_pending"
- SybStagePddCollecting = "pdd_collecting"
- SybStagePddFailed = "pdd_failed"
- SybStageMappingPending = "mapping_pending"
- SybStagePurchaseReady = "purchase_ready"
- SybStageTaskCreated = "task_created"
- SybStagePurchaseBlocked = "purchase_blocked"
+ SybStageSpecMissing = "spec_missing"
+ SybStagePddMissing = "pdd_missing"
+ SybStagePddPending = "pdd_pending"
+ SybStagePddCollecting = "pdd_collecting"
+ SybStagePddCollectingStale = "pdd_collecting_stale"
+ SybStagePddFailed = "pdd_failed"
+ SybStageMappingPending = "mapping_pending"
+ SybStagePurchaseReady = "purchase_ready"
+ SybStageTaskCreated = "task_created"
+ SybStagePurchaseBlocked = "purchase_blocked"
)
// SybStageOption 是顺运宝处理阶段筛选项。
@@ -974,6 +975,7 @@ func SybStageOptions() []SybStageOption {
{Value: SybStagePddMissing, Text: "未关联 PDD"},
{Value: SybStagePddPending, Text: "PDD 待采集"},
{Value: SybStagePddCollecting, Text: "PDD 采集中"},
+ {Value: SybStagePddCollectingStale, Text: "PDD 采集中(超时)"},
{Value: SybStagePddFailed, Text: "PDD 采集失败"},
{Value: SybStageMappingPending, Text: "规格待匹配"},
{Value: SybStagePurchaseReady, Text: "可创建采购任务"},
@@ -1008,6 +1010,8 @@ func sybStageFor(c repository.SybOrderContext) (stage, text, help, action string
return SybStagePddMissing, "未关联 PDD", "下一步关联采购商品。", "关联 PDD"
case c.PddCollectStatus == string(model.CollectPending):
return SybStagePddPending, "PDD 待采集", "已关联 PDD 商品,尚未创建采集任务。", "创建采集任务"
+ case c.PddCollectStatus == string(model.CollectCollecting) && sybCollectingIsStale(c):
+ return SybStagePddCollectingStale, "PDD 采集中(超时)", "采集状态已超时或找不到有效任务,可以重新创建采集任务。", "重新创建采集任务"
case c.PddCollectStatus == string(model.CollectCollecting):
return SybStagePddCollecting, "PDD 采集中", "客户端正在采集规格和价格,请稍后刷新。", "查看采集状态"
case c.PddCollectStatus == string(model.CollectFailed):
@@ -1027,6 +1031,14 @@ func sybStageFor(c repository.SybOrderContext) (stage, text, help, action string
}
}
+func sybCollectingIsStale(c repository.SybOrderContext) bool {
+ if !c.HasActiveCollectTask {
+ return true
+ }
+ updatedAt, ok := model.ParseISO(c.PddUpdatedAt)
+ return !ok || time.Since(updatedAt) > model.CollectStaleAfter
+}
+
// SybListResult 是列表页要的全部数据。
type SybListResult struct {
Rows []SybOrderView
@@ -1045,7 +1057,8 @@ type SybListResult struct {
func ListSybOrdersView(db *sql.DB, keyword, shop, stage string, page int) (*SybListResult, error) {
shop = strings.TrimSpace(shop)
stage = ParseSybStage(stage)
- if stage == SybStageMappingPending || stage == SybStagePurchaseReady ||
+ if stage == SybStagePddCollecting || stage == SybStagePddCollectingStale ||
+ stage == SybStageMappingPending || stage == SybStagePurchaseReady ||
stage == SybStageTaskCreated || stage == SybStagePurchaseBlocked {
return listAdvancedSybStage(db, keyword, shop, stage, page)
}
@@ -1131,7 +1144,8 @@ func sybOrderViewFor(context repository.SybOrderContext) SybOrderView {
Quantity: o.Quantity, ImageURL: o.ImageURL,
UpdatedAt: formatLocalTime(o.UpdatedAt)}
v.Stage, v.StageText, v.StageHelp, v.ActionText = sybStageFor(context)
- v.CanCollect = v.Stage == SybStagePddPending || v.Stage == SybStagePddFailed
+ v.CanCollect = v.Stage == SybStagePddPending || v.Stage == SybStagePddFailed ||
+ v.Stage == SybStagePddCollectingStale
v.CanPurchase = v.Stage == SybStagePurchaseReady
switch {
case v.CanCollect:
@@ -1205,9 +1219,10 @@ func GetSybProcessingDetail(db *sql.DB, sybID string) (*SybProcessingDetail, err
PddGoodsID: context.PddGoodsID, PddURL: context.PddGoodsURL,
CollectStatus: context.PddCollectStatus, CollectMsg: context.PddCollectMsg,
}
+ d.Stage, d.StageText, d.StageHelp, _ = sybStageFor(*context)
d.CanCollect = context.PddGoodsID != "" &&
- (context.PddCollectStatus == string(model.CollectPending) ||
- context.PddCollectStatus == string(model.CollectFailed))
+ (d.Stage == SybStagePddPending || d.Stage == SybStagePddFailed ||
+ d.Stage == SybStagePddCollectingStale)
d.HasActiveTask = context.HasActiveTask
if context.PddCollectStatus == string(model.CollectCollected) && context.PddSkusJSON != "" {
choices, keys, names, parseErr := pddOptionChoices(context.PddSkusJSON)
@@ -1226,7 +1241,6 @@ func GetSybProcessingDetail(db *sql.DB, sybID string) (*SybProcessingDetail, err
}
}
d.ContextVersion = mappingContextVersion(*context)
- d.Stage, d.StageText, d.StageHelp, _ = sybStageFor(*context)
return d, nil
}
diff --git a/admin/service/syb_workflow_test.go b/admin/service/syb_workflow_test.go
index 6d013f1..199c7f1 100644
--- a/admin/service/syb_workflow_test.go
+++ b/admin/service/syb_workflow_test.go
@@ -113,7 +113,15 @@ func TestSybStageFor_全部筛选阶段与逐行推导一致(t *testing.T) {
{SybStageSpecMissing, "核对数据", copyWith(func(c *repository.SybOrderContext) { c.Order.SpecKey = "" })},
{SybStagePddMissing, "关联 PDD", copyWith(func(c *repository.SybOrderContext) { c.PddGoodsID, c.PddCollectStatus = "", "" })},
{SybStagePddPending, "创建采集任务", copyWith(func(c *repository.SybOrderContext) { c.PddCollectStatus = string(model.CollectPending) })},
- {SybStagePddCollecting, "查看采集状态", copyWith(func(c *repository.SybOrderContext) { c.PddCollectStatus = string(model.CollectCollecting) })},
+ {SybStagePddCollecting, "查看采集状态", copyWith(func(c *repository.SybOrderContext) {
+ c.PddCollectStatus = string(model.CollectCollecting)
+ c.PddUpdatedAt = model.NowISO()
+ c.HasActiveCollectTask = true
+ })},
+ {SybStagePddCollectingStale, "重新创建采集任务", copyWith(func(c *repository.SybOrderContext) {
+ c.PddCollectStatus = string(model.CollectCollecting)
+ c.PddUpdatedAt = model.NowISO()
+ })},
{SybStagePddFailed, "重新采集", copyWith(func(c *repository.SybOrderContext) { c.PddCollectStatus = string(model.CollectFailed) })},
{SybStageMappingPending, "匹配规格", copyWith(func(c *repository.SybOrderContext) { c.MappingOptionKey = "" })},
{SybStageTaskCreated, "查看采购任务", copyWith(func(c *repository.SybOrderContext) { c.HasActiveTask = true })},
@@ -150,6 +158,10 @@ func TestSybOrderViewFor_批量动作能力按阶段隔离(t *testing.T) {
context := base
context.PddCollectStatus = string(tc.status)
context.MappingOptionKey = tc.mappingKey
+ if tc.status == model.CollectCollecting {
+ context.PddUpdatedAt = model.NowISO()
+ context.HasActiveCollectTask = true
+ }
view := sybOrderViewFor(context)
if view.CanCollect != tc.wantCollect || view.CanPurchase != tc.wantPurchase {
t.Errorf("status=%s: CanCollect=%v CanPurchase=%v", tc.status, view.CanCollect, view.CanPurchase)
@@ -171,3 +183,14 @@ func TestSybOrderViewFor_批量动作能力按阶段隔离(t *testing.T) {
t.Fatalf("采购就绪行应只允许采购:CanCollect=%v CanPurchase=%v", view.CanCollect, view.CanPurchase)
}
}
+
+func TestSybCollectingIsStale_有效任务但时间损坏时允许恢复(t *testing.T) {
+ context := repository.SybOrderContext{
+ PddCollectStatus: string(model.CollectCollecting),
+ PddUpdatedAt: "not-a-time",
+ HasActiveCollectTask: true,
+ }
+ if !sybCollectingIsStale(context) {
+ t.Fatal("无法解析更新时间时不应永久停留在正常采集中")
+ }
+}
diff --git a/admin/service/task.go b/admin/service/task.go
index 1f147e6..34a07da 100644
--- a/admin/service/task.go
+++ b/admin/service/task.go
@@ -404,7 +404,7 @@ func (r *TaskListResult) StatusLine() string {
// DeleteTasks 批量删除任务,返回实际删掉的条数。
func DeleteTasks(db *sql.DB, taskIDs []string) (int64, error) {
- return repository.DeleteTasks(db, dedupe(taskIDs))
+ return deleteTasksInScope(db, dedupe(taskIDs), "", false)
}
// DeleteTasksForUser 保证整批删除要么全部在当前范围内成功,要么全部回滚。
@@ -413,7 +413,10 @@ func DeleteTasksForUser(db *sql.DB, actor *model.User, taskIDs []string) (int64,
if err != nil {
return 0, err
}
- ids := dedupe(taskIDs)
+ return deleteTasksInScope(db, dedupe(taskIDs), visibleUserID, true)
+}
+
+func deleteTasksInScope(db *sql.DB, ids []string, visibleUserID string, enforceExactScope bool) (int64, error) {
if len(ids) == 0 {
return 0, nil
}
@@ -422,13 +425,22 @@ func DeleteTasksForUser(db *sql.DB, actor *model.User, taskIDs []string) (int64,
return 0, fmt.Errorf("开始删除任务事务失败: %w", err)
}
defer tx.Rollback()
+ goodsIDs, err := repository.ListCollectTaskGoodsIDsInScope(tx, ids, visibleUserID)
+ if err != nil {
+ return 0, err
+ }
n, err := repository.DeleteTasksInScope(tx, ids, visibleUserID)
if err != nil {
return 0, err
}
- if n != int64(len(ids)) {
+ if enforceExactScope && n != int64(len(ids)) {
return 0, ErrTaskNotVisible
}
+ for _, goodsID := range goodsIDs {
+ if _, err := repository.ResetCollectingIfNoActiveTask(tx, goodsID); err != nil {
+ return 0, err
+ }
+ }
if err := tx.Commit(); err != nil {
return 0, fmt.Errorf("提交删除任务事务失败: %w", err)
}
diff --git a/admin/service/task_test.go b/admin/service/task_test.go
index 94667e4..ec4b970 100644
--- a/admin/service/task_test.go
+++ b/admin/service/task_test.go
@@ -594,6 +594,33 @@ func TestDeleteTasks_批量删除(t *testing.T) {
}
}
+func TestDeleteTasks_删除最后有效采集任务才回收商品状态(t *testing.T) {
+ db := newTestDB(t)
+ if _, err := repository.EnsurePddProduct(db, "111", "https://mobile.yangkeduo.com/goods.html?goods_id=111"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Exec(`UPDATE pdd_products SET collect_status='collecting' WHERE goods_id='111'`); err != nil {
+ t.Fatal(err)
+ }
+ insertTestTask(t, db, testTaskParams{taskID: "COL-A", taskType: model.TaskCollect, status: model.TaskPending, pddGoodsID: "111"})
+ insertTestTask(t, db, testTaskParams{taskID: "COL-B", taskType: model.TaskCollect, status: model.TaskAssigned, pddGoodsID: "111"})
+
+ if _, err := DeleteTasks(db, []string{"COL-A"}); err != nil {
+ t.Fatal(err)
+ }
+ product, err := repository.GetPddProductByGoodsID(db, "111")
+ if err != nil || product.CollectStatus != model.CollectCollecting {
+ t.Fatalf("仍有有效任务时不应回收:status=%s err=%v", product.CollectStatus, err)
+ }
+ if _, err := DeleteTasks(db, []string{"COL-B"}); err != nil {
+ t.Fatal(err)
+ }
+ product, err = repository.GetPddProductByGoodsID(db, "111")
+ if err != nil || product.CollectStatus != model.CollectPending {
+ t.Fatalf("删除最后有效任务后应回收为 pending:status=%s err=%v", product.CollectStatus, err)
+ }
+}
+
func TestTaskScope_采购员只能查看自己的任务且不能伪造筛选(t *testing.T) {
db := newTestDB(t)
admin, buyerA, buyerB := insertTaskUsers(t, db)
diff --git a/admin/templates/syb/detail_modal.html b/admin/templates/syb/detail_modal.html
index 40bf58d..ef516bd 100644
--- a/admin/templates/syb/detail_modal.html
+++ b/admin/templates/syb/detail_modal.html
@@ -47,7 +47,7 @@
{{if .CanCollect}}
-
+
{{else}}
{{if eq .CollectStatus "collecting"}}采集任务正在执行,请稍后刷新。{{else}}PDD 数据已采集,下一步匹配采购规格。{{end}}
{{end}}
diff --git a/docs/admin/01-requirements.md b/docs/admin/01-requirements.md
index 5622c37..434f763 100644
--- a/docs/admin/01-requirements.md
+++ b/docs/admin/01-requirements.md
@@ -266,7 +266,8 @@ PDD 商品之所以单独一个模块,是因为它在数据上就是**独立
- **类型**:全部 / 采集 / 采购。
- **状态**:全部 + 7 个状态,见 §6.2。
-- **关键词**:同时匹配任务编号、订单号、PDD 商品 ID。
+- **关键词**:同时匹配任务编号、采购订单号、PDD 商品 ID;从顺运宝创建的采集任务
+ 还要匹配其全部来源顺运宝订单号和明细 ID。
- **创建人**:仅管理员显示,可选全部账号(包含已禁用账号)或“历史任务”。
`[必须]` 采购员登录后只可查看自己创建的任务;URL 参数不能扩大范围。管理员查看
diff --git a/docs/admin/03-data-model.md b/docs/admin/03-data-model.md
index 8d9c80a..5f85ce8 100644
--- a/docs/admin/03-data-model.md
+++ b/docs/admin/03-data-model.md
@@ -108,6 +108,10 @@ SQLite 同一时刻只允许一个写事务,连接放太开会互相抢锁、
| v8(#59) | 顺运宝同步记录:新增 `syb_sync_runs` 和开始时间倒序索引;未改写 v1–v7。 |
| v9(#151) | `syb_orders` 增加可空 `shop_name`,从有效 `syb_data.stock.shopName` 回填历史店铺名;无效 JSON、缺字段和空白值保持 `NULL`。 |
+生产 MySQL 迁移继续独立追加;v12(#165)新增 `task_syb_sources`,并把没有任何
+有效采集任务的孤立 `pdd_products.collect_status='collecting'` 回收到 `pending`。
+历史 SQLite migrations 保持冻结,不追加该表。
+
**v3 为什么丢弃旧 `sku_mappings` 数据(见 #20):** 新主键需要 `pdd_option_key`,
这是 Go 的 `service.OptionKey()` 用 `json.Marshal` 算出来的规范化键,SQL 语句
复现不了。硬凑一个键出来有两种后果:算错了会让映射静默失效,需要人工重新匹配一遍;
@@ -290,8 +294,8 @@ CREATE INDEX idx_pdd_products_status ON pdd_products(collect_status);
`collect_status` 就永远卡在 `collecting`,界面上没有任何入口能救回来,
只能改数据库(见 #24)。
-修复:`collecting` 状态超过 `model.CollectStaleAfter`(15 分钟)**也允许**
-重新建采集任务:
+修复:`collecting` 状态超过 `model.CollectStaleAfter`(15 分钟),或已经找不到
+`pending/assigned/claimed` 采集任务时,**也允许**重新建采集任务:
```sql
-- repository.MarkCollecting,简化版
@@ -299,7 +303,8 @@ UPDATE pdd_products
SET collect_status = 'collecting', updated_at = ?
WHERE goods_id = ? AND deleted_at IS NULL
AND ( collect_status IN ('pending', 'failed')
- OR (collect_status = 'collecting' AND updated_at < ?) )
+ OR (collect_status = 'collecting' AND
+ (updated_at < ? OR NOT EXISTS (有效采集任务))) )
```
`[必须]` 15 分钟为什么是这个数:采集本身几十秒到两分钟,加上排队等客户端来领,
@@ -698,6 +703,29 @@ CREATE INDEX idx_tasks_order ON tasks(order_no);
状态含义见 [01 需求](01-requirements.md) §6.2。
+### 7.1 `task_syb_sources` 采集任务来源(生产 MySQL v12)
+
+从顺运宝创建采集任务时,同一个 PDD 商品可能对应多个货运单明细。来源使用关联表,
+不能压进 `tasks.syb_id` 单列;后者仍只表示采购任务自身的顺运宝明细。
+
+```sql
+CREATE TABLE task_syb_sources (
+ task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+ syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
+ created_at VARCHAR(35) NOT NULL,
+ PRIMARY KEY (task_id, syb_id),
+ KEY idx_task_syb_sources_syb (syb_id, task_id),
+ FOREIGN KEY (task_id) REFERENCES tasks(task_id) ON DELETE CASCADE,
+ FOREIGN KEY (syb_id) REFERENCES syb_orders(syb_id) ON DELETE CASCADE
+);
+```
+
+- `[必须]` 任务和全部来源在同一事务写入;按 PDD 商品去重后仍要保留每条来源。
+- `[必须]` 采集采购关键字搜索通过本表支持顺运宝订单号和明细 ID。
+- `[必须]` 删除任务后若同商品已无 `pending/assigned/claimed` 采集任务,必须在同一事务
+ 把仍为 `collecting` 的商品回收到 `pending`;有其他有效任务时不得回收。
+- 历史任务没有可靠来源,不做猜测回填,仍可按任务号和 PDD 商品 ID 查询。
+
## 8. `task_claims` 领取历史
记录"哪台客户端领过哪个任务"。
diff --git a/docs/admin/05-ui-specification.md b/docs/admin/05-ui-specification.md
index 89fa1e0..c9b68b2 100644
--- a/docs/admin/05-ui-specification.md
+++ b/docs/admin/05-ui-specification.md
@@ -526,8 +526,10 @@ HTML 片段,只替换弹窗内部的表格、总数和分页。主货运单表
- 图片显示小缩略图。点击缩略图在弹窗中按比例查看顺运宝原图,并提供新窗口打开入口;
图片加载失败时显示可读提示。`[必须]` 只存 URL,不代理、缓存或把图片塞进数据库。
- 处理阶段是按当前关联事实**算出来的**,不是存一份容易过期的状态字段。阶段至少区分:
- 采购数据异常(顺运宝未提供规格)、未关联 PDD、PDD 待采集、采集中、
+ 采购数据异常(顺运宝未提供规格)、未关联 PDD、PDD 待采集、采集中、采集中(超时)、
采集失败、规格待匹配、可采购和已创建任务。
+- `collecting` 已超过 15 分钟,或已找不到 `pending/assigned/claimed` 采集任务时,
+ 显示“PDD 采集中(超时)”和“重新创建采集任务”,不能继续伪装成正常进行中。
- 每行显示一个有文字的“下一步”按钮。商品标题是服务端渲染处理弹窗的明确入口,
支持鼠标和键盘;双击其他普通单元格不执行动作。图片入口只负责查看原图,不能打开处理详情。
- “下一步”随实时阶段变化:未关联 PDD、待采集、采集中、采集失败、规格待匹配、
@@ -536,7 +538,8 @@ HTML 片段,只替换弹窗内部的表格、总数和分页。主货运单表
- 规格原文为空时不显示匹配表单,明确提示“顺运宝未提供规格”,并禁止建立映射或创建采购任务。
- 完整货运单 JSON 不作为列显示,落在 `syb_data` 里,供后续排查用。
-复选框按批量动作能力启用,不再等同于“可采购”:PDD 待采集、PDD 采集失败行可用于
+复选框按批量动作能力启用,不再等同于“可采购”:PDD 待采集、PDD 采集失败、
+PDD 采集中(超时)行可用于
“创建 PDD 采集任务”,可创建采购任务行可用于“创建采购任务”,其他阶段继续禁用并通过
`title` 和无障碍标签说明原因。混合勾选时两个按钮分别统计和提交自身兼容的行,不能把
待采集商品带进采购确认,也不能把采购就绪商品带进采集请求。
@@ -639,7 +642,7 @@ ID 和在线状态;所有可见 Client 都可选择且只有一个时自动选
- **类型**:全部 / 采集 / 采购。
- **状态**:全部 + 7 个状态,见 [01 需求](01-requirements.md) §6.2。
-- **关键词**:同时匹配任务编号、订单号、PDD 商品 ID。
+- **关键词**:同时匹配任务编号、采购订单号、PDD 商品 ID、来源顺运宝订单号和明细 ID。
- **创建人**:仅管理员显示;包含全部账号、已禁用账号和“历史任务”。采购员不显示
此控件,服务端固定只查询本人创建的任务。