diff --git a/internal/client/client.go b/internal/client/client.go index b875e66..91c88a3 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -1118,12 +1118,7 @@ func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID string, csrf *strin if csrf != nil { currentCSRF = *csrf } - var result struct { - types.APIResponse - Data struct { - Limit string `json:"limit"` - } `json:"data"` - } + var result types.APIResponse if err := c.postMultipart(c.Activity.BatchLimitURL, cookie, csrf, "批量限制", map[string]string{ "manualID": c.Activity.ManualID, "commodityID": commodityID, @@ -1135,9 +1130,15 @@ func (c *DouyuClient) BatchExchangeLimit(cookie, commodityID string, csrf *strin if result.Error != 0 { return 0, fmt.Errorf("批量限制接口失败: %s", result.Msg) } - limit, err := strconv.Atoi(result.Data.Limit) + var data struct { + Limit string `json:"limit"` + } + if err := json.Unmarshal(result.Data, &data); err != nil { + return 0, fmt.Errorf("批量限制数据解析失败: %w", err) + } + limit, err := strconv.Atoi(data.Limit) if err != nil { - return 0, fmt.Errorf("批量限制值无效: %q", result.Data.Limit) + return 0, fmt.Errorf("批量限制值无效: %q", data.Limit) } return limit, nil } @@ -1175,14 +1176,7 @@ func (c *DouyuClient) Exchange(cookie, commodityID string, csrf *string) (string if c.Activity.IsXiaodian { return c.xiaodianExchange(cookie, types.Product{CommodityID: commodityID}) } - var result struct { - types.APIResponse - Data struct { - ExchangeID int `json:"exchangeId"` - ExchangeNum string `json:"exchangeNum"` - CommodityTyp int `json:"commodityType"` - } `json:"data"` - } + var result types.APIResponse if c.Activity.UseCTN { // 电竞手册：用 ctn(acf_ccn) 鉴权，multipart 表单（与浏览器抓包一致） ctn := "" @@ -1234,7 +1228,15 @@ func (c *DouyuClient) Exchange(cookie, commodityID string, csrf *string) (string } return "", fmt.Errorf("兑换失败: %s", result.Msg) } - return fmt.Sprintf("exchangeId=%d,num=%s", result.Data.ExchangeID, result.Data.ExchangeNum), nil + var data struct { + ExchangeID int `json:"exchangeId"` + ExchangeNum string `json:"exchangeNum"` + CommodityTyp int `json:"commodityType"` + } + if err := json.Unmarshal(result.Data, &data); err != nil { + return "", fmt.Errorf("兑换数据解析失败: %w", err) + } + return fmt.Sprintf("exchangeId=%d,num=%s", data.ExchangeID, data.ExchangeNum), nil } // ExchangeProduct 正式兑换一个商品。和平小店需要商品支付类型/价格，因此应优先走这个入口。 diff --git a/internal/seckill/schedule_timer.go b/internal/seckill/schedule_timer.go index 8fe5e47..159490f 100644 --- a/internal/seckill/schedule_timer.go +++ b/internal/seckill/schedule_timer.go @@ -3,6 +3,7 @@ package seckill import ( "errors" "fmt" + "math" "sync/atomic" "time" @@ -30,7 +31,8 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, calibration := s.calibrateDouyuTime("定时") serverTime := calibration.EstimatedServerNow() - triggerTime := targetTime.Add(-internalAdvance) + advance := dynamicAdvance(calibration) + triggerTime := targetTime.Add(-advance) remain := triggerTime.Sub(serverTime) if remain <= 0 { return types.ScheduleResult{}, errors.New("目标时间已过") @@ -52,7 +54,8 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, s.scheduleMu.Lock() s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() { newCalibration := s.calibrateDouyuTime("临近") - s.scheduleTimers(ap, targetTime, triggerTime, newCalibration, req.Attempts, req.DelayMs, true) + newTriggerTime := targetTime.Add(-dynamicAdvance(newCalibration)) + s.scheduleTimers(ap, targetTime, newTriggerTime, newCalibration, req.Attempts, req.DelayMs, true) }) s.scheduleMu.Unlock() @@ -64,7 +67,7 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s", targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), - internalAdvance.Milliseconds(), + advance.Milliseconds(), serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), calibration.Offset.Milliseconds(), calibration.ErrorBound.Milliseconds(), @@ -73,7 +76,7 @@ func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, time.Now().Add(resyncRemain).In(constants.CST).Format("15:04:05.000"), ) - remainSecond := int(triggerTime.Sub(serverTime).Seconds()) + remainSecond := int(math.Ceil(triggerTime.Sub(serverTime).Seconds())) if remainSecond < 0 { remainSecond = 0 } @@ -95,26 +98,27 @@ func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime if remain < 0 { remain = 0 } - warmupRemain := remain - warmupBeforeTrigger - if warmupRemain < 0 { - warmupRemain = 0 - } - s.scheduleMu.Lock() defer s.scheduleMu.Unlock() - gen := atomic.AddInt64(&s.warmupGen, 1) - s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() { - if atomic.LoadInt64(&s.warmupGen) != gen { - return + if !resync { + warmupRemain := remain - warmupBeforeTrigger + if warmupRemain < 0 { + warmupRemain = 0 } - if s.clientSnapshot().Activity.IsXiaodian { - return - } - logger.LogEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(constants.CST).Format("15:04:05.000")) - settings := s.settingsSnapshot() - attempts := s.buildAttempts(accountProducts, settings.attempts) - s.warmupAttempts(attempts) - }) + gen := atomic.AddInt64(&s.warmupGen, 1) + s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() { + if atomic.LoadInt64(&s.warmupGen) != gen { + return + } + if s.clientSnapshot().Activity.IsXiaodian { + return + } + logger.LogEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(constants.CST).Format("15:04:05.000")) + settings := s.settingsSnapshot() + attempts := s.buildAttempts(accountProducts, settings.attempts) + s.warmupAttempts(attempts) + }) + } s.replaceTimerLocked(&s.schedule, remain, func() { now := time.Now() estimatedServer := now.Add(calibration.Offset) @@ -134,14 +138,13 @@ func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime s.RunNow(accountProducts) }) if resync { - logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s", + logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms", targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), calibration.Offset.Milliseconds(), calibration.ErrorBound.Milliseconds(), - calibration.RTT.Milliseconds(), - time.Now().Add(warmupRemain).In(constants.CST).Format("15:04:05.000")) + calibration.RTT.Milliseconds()) } } @@ -164,11 +167,19 @@ func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration { if err != nil { logger.LogEvent("时间", "warn", "%s校准降级: %v", label, err) } - logger.SetDouyuOffset(calibration.Offset) - logger.LogEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s", + if !calibration.Fallback { + s.recordCalibrationOffset(calibration.Offset) + } + smoothed := s.smoothedOffset() + if smoothed == 0 && calibration.Fallback { + smoothed = calibration.Offset + } + calibration.Offset = smoothed + logger.SetDouyuOffset(smoothed) + logger.LogEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 平滑偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s", label, calibration.EstimatedServerNow().In(constants.CST).Format("2006-01-02 15:04:05.000"), - calibration.Offset.Milliseconds(), + smoothed.Milliseconds(), calibration.ErrorBound.Milliseconds(), calibration.RTT.Milliseconds(), calibration.Samples, diff --git a/internal/seckill/scheduler.go b/internal/seckill/scheduler.go index e97747c..3798ac3 100644 --- a/internal/seckill/scheduler.go +++ b/internal/seckill/scheduler.go @@ -3,6 +3,7 @@ package seckill import ( "fmt" "math/rand" + "sort" "strings" "sync" "time" @@ -18,18 +19,19 @@ import ( ) const ( - warmupBeforeTrigger = 2 * time.Second - warmupMaxAge = 30 * time.Second - qualifiedMaxAge = 30 * time.Second - timeCalibrateRounds = 3 - timeCalibrateWait = 1200 * time.Millisecond - resyncBeforeTrigger = 10 * time.Second - seckillMaxDuration = 75 * time.Second - behaviorCooldown = 5 * time.Second - clickFastCooldown = 3 * time.Second - internalAdvance = 50 * time.Millisecond - maxCooldownCount = 5 - xiaodianDefaultDelay = 500 * time.Millisecond + warmupBeforeTrigger = 2 * time.Second + warmupMaxAge = 30 * time.Second + qualifiedMaxAge = 30 * time.Second + timeCalibrateRounds = 3 + timeCalibrateWait = 1200 * time.Millisecond + resyncBeforeTrigger = 3 * time.Second + calibrationHistorySize = 3 + seckillMaxDuration = 75 * time.Second + behaviorCooldown = 5 * time.Second + clickFastCooldown = 3 * time.Second + minInternalAdvance = 50 * time.Millisecond + maxCooldownCount = 5 + xiaodianDefaultDelay = 500 * time.Millisecond ) type roundOutcome struct { @@ -94,6 +96,9 @@ type Scheduler struct { productsMu sync.RWMutex products map[string]types.Product + + calibrationMu sync.Mutex + calibrationOffsets []time.Duration // 最近N次校准偏移，用于中位数平滑 } // SetStateChangeCallback 设置运行状态变化回调。 @@ -131,13 +136,14 @@ func New(cli *client.DouyuClient, paths types.AppPaths, font *truetype.Font, con 1, 256, ), - concurrency: concurrency, - attempts: attempts, - retryStages: cloneRetryStages(retryStages), - seckillMode: normalizeSeckillMode(seckillMode), - results: make(map[string]string), - attemptCache: make(map[string]*types.AttemptCache), - products: make(map[string]types.Product), + concurrency: concurrency, + attempts: attempts, + retryStages: cloneRetryStages(retryStages), + seckillMode: normalizeSeckillMode(seckillMode), + results: make(map[string]string), + attemptCache: make(map[string]*types.AttemptCache), + products: make(map[string]types.Product), + calibrationOffsets: make([]time.Duration, 0, calibrationHistorySize), } } @@ -939,6 +945,37 @@ func classifyXiaodianRetryStop(message string) (account bool, product bool, reas } } +// recordCalibrationOffset 记录校准偏移，保留最近N次用于中位数平滑。 +func (s *Scheduler) recordCalibrationOffset(offset time.Duration) { + s.calibrationMu.Lock() + defer s.calibrationMu.Unlock() + s.calibrationOffsets = append(s.calibrationOffsets, offset) + if len(s.calibrationOffsets) > calibrationHistorySize { + s.calibrationOffsets = s.calibrationOffsets[len(s.calibrationOffsets)-calibrationHistorySize:] + } +} + +// smoothedOffset 返回最近校准偏移的中位数，过滤跳变异常。 +func (s *Scheduler) smoothedOffset() time.Duration { + s.calibrationMu.Lock() + defer s.calibrationMu.Unlock() + if len(s.calibrationOffsets) == 0 { + return 0 + } + sorted := append([]time.Duration(nil), s.calibrationOffsets...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] }) + return sorted[len(sorted)/2] +} + +// dynamicAdvance 根据校准误差边界和 RTT 计算动态提前量，最少 50ms。 +func dynamicAdvance(calibration types.TimeCalibration) time.Duration { + advance := calibration.ErrorBound + calibration.RTT/2 + if advance < minInternalAdvance { + advance = minInternalAdvance + } + return advance +} + func uniqueAttemptAccounts(attempts []types.ExchangeAttempt) int { seen := make(map[string]struct{}, len(attempts)) for _, attempt := range attempts {