chengmaandClaude Opus 5
eb8d357f5a
打通 Client↔Admin 闭环的一环：录入 PDD 链接 → 创建采集任务 → 客户端领走去采 → 规格和价格显示在页面上。#16 写的 MarkCollecting 和 SoftDeletePddProduct 至此才有生产调用方。 链接解析严格、不做容错兜底：goods_id 上有 UNIQUE 约束，防重全靠它。 猜一个的话同一商品会存成好几行、采好几遍，规格映射还说不清指向哪一行。 短链一律拒绝，卡域名是为了拦"粘了淘宝链接"这种失误。 创建采集任务先占状态再建任务：MarkCollecting 只在 pending/failed 时成功， 同时充当"有没有人已经在采"的判断，与建任务在同一事务里， 所以并发点多次只会建出一个任务（实测 4 并发 → 1 条）。 submit.go 的 collectedData 增加 Dimensions —— 没有它就只能按 Go 的 map 遍历，而 map 无序，同一商品每次刷新"颜色/尺码"的先后都可能变。 顺带修 #16 一处缺陷：EnsurePddProduct 复活分支清空了 skus_json 却漏了 title，导致复活后状态显示"未采集"但标题还留着旧值。 审查打回一次：清理"PDD 链接唯一的录入口"这一过期说法（全库 5 处）， 以及 shopee.go 里"空 → no_link"的过期 TODO —— no_link 已在 #16 从 CHECK 约束删除，照写会直接撞约束。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
289 lines
9.9 KiB
Go
289 lines
9.9 KiB
Go
package repository
|
||
|
||
import (
|
||
"database/sql"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"cmautobuy/admin/model"
|
||
)
|
||
|
||
// claimCandidateLimit 是一次最多尝试抢多少条。
|
||
// 抢不到说明被别的客户端拿走了，再试下一条；都抢不到就当作没任务。
|
||
const claimCandidateLimit = 10
|
||
|
||
// ClaimNextTask 为指定客户端领取一个任务。
|
||
//
|
||
// 没有可领的任务时返回 (nil, nil) —— 调用方据此返回 204。
|
||
//
|
||
// # 两种任务都能领
|
||
//
|
||
// 指定给本机的 assigned_client = 我 且 status = 'assigned'
|
||
// 无主的 assigned_client 为空 且 status = 'pending'
|
||
//
|
||
// 采集任务创建时不指定客户端（浏览商品页没有副作用，哪台设备采都一样），
|
||
// 所以必须支持第二种，否则那些任务永远没人能领。
|
||
// 采购任务可以指定也可以留空——涉及钱和账号时应当显式分配。
|
||
//
|
||
// **指定给本机的优先。** 显式分配是人为决定，应当先兑现；
|
||
// 无主任务谁抢都一样，可以等。
|
||
//
|
||
// # 防并发
|
||
//
|
||
// 做法是**条件更新 + 检查影响行数**：先查出候选，再带着状态条件去更新，
|
||
// 影响行数为 0 就说明被别人抢先了，换下一条。
|
||
// 不用 SELECT ... FOR UPDATE，SQLite 没有那个。
|
||
func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) {
|
||
if clientID == "" {
|
||
return nil, fmt.Errorf("client_id 不能为空")
|
||
}
|
||
|
||
query := `SELECT task_id FROM tasks
|
||
WHERE ( (assigned_client = ? AND status = 'assigned')
|
||
OR (assigned_client IS NULL AND status = 'pending') )`
|
||
args := []any{clientID}
|
||
|
||
// 客户端只声明支持某些类型时，不要给它别的类型
|
||
if len(supportedTypes) > 0 {
|
||
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(supportedTypes)), ",")
|
||
query += ` AND task_type IN (` + placeholders + `)`
|
||
for _, t := range supportedTypes {
|
||
args = append(args, t)
|
||
}
|
||
}
|
||
// (assigned_client IS NULL) 为 0/1，0 排前面 —— 指定给本机的优先于无主的
|
||
query += ` ORDER BY (assigned_client IS NULL), priority DESC, created_at LIMIT ?`
|
||
args = append(args, claimCandidateLimit)
|
||
|
||
rows, err := db.Query(query, args...)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("查询可领任务失败: %w", err)
|
||
}
|
||
var candidates []string
|
||
for rows.Next() {
|
||
var id string
|
||
if err := rows.Scan(&id); err != nil {
|
||
rows.Close()
|
||
return nil, fmt.Errorf("读取候选任务失败: %w", err)
|
||
}
|
||
candidates = append(candidates, id)
|
||
}
|
||
rows.Close()
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
now := model.NowISO()
|
||
for _, taskID := range candidates {
|
||
// 两种情况合成一条语句：对"指定给我的"那种，写 assigned_client
|
||
// 是写同一个值，无副作用；对无主的，这一步就是"谁领到就标记谁"。
|
||
res, err := db.Exec(`
|
||
UPDATE tasks
|
||
SET status = 'claimed', assigned_client = ?,
|
||
claimed_at = ?, updated_at = ?
|
||
WHERE task_id = ?
|
||
AND ( (status = 'assigned' AND assigned_client = ?)
|
||
OR (status = 'pending' AND assigned_client IS NULL) )`,
|
||
clientID, now, now, taskID, clientID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("领取任务 %s 失败: %w", taskID, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if n == 0 {
|
||
continue // 被别的客户端抢先了，换下一条
|
||
}
|
||
// 记一笔领取历史。提交结果时要靠它判断这台客户端有没有领过——
|
||
// 任务重派后 assigned_client 会变，只看它就查不出来了。
|
||
if err := RecordClaim(db, taskID, clientID, now); err != nil {
|
||
return nil, err
|
||
}
|
||
return GetTask(db, taskID)
|
||
}
|
||
return nil, nil // 没有可领的任务
|
||
}
|
||
|
||
// GetTask 按编号读一条任务。
|
||
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 quantity, maxPrice sql.NullInt64
|
||
|
||
err := db.QueryRow(`
|
||
SELECT task_id, task_type, status, 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,
|
||
created_at, updated_at
|
||
FROM tasks WHERE task_id = ?`, taskID).Scan(
|
||
&t.TaskID, &t.TaskType, &t.Status, &t.Version, &t.Priority,
|
||
&assigned, &claimedAt,
|
||
&sybID, &orderNo, &goodsID, &skuID,
|
||
&t.PddGoodsURL, &pddGoodsID, &pddOptions,
|
||
&quantity, &maxPrice,
|
||
&resultData, &errCode, &errMsg, &finishedAt,
|
||
&t.CreatedAt, &t.UpdatedAt)
|
||
if err == sql.ErrNoRows {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, fmt.Errorf("读取任务 %s 失败: %w", taskID, err)
|
||
}
|
||
|
||
t.AssignedClient = assigned.String
|
||
t.ClaimedAt = claimedAt.String
|
||
t.SybID = sybID.String
|
||
t.OrderNo = orderNo.String
|
||
t.GoodsID = goodsID.String
|
||
t.ShopeeSKUID = skuID.String
|
||
t.PddGoodsID = pddGoodsID.String
|
||
t.PddOptions = pddOptions.String
|
||
t.Quantity = int(quantity.Int64)
|
||
t.MaxPriceCent = maxPrice.Int64
|
||
t.ResultData = resultData.String
|
||
t.ErrorCode = errCode.String
|
||
t.ErrorMessage = errMsg.String
|
||
t.FinishedAt = finishedAt.String
|
||
return &t, nil
|
||
}
|
||
|
||
// RecordClaim 记一笔"某客户端领过某任务"。
|
||
//
|
||
// 同一台客户端重复领同一个任务时只更新时间，不报错。
|
||
func RecordClaim(q Execer, taskID, clientID, claimedAt string) error {
|
||
_, err := q.Exec(`
|
||
INSERT INTO task_claims (task_id, client_id, claimed_at)
|
||
VALUES (?, ?, ?)
|
||
ON CONFLICT(task_id, client_id) DO UPDATE SET claimed_at = excluded.claimed_at`,
|
||
taskID, clientID, claimedAt)
|
||
if err != nil {
|
||
return fmt.Errorf("记录领取历史失败 task=%s client=%s: %w", taskID, clientID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// HasEverClaimed 判断这台客户端**曾经**领过这个任务。
|
||
//
|
||
// 用于提交结果时的权限判断：只有从没领过的才拒绝（403）。
|
||
// 任务已取消、已重派给别人，都**不影响**这个判断——
|
||
// 契约要求那些情况仍然要接受结果，见 docs/admin/04-client-api.md §4.1。
|
||
func HasEverClaimed(q Execer, taskID, clientID string) (bool, error) {
|
||
var one int
|
||
err := q.QueryRow(
|
||
`SELECT 1 FROM task_claims WHERE task_id = ? AND client_id = ?`,
|
||
taskID, clientID).Scan(&one)
|
||
if err == sql.ErrNoRows {
|
||
return false, nil
|
||
}
|
||
if err != nil {
|
||
return false, fmt.Errorf("查询领取历史失败: %w", err)
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// TaskInfo 是提交结果时需要知道的任务基本信息。
|
||
type TaskInfo struct {
|
||
TaskType model.TaskType
|
||
// GoodsID 是关联的**蝦皮**商品编号。
|
||
GoodsID string
|
||
// PddGoodsID 是要采集/购买的**拼多多**商品编号。
|
||
// 采集结果落到 pdd_products 时用的是它，不是 GoodsID —— 被采集的是 PDD 商品。
|
||
PddGoodsID string
|
||
}
|
||
|
||
// GetTaskInfo 查任务的类型和关联商品。任务不存在时返回 (nil, nil)。
|
||
func GetTaskInfo(q Execer, taskID string) (*TaskInfo, error) {
|
||
var info TaskInfo
|
||
var gid, pddGID sql.NullString
|
||
err := q.QueryRow(
|
||
`SELECT task_type, goods_id, pdd_goods_id FROM tasks WHERE task_id = ?`,
|
||
taskID).Scan(&info.TaskType, &gid, &pddGID)
|
||
if err == sql.ErrNoRows {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, fmt.Errorf("查询任务 %s 失败: %w", taskID, err)
|
||
}
|
||
info.GoodsID = gid.String
|
||
info.PddGoodsID = pddGID.String
|
||
return &info, nil
|
||
}
|
||
|
||
// MarkTaskSucceeded 记录成功结果。
|
||
//
|
||
// `[必须]` **不检查任务当前状态**。任务已取消、已重派给别的客户端，
|
||
// 都照样接受——客户端可能真的已经下单了，这些数据必须能留痕。
|
||
// 理由见 docs/admin/04-client-api.md §4.1。
|
||
func MarkTaskSucceeded(q Execer, taskID, resultData string) error {
|
||
now := model.NowISO()
|
||
_, err := q.Exec(`
|
||
UPDATE tasks
|
||
SET status = 'succeeded', result_data = ?, finished_at = ?, updated_at = ?
|
||
WHERE task_id = ?`,
|
||
resultData, now, now, taskID)
|
||
if err != nil {
|
||
return fmt.Errorf("标记任务 %s 成功失败: %w", taskID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// MarkTaskFailure 记录失败/需人工处理的结果。
|
||
//
|
||
// status 由客户端报告，映射规则见 docs/admin/04-client-api.md §5：
|
||
//
|
||
// retry_wait -> assigned（放回去等它再来领）
|
||
// manual_review -> manual_review
|
||
// failed -> failed
|
||
// cancelled -> cancelled
|
||
//
|
||
// 同样**不检查任务当前状态**，理由同 MarkTaskSucceeded。
|
||
func MarkTaskFailure(q Execer, taskID string, newStatus model.TaskStatus, errCode, errMsg string) error {
|
||
now := model.NowISO()
|
||
|
||
// 放回待领取的话不算结束，finished_at 保持为空
|
||
finishedAt := any(now)
|
||
if newStatus == model.TaskAssigned {
|
||
finishedAt = nil
|
||
}
|
||
|
||
_, err := q.Exec(`
|
||
UPDATE tasks
|
||
SET status = ?, error_code = ?, error_message = ?,
|
||
finished_at = ?, updated_at = ?
|
||
WHERE task_id = ?`,
|
||
newStatus, errCode, errMsg, finishedAt, now, taskID)
|
||
if err != nil {
|
||
return fmt.Errorf("标记任务 %s 失败状态出错: %w", taskID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// InsertCollectTask 建一条采集任务。
|
||
//
|
||
// `[必须]` 采集任务**不指定客户端**：assigned_client 为 NULL、status 为 pending，
|
||
// 谁领到就在领取时标记谁（见 #17 和 ClaimNextTask）。采集是纯读取操作，
|
||
// 哪台机器跑都一样，指定了反而会在那台机器关着的时候干等。
|
||
//
|
||
// goodsURL 必填 —— Client 契约里 pdd_goods_url 是 NOT NULL，
|
||
// 没有它客户端拿到任务也不知道去哪采。
|
||
func InsertCollectTask(q Execer, taskID, goodsID, goodsURL string) error {
|
||
if goodsID == "" || goodsURL == "" {
|
||
return fmt.Errorf("采集任务的商品 ID 和链接都不能为空")
|
||
}
|
||
now := model.NowISO()
|
||
_, err := q.Exec(`
|
||
INSERT INTO tasks (task_id, task_type, status, assigned_client,
|
||
pdd_goods_url, pdd_goods_id, created_at, updated_at)
|
||
VALUES (?, 'collect', 'pending', NULL, ?, ?, ?, ?)`,
|
||
taskID, goodsURL, goodsID, now, now)
|
||
if err != nil {
|
||
return fmt.Errorf("创建商品 %s 的采集任务失败: %w", goodsID, err)
|
||
}
|
||
return nil
|
||
}
|