756 lines
31 KiB
Go
756 lines
31 KiB
Go
package purchase
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"time"
|
|
|
|
"go-admin/app/goauto/aimatching"
|
|
"go-admin/app/goauto/device"
|
|
"go-admin/app/goauto/models"
|
|
"go-admin/app/goauto/purchasecontract"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
func (s *Service) Next(ctx context.Context, token string) (*TaskPayload, error) {
|
|
d, err := device.NewService(s.DB).Authenticate(ctx, token)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if d.Status != models.DeviceStatusOnline {
|
|
return nil, fail(CodeStateConflict, "设备离线，不能领取采购任务")
|
|
}
|
|
var running models.PurchaseTask
|
|
if err = s.DB.WithContext(ctx).Where("device_id = ? AND status IN ?", d.ID, []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted}).First(&running).Error; err == nil {
|
|
return s.payload(running, nil, false)
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, internal(err)
|
|
}
|
|
// A completed probe reserves the device's purchase flow while the server
|
|
// resolves the exact specs. Returning that task as a waiting payload keeps
|
|
// the Agent from claiming another purchase or collection task and preserves
|
|
// the PDD page that the probe just inspected.
|
|
var waiting models.PurchaseTask
|
|
if err = s.DB.WithContext(ctx).
|
|
Where("device_id = ? AND status = ?", d.ID, models.PurchaseTaskStatusSpecProbePending).
|
|
Order("created_at, id").
|
|
First(&waiting).Error; err == nil {
|
|
return s.payload(waiting, nil, false)
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, internal(err)
|
|
}
|
|
now := s.Now()
|
|
var candidates []models.PurchaseTask
|
|
if err = s.DB.WithContext(ctx).Where("status IN ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?) AND (device_id IS NULL OR device_id = ?)", []string{models.PurchaseTaskStatusPending, models.PurchaseTaskStatusSpecProbePending}, now, d.ID).Order("CASE WHEN device_id IS NULL THEN 1 ELSE 0 END, created_at, id").Limit(100).Find(&candidates).Error; err != nil {
|
|
return nil, internal(err)
|
|
}
|
|
for _, t := range candidates {
|
|
if t.Status == models.PurchaseTaskStatusSpecProbePending && t.MappedColorSnapshot == "" && t.MappedSizeSnapshot == "" {
|
|
continue
|
|
}
|
|
if active, er := activePurchaseMatch(s.DB.WithContext(ctx), t.ID); er != nil {
|
|
return nil, er
|
|
} else if active {
|
|
continue
|
|
}
|
|
required, er := decodeStrings(t.RequiredCapabilitiesJSON)
|
|
if er != nil {
|
|
return nil, internal(er)
|
|
}
|
|
if er = ensureCapabilities(d, required); er == nil {
|
|
return s.payload(t, nil, false)
|
|
}
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
func (s *Service) Claim(ctx context.Context, taskID uint64, req ActionRequest, token string) (TaskPayload, error) {
|
|
if _, err := uuid.Parse(req.RequestID); err != nil {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "requestId 无效")
|
|
}
|
|
var out TaskPayload
|
|
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
d, err := device.NewService(tx).Authenticate(ctx, token)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if d.Status != models.DeviceStatusOnline {
|
|
return fail(CodeStateConflict, "设备离线，不能领取采购任务")
|
|
}
|
|
var t models.PurchaseTask
|
|
if err = tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&t, taskID).Error; err != nil {
|
|
return purchaseNotFound(err)
|
|
}
|
|
if t.ClaimRequestID != nil && *t.ClaimRequestID == req.RequestID && t.DeviceID != nil && *t.DeviceID == d.ID {
|
|
p, e := s.payload(t, nil, true)
|
|
out = *p
|
|
return e
|
|
}
|
|
if t.Status != models.PurchaseTaskStatusPending && t.Status != models.PurchaseTaskStatusSpecProbePending {
|
|
return fail(CodeStateConflict, "任务当前状态不能领取")
|
|
}
|
|
if active, e := activePurchaseMatch(tx, t.ID); e != nil {
|
|
return e
|
|
} else if active {
|
|
return fail(CodeStateConflict, "任务规格匹配尚未完成")
|
|
}
|
|
if t.DeviceID != nil && *t.DeviceID != d.ID {
|
|
return fail(CodeStateConflict, "任务已指定给其他设备")
|
|
}
|
|
if t.LeaseExpiresAt != nil && t.LeaseExpiresAt.After(s.Now()) {
|
|
return fail(CodeTaskClaimed, "任务已被领取")
|
|
}
|
|
if err = tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).Where("status IN ? AND lease_expires_at <= ?", []string{models.PurchaseTaskStatusPending, models.PurchaseTaskStatusSpecProbePending}, s.Now()).Updates(map[string]any{"device_run_slot": nil, "account_run_slot": nil}).Error; err != nil {
|
|
return internal(err)
|
|
}
|
|
required, e := decodeStrings(t.RequiredCapabilitiesJSON)
|
|
if e != nil {
|
|
return internal(e)
|
|
}
|
|
if e = ensureCapabilities(d, required); e != nil {
|
|
return e
|
|
}
|
|
if e = ensureDeviceFree(tx, d.ID, t.ID, s.Now()); e != nil {
|
|
return e
|
|
}
|
|
if e = ensureAccountFree(tx, t.PDDAccountID, t.ID, s.Now()); e != nil {
|
|
return e
|
|
}
|
|
lease := s.Now().Add(s.lease())
|
|
one := uint8(1)
|
|
updates := map[string]any{"device_id": d.ID, "lease_expires_at": lease, "lease_version": gorm.Expr("lease_version + 1"), "claim_request_id": req.RequestID, "device_run_slot": one}
|
|
if t.PDDAccountID != nil {
|
|
updates["account_run_slot"] = one
|
|
}
|
|
result := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).Where("id = ? AND status IN ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?)", t.ID, []string{models.PurchaseTaskStatusPending, models.PurchaseTaskStatusSpecProbePending}, s.Now()).Updates(updates)
|
|
if result.Error != nil {
|
|
return conflictOrInternal(result.Error)
|
|
}
|
|
if result.RowsAffected != 1 {
|
|
return fail(CodeTaskClaimed, "任务已被其他设备领取")
|
|
}
|
|
if e = tx.First(&t, t.ID).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
p, e := s.payload(t, nil, false)
|
|
out = *p
|
|
return e
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
func (s *Service) Start(ctx context.Context, taskID uint64, req ActionRequest, token string) (TaskPayload, error) {
|
|
if _, err := uuid.Parse(req.RequestID); err != nil {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "requestId 无效")
|
|
}
|
|
var out TaskPayload
|
|
var committedFailure error
|
|
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
d, err := device.NewService(tx).Authenticate(ctx, token)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var t models.PurchaseTask
|
|
if err = tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&t, taskID).Error; err != nil {
|
|
return purchaseNotFound(err)
|
|
}
|
|
var existing models.PurchaseTaskAttempt
|
|
if err = tx.Where("start_request_id = ?", req.RequestID).First(&existing).Error; err == nil {
|
|
if existing.TaskID != t.ID || existing.DeviceID == nil || *existing.DeviceID != d.ID {
|
|
return fail(CodeResultConflict, "start requestId 已用于其他任务")
|
|
}
|
|
p, e := s.payload(t, &existing, true)
|
|
out = *p
|
|
return e
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return internal(err)
|
|
}
|
|
if t.Status != models.PurchaseTaskStatusPending && t.Status != models.PurchaseTaskStatusSpecProbePending {
|
|
return fail(CodeStateConflict, "任务当前状态不能开始")
|
|
}
|
|
if active, e := activePurchaseMatch(tx, t.ID); e != nil {
|
|
return e
|
|
} else if active {
|
|
return fail(CodeStateConflict, "任务规格匹配尚未完成")
|
|
}
|
|
if t.DeviceID == nil || *t.DeviceID != d.ID {
|
|
return fail(CodeStateConflict, "任务不属于当前设备")
|
|
}
|
|
if t.LeaseExpiresAt == nil || !t.LeaseExpiresAt.After(s.Now()) {
|
|
return fail(CodeLeaseExpired, "任务租约已过期，请重新领取")
|
|
}
|
|
required, e := decodeStrings(t.RequiredCapabilitiesJSON)
|
|
if e != nil {
|
|
return internal(e)
|
|
}
|
|
if e = ensureCapabilities(d, required); e != nil {
|
|
return e
|
|
}
|
|
if e = ensureDeviceFree(tx, d.ID, t.ID, s.Now()); e != nil {
|
|
return e
|
|
}
|
|
if e = ensureAccountFree(tx, t.PDDAccountID, t.ID, s.Now()); e != nil {
|
|
return e
|
|
}
|
|
phase := purchaseAttemptPhase(t)
|
|
ruleSnapshotHash := purchaseRuleSnapshotHash(t.RuleSnapshot)
|
|
now := s.Now()
|
|
var a models.PurchaseTaskAttempt
|
|
if e = tx.Where("task_id = ? AND status = ?", t.ID, models.PurchaseAttemptStatusPending).Order("attempt_number DESC, id DESC").First(&a).Error; e == nil {
|
|
if a.DeviceID == nil || *a.DeviceID != d.ID || a.RuleSnapshotHash != ruleSnapshotHash || a.Phase != phase {
|
|
failureCode := CodeStateConflict
|
|
message := "采购任务执行快照校验失败，请重新创建任务"
|
|
a.Status = models.PurchaseAttemptStatusFailed
|
|
a.ErrorCode = &failureCode
|
|
a.ErrorMessage = &message
|
|
a.FinishedAt = &now
|
|
if e = tx.Omit("Task").Save(&a).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
if e = t.SetStatus(models.PurchaseTaskStatusFailed); e != nil {
|
|
return internal(e)
|
|
}
|
|
t.ErrorCode = &failureCode
|
|
t.ErrorMessage = &message
|
|
t.LeaseExpiresAt = nil
|
|
t.ClaimRequestID = nil
|
|
t.StatusVersion++
|
|
t.StatusChangedAt = now
|
|
if e = tx.Save(&t).Error; e != nil {
|
|
return conflictOrInternal(e)
|
|
}
|
|
committedFailure = fail(CodeStateConflict, message)
|
|
return nil
|
|
}
|
|
a.Status = models.PurchaseAttemptStatusRunning
|
|
a.StartRequestID = &req.RequestID
|
|
a.StartedAt = &now
|
|
if e = tx.Save(&a).Error; e != nil {
|
|
return conflictOrInternal(e)
|
|
}
|
|
} else if errors.Is(e, gorm.ErrRecordNotFound) {
|
|
var count int64
|
|
if e = tx.Model(&models.PurchaseTaskAttempt{}).Where("task_id = ?", t.ID).Count(&count).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
a = models.PurchaseTaskAttempt{TaskID: t.ID, AttemptID: uuid.NewString(), AttemptNumber: int(count) + 1, Phase: phase, Status: models.PurchaseAttemptStatusRunning, DeviceID: &d.ID, RuleSnapshotHash: ruleSnapshotHash, SpecDecisionSnapshot: t.SpecDecisionSnapshot, StartRequestID: &req.RequestID, StartedAt: &now}
|
|
if e = tx.Omit("Task").Create(&a).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
} else {
|
|
return internal(e)
|
|
}
|
|
if e = t.SetStatus(models.PurchaseTaskStatusRunning); e != nil {
|
|
return internal(e)
|
|
}
|
|
t.StatusVersion++
|
|
t.StatusChangedAt = now
|
|
t.LeaseExpiresAt = ptrTime(now.Add(s.lease()))
|
|
if e = tx.Save(&t).Error; e != nil {
|
|
return conflictOrInternal(e)
|
|
}
|
|
p, e := s.payload(t, &a, false)
|
|
out = *p
|
|
return e
|
|
})
|
|
if err == nil && committedFailure != nil {
|
|
return out, committedFailure
|
|
}
|
|
return out, err
|
|
}
|
|
|
|
func purchaseRuleSnapshotHash(snapshot string) string {
|
|
digest := sha256.Sum256([]byte(snapshot))
|
|
return hex.EncodeToString(digest[:])
|
|
}
|
|
|
|
func (s *Service) MarkOrderSubmitStarted(ctx context.Context, taskID uint64, req ActionRequest, token string) (TaskPayload, error) {
|
|
return s.withRunning(ctx, taskID, token, func(tx *gorm.DB, t *models.PurchaseTask, a *models.PurchaseTaskAttempt, d models.AgentDevice) (TaskPayload, error) {
|
|
if t.ExecutionMode != models.PurchaseExecutionModeLive {
|
|
return TaskPayload{}, fail(CodeStateConflict, "演练任务不能创建订单")
|
|
}
|
|
if a.Phase != models.PurchaseAttemptPhasePurchase {
|
|
return TaskPayload{}, fail(CodeStateConflict, "规格探测阶段不能进入创建订单边界")
|
|
}
|
|
if _, e := uuid.Parse(req.RequestID); e != nil {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "requestId 无效")
|
|
}
|
|
if t.OrderSubmitRequestID != nil {
|
|
if *t.OrderSubmitRequestID == req.RequestID {
|
|
return valuePayload(s, t, a, true)
|
|
}
|
|
return TaskPayload{}, fail(CodeResultConflict, "订单提交状态已记录")
|
|
}
|
|
now := s.Now()
|
|
if e := t.SetStatus(models.PurchaseTaskStatusOrderSubmitStarted); e != nil {
|
|
return TaskPayload{}, internal(e)
|
|
}
|
|
t.OrderSubmitRequestID = &req.RequestID
|
|
t.IrreversibleAt = &now
|
|
t.StatusVersion++
|
|
t.StatusChangedAt = now
|
|
if e := tx.Save(t).Error; e != nil {
|
|
return TaskPayload{}, conflictOrInternal(e)
|
|
}
|
|
return valuePayload(s, t, a, false)
|
|
})
|
|
}
|
|
|
|
func (s *Service) SubmitResult(ctx context.Context, taskID uint64, req ResultRequest, token string) (TaskPayload, error) {
|
|
raw, _ := json.Marshal(req)
|
|
digest := hashBytes(raw)
|
|
deviceRecord, authErr := device.NewService(s.DB).Authenticate(ctx, token)
|
|
if authErr != nil {
|
|
return TaskPayload{}, authErr
|
|
}
|
|
var replayTask models.PurchaseTask
|
|
var replayAttempt models.PurchaseTaskAttempt
|
|
if err := s.DB.WithContext(ctx).First(&replayTask, taskID).Error; err == nil {
|
|
if replayTask.DeviceID == nil || *replayTask.DeviceID != deviceRecord.ID {
|
|
return TaskPayload{}, fail(CodeStateConflict, "任务不属于当前设备")
|
|
}
|
|
if err := s.DB.WithContext(ctx).Where("task_id = ? AND attempt_id = ?", taskID, req.TaskAttemptID).First(&replayAttempt).Error; err == nil && replayAttempt.ResultRequestID != nil {
|
|
if *replayAttempt.ResultRequestID == req.RequestID && replayAttempt.ResultHash != nil && *replayAttempt.ResultHash == digest {
|
|
return valuePayload(s, &replayTask, &replayAttempt, true)
|
|
}
|
|
return TaskPayload{}, fail(CodeResultConflict, "同一次执行已提交不同结果")
|
|
}
|
|
}
|
|
payload, err := s.withRunning(ctx, taskID, token, func(tx *gorm.DB, t *models.PurchaseTask, a *models.PurchaseTaskAttempt, d models.AgentDevice) (TaskPayload, error) {
|
|
if req.TaskAttemptID == "" || req.TaskAttemptID != a.AttemptID {
|
|
return TaskPayload{}, fail(CodeStateConflict, "taskAttemptId 与当前执行不一致")
|
|
}
|
|
if strings.TrimSpace(req.RequestID) == "" {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "requestId 不能为空")
|
|
}
|
|
if a.ResultRequestID != nil {
|
|
if *a.ResultRequestID == req.RequestID && a.ResultHash != nil && *a.ResultHash == digest {
|
|
return valuePayload(s, t, a, true)
|
|
}
|
|
return TaskPayload{}, fail(CodeResultConflict, "同一次执行已提交不同结果")
|
|
}
|
|
if req.ActualUnitPriceCent != nil && *req.ActualUnitPriceCent < 0 {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "实际单价无效")
|
|
}
|
|
now := s.Now()
|
|
next := ""
|
|
switch req.ResultType {
|
|
case "spec_probe_completed":
|
|
if len(req.ProbedSpecs) == 0 {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "规格探测结果无效")
|
|
}
|
|
if t.SpecDecisionRequestID != nil {
|
|
// One slow-path decision has already been frozen for this task. A
|
|
// second probe means the Agent still could not select that exact
|
|
// decision. Fail closed instead of clearing the auditable decision
|
|
// and leaving an active task that can be dispatched forever.
|
|
next = models.PurchaseTaskStatusFailed
|
|
a.Status = models.PurchaseAttemptStatusFailed
|
|
code, message := CodeSpecReprobeRejected, "该任务的规格探测资格已经使用，重复探测已被拒绝"
|
|
t.ErrorCode, t.ErrorMessage = &code, &message
|
|
} else {
|
|
next = models.PurchaseTaskStatusSpecProbePending
|
|
a.Status = models.PurchaseAttemptStatusCompleted
|
|
t.SpecSource = "unresolved"
|
|
t.MappedColorSnapshot = ""
|
|
t.MappedSizeSnapshot = ""
|
|
}
|
|
case "rehearsal_completed":
|
|
if t.ExecutionMode != models.PurchaseExecutionModeRehearsal {
|
|
return TaskPayload{}, fail(CodeStateConflict, "正式任务不能提交演练结果")
|
|
}
|
|
next = models.PurchaseTaskStatusRehearsalCompleted
|
|
a.Status = models.PurchaseAttemptStatusCompleted
|
|
t.ActualUnitPriceCent = req.ActualUnitPriceCent
|
|
case "order_created":
|
|
if t.ExecutionMode != models.PurchaseExecutionModeLive || t.Status != models.PurchaseTaskStatusOrderSubmitStarted || strings.TrimSpace(req.PDDOrderNo) == "" || req.OrderSubmittedAt == nil {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "订单号或下单时间缺失")
|
|
}
|
|
next = models.PurchaseTaskStatusOrderCreated
|
|
a.Status = models.PurchaseAttemptStatusCompleted
|
|
t.PDDOrderNo = &req.PDDOrderNo
|
|
t.OrderSubmittedAt = req.OrderSubmittedAt
|
|
t.ActualUnitPriceCent = req.ActualUnitPriceCent
|
|
case "order_result_unknown":
|
|
if t.ExecutionMode != models.PurchaseExecutionModeLive || t.Status != models.PurchaseTaskStatusOrderSubmitStarted {
|
|
return TaskPayload{}, fail(CodeStateConflict, "当前任务不能标记订单结果未知")
|
|
}
|
|
if strings.TrimSpace(req.PDDOrderNo) != "" || req.OrderSubmittedAt != nil {
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "订单结果未知时不能提交订单号或下单时间")
|
|
}
|
|
failureCode, failureMessage, failureErr := normalizeOrderUnknownFailure(req.ErrorCode, req.ErrorMessage)
|
|
if failureErr != nil {
|
|
return TaskPayload{}, failureErr
|
|
}
|
|
next = models.PurchaseTaskStatusOrderResultUnknown
|
|
a.Status = models.PurchaseAttemptStatusFailed
|
|
a.ErrorCode, a.ErrorMessage = &failureCode, &failureMessage
|
|
t.ErrorCode, t.ErrorMessage = &failureCode, &failureMessage
|
|
t.ActualUnitPriceCent = req.ActualUnitPriceCent
|
|
case "failed":
|
|
next = models.PurchaseTaskStatusFailed
|
|
a.Status = models.PurchaseAttemptStatusFailed
|
|
t.ErrorCode = &req.ErrorCode
|
|
t.ErrorMessage = &req.ErrorMessage
|
|
t.ActualUnitPriceCent = req.ActualUnitPriceCent
|
|
default:
|
|
return TaskPayload{}, fail(CodeInvalidRequest, "resultType 无效")
|
|
}
|
|
if e := t.SetStatus(next); e != nil {
|
|
return TaskPayload{}, internal(e)
|
|
}
|
|
t.LeaseExpiresAt = nil
|
|
t.StatusVersion++
|
|
t.StatusChangedAt = now
|
|
a.ResultRequestID = &req.RequestID
|
|
a.ResultHash = &digest
|
|
a.ResultType = &req.ResultType
|
|
a.FinishedAt = &now
|
|
if req.ResultType == "spec_probe_completed" {
|
|
a.SpecDecisionSnapshot = string(req.ProbedSpecs)
|
|
}
|
|
if e := tx.Omit("Task").Save(a).Error; e != nil {
|
|
return TaskPayload{}, internal(e)
|
|
}
|
|
if e := tx.Save(t).Error; e != nil {
|
|
return TaskPayload{}, conflictOrInternal(e)
|
|
}
|
|
return valuePayload(s, t, a, false)
|
|
})
|
|
if err != nil || req.ResultType != "spec_probe_completed" || payload.Replayed || payload.Status != models.PurchaseTaskStatusSpecProbePending {
|
|
return payload, err
|
|
}
|
|
return s.resolveProbedSpecs(ctx, taskID, req.TaskAttemptID, req.ProbedSpecs)
|
|
}
|
|
|
|
var allowedOrderUnknownFailures = map[string]string{
|
|
CodeOrderResultUnknown: "无法确认订单是否创建，请人工检查",
|
|
CodeOrderEmptyTimeout: "等待订单页面时无障碍窗口持续为空",
|
|
CodeOrderChooserBack: "系统应用选择页无法安全返回",
|
|
CodeOrderWechatRestore: "从微信恢复拼多多的请求失败",
|
|
CodeOrderWechatTimeout: "从微信恢复拼多多后未在限定时间到达订单页面",
|
|
CodeOrderUnexpectedApp: "核单期间出现未授权应用",
|
|
CodeOrderPaymentBack: "支付页无法安全返回订单详情",
|
|
CodeOrderPaymentRepeat: "支付页重复出现，已停止自动核单",
|
|
CodeOrderContextMissing: "限定时间内没有出现订单详情证据",
|
|
CodeOrderNoMissing: "订单详情缺少可读取的订单号",
|
|
CodeOrderNoAmbiguous: "订单详情出现多个订单号，无法唯一确认",
|
|
CodeOrderTimeMissing: "订单详情缺少可读取的下单时间",
|
|
CodeOrderTimeInvalid: "订单详情的下单时间格式无法确认",
|
|
CodeOrderUnpaidMissing: "订单详情缺少待付款状态证据",
|
|
}
|
|
|
|
func normalizeOrderUnknownFailure(code, message string) (string, string, error) {
|
|
code, message = strings.TrimSpace(code), strings.TrimSpace(message)
|
|
if code == "" && message == "" {
|
|
return CodeOrderResultUnknown, "无法确认订单是否创建，请人工检查", nil
|
|
}
|
|
canonicalMessage, ok := allowedOrderUnknownFailures[code]
|
|
if !ok || message == "" {
|
|
return "", "", fail(CodeInvalidRequest, "订单核单失败阶段无效")
|
|
}
|
|
return code, canonicalMessage, nil
|
|
}
|
|
|
|
func (s *Service) ApplySpecDecision(ctx context.Context, taskID uint64, req SpecDecisionRequest) (models.PurchaseTask, bool, error) {
|
|
operatorID := req.OperatorID
|
|
return s.applySpecDecision(ctx, taskID, req, &operatorID)
|
|
}
|
|
|
|
func (s *Service) applySpecDecision(ctx context.Context, taskID uint64, req SpecDecisionRequest, operatorID *uint64) (models.PurchaseTask, bool, error) {
|
|
var t models.PurchaseTask
|
|
replayed := false
|
|
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if e := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&t, taskID).Error; e != nil {
|
|
return purchaseNotFound(e)
|
|
}
|
|
if strings.TrimSpace(req.RequestID) == "" || (operatorID != nil && *operatorID == 0) {
|
|
return fail(CodeInvalidRequest, "requestId 或操作人无效")
|
|
}
|
|
var a models.PurchaseTaskAttempt
|
|
if e := tx.Where("task_id = ? AND attempt_id = ?", t.ID, req.TaskAttemptID).First(&a).Error; e != nil {
|
|
return purchaseNotFound(e)
|
|
}
|
|
if a.Status != models.PurchaseAttemptStatusCompleted || a.ResultType == nil || *a.ResultType != "spec_probe_completed" {
|
|
return fail(CodeStateConflict, "该 attempt 没有可固化的规格探测结果")
|
|
}
|
|
if t.SpecDecisionRequestID != nil {
|
|
same := t.MappedColorSnapshot == req.MappedColor && t.MappedSizeSnapshot == req.MappedSize
|
|
if same && *t.SpecDecisionRequestID == req.RequestID {
|
|
replayed = true
|
|
return nil
|
|
}
|
|
return fail(CodeResultConflict, "同一次规格探测的决策已经固化，不能修改")
|
|
}
|
|
if t.Status != models.PurchaseTaskStatusSpecProbePending {
|
|
return fail(CodeStateConflict, "任务不在待规格决策状态")
|
|
}
|
|
if req.Source != "ai_match" && req.Source != "exact_match" && req.Source != "manual_mapping" {
|
|
return fail(CodeInvalidRequest, "规格决策来源无效")
|
|
}
|
|
decision := req.Decision
|
|
if len(decision) == 0 {
|
|
decision = []byte("{}")
|
|
}
|
|
t.MappedColorSnapshot = strings.TrimSpace(req.MappedColor)
|
|
t.MappedSizeSnapshot = strings.TrimSpace(req.MappedSize)
|
|
if !req.NoMatch && ((t.TargetColorSnapshot != "" && t.MappedColorSnapshot == "") || (t.TargetSizeSnapshot != "" && t.MappedSizeSnapshot == "")) {
|
|
return fail(CodeMappingRequired, "没有找到可用的商品规格")
|
|
}
|
|
t.SpecSource = req.Source
|
|
t.SpecDecisionSnapshot = string(decision)
|
|
t.SpecDecisionRequestID = &req.RequestID
|
|
t.SpecDecisionBy = operatorID
|
|
if req.NoMatch {
|
|
code, message := strings.TrimSpace(req.FailureCode), strings.TrimSpace(req.FailureMessage)
|
|
if code == "" {
|
|
code = "PURCHASE_SPEC_NOT_MATCHED"
|
|
}
|
|
if message == "" {
|
|
message = "没有找到可用的商品规格"
|
|
}
|
|
t.ErrorCode, t.ErrorMessage = &code, &message
|
|
if e := t.SetStatus(models.PurchaseTaskStatusFailed); e != nil {
|
|
return internal(e)
|
|
}
|
|
} else if e := t.SetStatus(models.PurchaseTaskStatusPending); e != nil {
|
|
return internal(e)
|
|
}
|
|
t.StatusVersion++
|
|
t.StatusChangedAt = s.Now()
|
|
return tx.Save(&t).Error
|
|
})
|
|
return t, replayed, err
|
|
}
|
|
|
|
// resolveProbedSpecs runs after the probe attempt has committed and released
|
|
// the device lease. The network call therefore never holds a task-row lock.
|
|
// Every outcome is persisted before the Agent can receive a second attempt:
|
|
// a valid exact label returns the task to pending; no result fails the task.
|
|
func (s *Service) resolveProbedSpecs(ctx context.Context, taskID uint64, attemptID string, raw json.RawMessage) (TaskPayload, error) {
|
|
var task models.PurchaseTask
|
|
if err := s.DB.WithContext(ctx).First(&task, taskID).Error; err != nil {
|
|
return TaskPayload{}, purchaseNotFound(err)
|
|
}
|
|
candidates, complete := probedCandidates(raw, task.TargetColorSnapshot, task.TargetSizeSnapshot)
|
|
request := aimatching.MatchRequest{TargetColor: task.TargetColorSnapshot, TargetSize: task.TargetSizeSnapshot, Colors: candidates.Colors, Sizes: candidates.Sizes}
|
|
decision := SpecDecisionRequest{RequestID: uuid.NewString(), TaskAttemptID: attemptID, Source: aimatching.SourceAI}
|
|
if !complete {
|
|
snapshot, marshalErr := json.Marshal(aimatching.NoMatchDecision(request, aimatching.SourceAI, "规格探测结果没有包含所需的可选颜色或尺码"))
|
|
if marshalErr != nil {
|
|
return TaskPayload{}, internal(marshalErr)
|
|
}
|
|
decision.NoMatch, decision.Decision = true, snapshot
|
|
decision.FailureCode, decision.FailureMessage = "PURCHASE_SPEC_NOT_MATCHED", "没有找到可采购的 PDD 颜色或尺码"
|
|
} else {
|
|
matched, matchErr := s.resolveProbedMatch(ctx, task, request)
|
|
valid := matchErr == nil && (matched.Source == "manual_mapping" || matched.Source == aimatching.SourceExact || matched.Source == aimatching.SourceAI) &&
|
|
matchCandidateValid(request.TargetColor, matched.MappedColor, request.Colors) &&
|
|
matchCandidateValid(request.TargetSize, matched.MappedSize, request.Sizes)
|
|
if valid {
|
|
snapshot, marshalErr := json.Marshal(matched.Decision)
|
|
if marshalErr != nil {
|
|
return TaskPayload{}, internal(marshalErr)
|
|
}
|
|
decision.MappedColor, decision.MappedSize, decision.Source, decision.Decision = matched.MappedColor, matched.MappedSize, matched.Source, snapshot
|
|
} else {
|
|
reason := purchaseMatchReason(matchErr)
|
|
if matchErr == nil {
|
|
reason = "AI 规格匹配结果不属于当次 PDD 候选"
|
|
}
|
|
snapshot, marshalErr := json.Marshal(aimatching.NoMatchDecision(request, aimatching.SourceAI, reason))
|
|
if marshalErr != nil {
|
|
return TaskPayload{}, internal(marshalErr)
|
|
}
|
|
decision.NoMatch, decision.Decision = true, snapshot
|
|
decision.FailureCode, decision.FailureMessage = "PURCHASE_SPEC_NOT_MATCHED", reason
|
|
}
|
|
}
|
|
updated, _, err := s.applySpecDecision(ctx, taskID, decision, nil)
|
|
if err != nil {
|
|
return TaskPayload{}, err
|
|
}
|
|
return valuePayload(s, &updated, nil, false)
|
|
}
|
|
|
|
func purchaseMatchReason(err error) string {
|
|
var matchErr *aimatching.Error
|
|
if errors.As(err, &matchErr) {
|
|
switch matchErr.Code {
|
|
case aimatching.CodeNotConfigured:
|
|
return "规格未匹配，AI 规格匹配未启用"
|
|
case aimatching.CodeProviderUnavailable:
|
|
return "AI 规格匹配暂时不可用，请稍后重新创建采购任务"
|
|
}
|
|
}
|
|
return "已采集到当前规格，但未能确定颜色或尺码映射"
|
|
}
|
|
|
|
type probedSpecCandidates struct {
|
|
Colors []string
|
|
Sizes []string
|
|
}
|
|
|
|
func probedCandidates(raw json.RawMessage, targetColor, targetSize string) (probedSpecCandidates, bool) {
|
|
var payload struct {
|
|
Dimensions []struct {
|
|
Key string `json:"key"`
|
|
Values []string `json:"values"`
|
|
} `json:"dimensions"`
|
|
}
|
|
if json.Unmarshal(raw, &payload) != nil {
|
|
return probedSpecCandidates{}, false
|
|
}
|
|
result := probedSpecCandidates{}
|
|
for _, dimension := range payload.Dimensions {
|
|
values := make([]string, 0, len(dimension.Values))
|
|
for _, value := range dimension.Values {
|
|
if value = strings.TrimSpace(value); value != "" {
|
|
values = append(values, value)
|
|
}
|
|
}
|
|
switch strings.ToLower(strings.TrimSpace(dimension.Key)) {
|
|
case "color":
|
|
result.Colors = append(result.Colors, values...)
|
|
case "size":
|
|
result.Sizes = append(result.Sizes, values...)
|
|
}
|
|
}
|
|
if strings.TrimSpace(targetColor) != "" && len(result.Colors) == 0 {
|
|
return result, false
|
|
}
|
|
if strings.TrimSpace(targetSize) != "" && len(result.Sizes) == 0 {
|
|
return result, false
|
|
}
|
|
return result, true
|
|
}
|
|
|
|
func (s *Service) withRunning(ctx context.Context, taskID uint64, token string, fn func(*gorm.DB, *models.PurchaseTask, *models.PurchaseTaskAttempt, models.AgentDevice) (TaskPayload, error)) (TaskPayload, error) {
|
|
var out TaskPayload
|
|
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
d, e := device.NewService(tx).Authenticate(ctx, token)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
var t models.PurchaseTask
|
|
if e = tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&t, taskID).Error; e != nil {
|
|
return purchaseNotFound(e)
|
|
}
|
|
if t.DeviceID == nil || *t.DeviceID != d.ID {
|
|
return fail(CodeStateConflict, "任务不属于当前设备")
|
|
}
|
|
if t.Status != models.PurchaseTaskStatusRunning && t.Status != models.PurchaseTaskStatusOrderSubmitStarted {
|
|
return fail(CodeStateConflict, "任务当前状态不能提交结果")
|
|
}
|
|
if t.LeaseExpiresAt == nil || !t.LeaseExpiresAt.After(s.Now()) {
|
|
return fail(CodeLeaseExpired, "任务租约已过期")
|
|
}
|
|
required, e := decodeStrings(t.RequiredCapabilitiesJSON)
|
|
if e != nil {
|
|
return internal(e)
|
|
}
|
|
if e = ensureCapabilities(d, required); e != nil {
|
|
return e
|
|
}
|
|
var a models.PurchaseTaskAttempt
|
|
if e = tx.Where("task_id = ? AND status = ?", t.ID, models.PurchaseAttemptStatusRunning).Order("attempt_number DESC").First(&a).Error; e != nil {
|
|
return purchaseNotFound(e)
|
|
}
|
|
p, e := fn(tx, &t, &a, d)
|
|
out = p
|
|
return e
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
func ensureDeviceFree(tx *gorm.DB, deviceID, taskID uint64, now time.Time) error {
|
|
var count int64
|
|
if e := tx.Model(&models.PurchaseTask{}).Where("id <> ? AND device_id = ? AND (status IN ? OR (status = ? AND lease_expires_at > ?))", taskID, deviceID, []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted, models.PurchaseTaskStatusSpecProbePending}, models.PurchaseTaskStatusPending, now).Count(&count).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
if count > 0 {
|
|
return fail(CodeDeviceBusy, "设备已有采购任务")
|
|
}
|
|
if e := tx.Model(&models.CollectionTask{}).Where("device_id = ? AND (status = ? OR (status = ? AND lease_expires_at > ?))", deviceID, models.TaskStatusRunning, models.TaskStatusPending, now).Count(&count).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
if count > 0 {
|
|
return fail(CodeDeviceBusy, "设备已有采集任务")
|
|
}
|
|
return nil
|
|
}
|
|
func ensureAccountFree(tx *gorm.DB, accountID *uint64, taskID uint64, now time.Time) error {
|
|
if accountID == nil {
|
|
return nil
|
|
}
|
|
var count int64
|
|
if e := tx.Model(&models.PurchaseTask{}).Where("id <> ? AND pdd_account_id = ? AND (status IN ? OR (status = ? AND lease_expires_at > ?))", taskID, *accountID, []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted, models.PurchaseTaskStatusSpecProbePending}, models.PurchaseTaskStatusPending, now).Count(&count).Error; e != nil {
|
|
return internal(e)
|
|
}
|
|
if count > 0 {
|
|
return fail(CodeDeviceBusy, "拼多多账号已有采购任务")
|
|
}
|
|
return nil
|
|
}
|
|
func (s *Service) payload(t models.PurchaseTask, a *models.PurchaseTaskAttempt, replayed bool) (*TaskPayload, error) {
|
|
p := &TaskPayload{TaskID: t.ID, ExecutionMode: t.ExecutionMode, Status: t.Status, DeviceID: t.DeviceID, PDDProductID: t.PDDProductID, PDDURL: t.PDDURLSnapshot, PDDGoodsID: t.PDDGoodsIDSnapshot, TargetColor: t.TargetColorSnapshot, TargetSize: t.TargetSizeSnapshot, MappedColor: t.MappedColorSnapshot, MappedSize: t.MappedSizeSnapshot, SpecResolutionAllowed: specResolutionAllowed(t), Quantity: t.Quantity, MinUnitPriceCent: t.MinUnitPriceCent, MaxUnitPriceCent: t.MaxUnitPriceCent, Currency: t.Currency, AddressSuffix: t.AddressSuffix, RuleSnapshot: json.RawMessage(t.RuleSnapshot), LeaseExpiresAt: t.LeaseExpiresAt, LeaseVersion: t.LeaseVersion, Replayed: replayed}
|
|
if a != nil {
|
|
p.TaskAttemptID = a.AttemptID
|
|
p.AttemptNumber = a.AttemptNumber
|
|
p.Phase = a.Phase
|
|
p.RuleSnapshotHash = a.RuleSnapshotHash
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func specResolutionAllowed(t models.PurchaseTask) bool {
|
|
if t.SpecDecisionRequestID != nil || t.TaskType == models.PurchaseTaskTypeStock || t.SpecSource == "direct_select" {
|
|
return false
|
|
}
|
|
required, err := decodeStrings(t.RequiredCapabilitiesJSON)
|
|
return err == nil && containsString(required, purchasecontract.CapabilitySpecProbeV1)
|
|
}
|
|
func valuePayload(s *Service, t *models.PurchaseTask, a *models.PurchaseTaskAttempt, replayed bool) (TaskPayload, error) {
|
|
p, e := s.payload(*t, a, replayed)
|
|
return *p, e
|
|
}
|
|
func purchaseNotFound(err error) error {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return fail(CodeTaskNotFound, "采购任务不存在")
|
|
}
|
|
return internal(err)
|
|
}
|
|
func conflictOrInternal(err error) error {
|
|
if isDuplicate(err) {
|
|
return fail(CodeDeviceBusy, "设备或拼多多账号已有运行任务")
|
|
}
|
|
return internal(err)
|
|
}
|
|
func decodeStrings(raw string) ([]string, error) {
|
|
var v []string
|
|
err := json.Unmarshal([]byte(raw), &v)
|
|
return v, err
|
|
}
|
|
func ptrTime(v time.Time) *time.Time { return &v }
|
|
func (s *Service) lease() time.Duration {
|
|
if s.LeaseDuration <= 0 {
|
|
return DefaultLeaseDuration
|
|
}
|
|
return s.LeaseDuration
|
|
}
|
|
|
|
func activePurchaseMatch(db *gorm.DB, taskID uint64) (bool, error) {
|
|
var count int64
|
|
if err := db.Model(&models.PurchaseSpecMatchWorkItem{}).Where("purchase_task_id = ? AND status IN ?", taskID, []string{models.PurchaseMatchPending, models.PurchaseMatchRunning, models.PurchaseMatchRetryWait, models.PurchaseMatchManualRequired}).Count(&count).Error; err != nil {
|
|
return false, internal(err)
|
|
}
|
|
return count > 0, nil
|
|
}
|