980 lines
30 KiB
Go
980 lines
30 KiB
Go
package seckill
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"errors"
|
||
"fmt"
|
||
"math/rand"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/golang/freetype/truetype"
|
||
|
||
"douyu_go/internal/client"
|
||
"douyu_go/internal/constants"
|
||
"douyu_go/internal/image"
|
||
"douyu_go/internal/logger"
|
||
"douyu_go/internal/types"
|
||
"douyu_go/internal/utils"
|
||
)
|
||
|
||
const (
|
||
warmupBeforeTrigger = 2 * time.Second
|
||
warmupMaxAge = 30 * time.Second
|
||
timeCalibrateRounds = 3
|
||
timeCalibrateWait = 1200 * time.Millisecond
|
||
resyncBeforeTrigger = 10 * time.Second
|
||
seckillMaxDuration = 75 * time.Second
|
||
)
|
||
|
||
type roundOutcome struct {
|
||
successAccounts map[string]bool
|
||
stoppedAccounts map[string]string
|
||
stoppedProducts map[string]string
|
||
attempts int
|
||
}
|
||
|
||
// Scheduler 抢购调度器。
|
||
type Scheduler struct {
|
||
client *client.DouyuClient
|
||
paths types.AppPaths
|
||
font *truetype.Font
|
||
|
||
concurrency int // 抢购并发数
|
||
attempts int // 单号重试次数
|
||
|
||
retryStages []types.RetryStage // 重试间隔阶段配置
|
||
seckillMode types.SeckillMode // 抢购模式
|
||
|
||
accountsMu sync.RWMutex
|
||
accounts []types.Account
|
||
|
||
resultsMu sync.RWMutex
|
||
results map[string]string
|
||
|
||
runMu sync.Mutex
|
||
running bool
|
||
|
||
scheduleMu sync.Mutex
|
||
schedule *time.Timer
|
||
warmupSchedule *time.Timer
|
||
resyncSchedule *time.Timer
|
||
|
||
warmupsMu sync.RWMutex
|
||
warmups map[string]types.WarmupData
|
||
|
||
attemptCacheMu sync.RWMutex
|
||
attemptCache map[string]*types.AttemptCache // 激进模式缓存
|
||
}
|
||
|
||
// New 创建抢购调度器。
|
||
func New(cli *client.DouyuClient, paths types.AppPaths, font *truetype.Font, concurrency, attempts int, retryStages []types.RetryStage, seckillMode types.SeckillMode) *Scheduler {
|
||
if concurrency <= 0 {
|
||
concurrency = constants.DefaultConcurrency
|
||
}
|
||
if attempts <= 0 {
|
||
attempts = constants.DefaultAttempts
|
||
}
|
||
return &Scheduler{
|
||
client: cli,
|
||
paths: paths,
|
||
font: font,
|
||
concurrency: concurrency,
|
||
attempts: attempts,
|
||
retryStages: retryStages,
|
||
seckillMode: seckillMode,
|
||
results: make(map[string]string),
|
||
attemptCache: make(map[string]*types.AttemptCache),
|
||
}
|
||
}
|
||
|
||
// UpdateSettings 运行时更新并发数、重试次数、重试间隔和抢购模式。
|
||
func (s *Scheduler) UpdateSettings(concurrency, attempts int, retryStages []types.RetryStage, seckillMode types.SeckillMode) {
|
||
if concurrency > 0 {
|
||
s.concurrency = concurrency
|
||
}
|
||
if attempts > 0 {
|
||
s.attempts = attempts
|
||
}
|
||
if len(retryStages) > 0 {
|
||
s.retryStages = retryStages
|
||
}
|
||
s.seckillMode = seckillMode
|
||
}
|
||
|
||
// LoadAccounts 从账号文件加载账号。
|
||
func (s *Scheduler) LoadAccounts() {
|
||
path := s.paths.AccountFile
|
||
if path == "" {
|
||
path = constants.DefaultAccountFile
|
||
}
|
||
|
||
// 文件不存在时自动创建目录和空文件，方便用户直接粘贴 Cookie
|
||
if _, err := os.Stat(path); os.IsNotExist(err) {
|
||
if dir := filepath.Dir(path); dir != "" && dir != "." {
|
||
if mkErr := os.MkdirAll(dir, 0755); mkErr != nil {
|
||
logger.LogEvent("账号", "warn", "账号目录创建失败: %s: %v", dir, mkErr)
|
||
}
|
||
}
|
||
// 使用 Windows 换行格式 \r\n
|
||
hint := "# 格式: 账号名----密码----Cookie\r\n# 每行一个账号，粘贴后保存即可\r\n"
|
||
if wrErr := os.WriteFile(path, []byte(hint), 0644); wrErr != nil {
|
||
logger.LogEvent("账号", "warn", "账号文件创建失败: %s: %v", path, wrErr)
|
||
} else {
|
||
logger.LogEvent("账号", "info", "已创建空账号文件: %s，请粘贴 Cookie 后重新加载", path)
|
||
}
|
||
}
|
||
|
||
// 读取文件内容，自动转换换行格式为 Windows 格式
|
||
content, err := os.ReadFile(path)
|
||
if err != nil {
|
||
logger.LogEvent("账号", "error", "读取账号文件失败: %s: %v", path, err)
|
||
s.accountsMu.Lock()
|
||
s.accounts = nil
|
||
s.accountsMu.Unlock()
|
||
return
|
||
}
|
||
|
||
// 检测并转换换行格式
|
||
normalized, converted := s.normalizeLineEndings(path, content)
|
||
if converted {
|
||
// 保存转换后的内容
|
||
if err := os.WriteFile(path, normalized, 0644); err != nil {
|
||
logger.LogEvent("账号", "warn", "账号文件换行格式转换失败: %s: %v", path, err)
|
||
} else {
|
||
logger.LogEvent("账号", "info", "账号文件已自动转换为 Windows 换行格式: %s", path)
|
||
}
|
||
content = normalized
|
||
}
|
||
|
||
var list []types.Account
|
||
scanner := bufio.NewScanner(bytes.NewReader(content))
|
||
scanner.Buffer(make([]byte, 1024), 1024*1024)
|
||
for scanner.Scan() {
|
||
line := strings.TrimSpace(scanner.Text())
|
||
if line == "" || strings.HasPrefix(line, "#") {
|
||
continue
|
||
}
|
||
parts := strings.SplitN(line, "----", 3)
|
||
if len(parts) != 3 {
|
||
logger.LogEvent("账号", "warn", "账号行格式错误，已跳过: %s", line)
|
||
continue
|
||
}
|
||
list = append(list, types.Account{
|
||
Name: strings.TrimSpace(parts[0]),
|
||
Password: strings.TrimSpace(parts[1]),
|
||
Cookie: strings.TrimSpace(parts[2]),
|
||
})
|
||
}
|
||
if err := scanner.Err(); err != nil {
|
||
logger.LogEvent("账号", "error", "读取账号文件中断: %v", err)
|
||
}
|
||
|
||
s.accountsMu.Lock()
|
||
s.accounts = list
|
||
s.accountsMu.Unlock()
|
||
logger.LogEvent("账号", "success", "账号加载完成: %d 个，来源=%s", len(list), path)
|
||
}
|
||
|
||
// AccountCount 返回当前账号数。
|
||
func (s *Scheduler) AccountCount() int {
|
||
s.accountsMu.RLock()
|
||
defer s.accountsMu.RUnlock()
|
||
return len(s.accounts)
|
||
}
|
||
|
||
// normalizeLineEndings 检测并转换换行格式为 Windows 格式（\r\n）。
|
||
// 返回转换后的内容和是否进行了转换。
|
||
func (s *Scheduler) normalizeLineEndings(path string, content []byte) ([]byte, bool) {
|
||
// 检查是否已经是 Windows 格式（包含 \r\n）
|
||
if bytes.Contains(content, []byte("\r\n")) {
|
||
// 已经是 Windows 格式，检查是否还有单独的 \n
|
||
// 先将所有 \r\n 替换为占位符，检查是否还有单独的 \n
|
||
temp := bytes.ReplaceAll(content, []byte("\r\n"), []byte("__RN_PLACEHOLDER__"))
|
||
if !bytes.Contains(temp, []byte("\n")) && !bytes.Contains(temp, []byte("\r")) {
|
||
// 完全是 Windows 格式，不需要转换
|
||
return content, false
|
||
}
|
||
// 混合格式，需要统一转换
|
||
}
|
||
|
||
// 转换逻辑：先统一为 \n，再转换为 \r\n
|
||
normalized := content
|
||
// 1. 将 \r\n 替换为 \n
|
||
normalized = bytes.ReplaceAll(normalized, []byte("\r\n"), []byte("\n"))
|
||
// 2. 将单独的 \r 替换为 \n
|
||
normalized = bytes.ReplaceAll(normalized, []byte("\r"), []byte("\n"))
|
||
// 3. 将 \n 替换为 \r\n
|
||
normalized = bytes.ReplaceAll(normalized, []byte("\n"), []byte("\r\n"))
|
||
|
||
return normalized, true
|
||
}
|
||
|
||
// AccountFilePath 返回账号文件路径。
|
||
func (s *Scheduler) AccountFilePath() string {
|
||
return s.paths.AccountFile
|
||
}
|
||
|
||
// AccountSnapshot 复制一份账号列表。
|
||
func (s *Scheduler) AccountSnapshot() []types.Account {
|
||
s.accountsMu.RLock()
|
||
defer s.accountsMu.RUnlock()
|
||
list := make([]types.Account, len(s.accounts))
|
||
copy(list, s.accounts)
|
||
return list
|
||
}
|
||
|
||
// AccountCookieByName 按账号名查找 Cookie。
|
||
func (s *Scheduler) AccountCookieByName(name string) (string, bool) {
|
||
s.accountsMu.RLock()
|
||
defer s.accountsMu.RUnlock()
|
||
for _, account := range s.accounts {
|
||
if account.Name == name {
|
||
return account.Cookie, true
|
||
}
|
||
}
|
||
return "", false
|
||
}
|
||
|
||
// Results 返回当前抢购结果快照。
|
||
func (s *Scheduler) Results() map[string]string {
|
||
s.resultsMu.RLock()
|
||
defer s.resultsMu.RUnlock()
|
||
out := make(map[string]string, len(s.results))
|
||
for k, v := range s.results {
|
||
out[k] = v
|
||
}
|
||
return out
|
||
}
|
||
|
||
// ClearResults 清空抢购结果。
|
||
func (s *Scheduler) ClearResults() {
|
||
s.resultsMu.Lock()
|
||
defer s.resultsMu.Unlock()
|
||
s.results = make(map[string]string)
|
||
}
|
||
|
||
func (s *Scheduler) setResult(key, value string) {
|
||
s.resultsMu.Lock()
|
||
defer s.resultsMu.Unlock()
|
||
s.results[key] = value
|
||
}
|
||
|
||
// TryStartRun 尝试开始抢购，返回 false 表示已有任务在运行。
|
||
func (s *Scheduler) TryStartRun() bool {
|
||
s.runMu.Lock()
|
||
defer s.runMu.Unlock()
|
||
if s.running {
|
||
return false
|
||
}
|
||
s.running = true
|
||
return true
|
||
}
|
||
|
||
func (s *Scheduler) finishRun() {
|
||
s.runMu.Lock()
|
||
defer s.runMu.Unlock()
|
||
s.running = false
|
||
}
|
||
|
||
// RunNow 立即执行抢购任务，持续循环直到超时或全部成功。
|
||
func (s *Scheduler) RunNow(accountProducts map[string]string) {
|
||
defer s.finishRun()
|
||
s.ClearResults()
|
||
|
||
allAttempts := s.buildAttempts(accountProducts)
|
||
if len(allAttempts) == 0 {
|
||
logger.LogEvent("抢购", "warn", "没有可执行的抢购任务")
|
||
return
|
||
}
|
||
|
||
successAccounts := make(map[string]bool)
|
||
stoppedAccounts := make(map[string]string)
|
||
stoppedProducts := make(map[string]string)
|
||
totalAccounts := uniqueAttemptAccounts(allAttempts)
|
||
totalAttempts := 0
|
||
startTime := time.Now()
|
||
round := 0
|
||
|
||
logger.LogEvent("抢购", "info", "开始抢购: %d 个账号，窗口=%v，前段高频后段降速", totalAccounts, seckillMaxDuration)
|
||
|
||
for {
|
||
elapsed := time.Since(startTime)
|
||
if elapsed >= seckillMaxDuration {
|
||
logger.LogEvent("抢购", "info", "已达到抢购窗口 %v，停止抢购", seckillMaxDuration)
|
||
break
|
||
}
|
||
|
||
round++
|
||
remaining := seckillMaxDuration - elapsed
|
||
|
||
pendingAttempts := s.filterPendingAttempts(allAttempts, stoppedAccounts, stoppedProducts)
|
||
if len(pendingAttempts) == 0 {
|
||
logger.LogEvent("抢购", "info", "所有账号都已成功或遇到终止条件，停止抢购")
|
||
break
|
||
}
|
||
|
||
roundAttempts := s.pickRoundAttempts(pendingAttempts, round)
|
||
if len(roundAttempts) == 0 {
|
||
logger.LogEvent("抢购", "info", "第%d轮无可执行账号，停止抢购", round)
|
||
break
|
||
}
|
||
|
||
// 激进模式每轮开始时清空缓存，强制重新获取 CSRF
|
||
if s.seckillMode == types.SeckillModeAggressive {
|
||
s.clearAttemptCache()
|
||
// 第1轮时从预热数据初始化缓存
|
||
if round == 1 {
|
||
s.initCacheFromWarmup(roundAttempts)
|
||
}
|
||
}
|
||
|
||
logger.LogEvent("抢购", "info", "第%d轮: 待抢购 %d 个账号，本轮执行=%d，剩余时间=%v，模式=%s",
|
||
round, uniqueAttemptAccounts(pendingAttempts), len(roundAttempts), utils.RoundDuration(remaining),
|
||
s.getModeName())
|
||
|
||
outcome := s.executeRound(roundAttempts)
|
||
totalAttempts += outcome.attempts
|
||
|
||
for account := range outcome.successAccounts {
|
||
successAccounts[account] = true
|
||
stoppedAccounts[account] = "已抢到"
|
||
}
|
||
for account, reason := range outcome.stoppedAccounts {
|
||
if stoppedAccounts[account] == "" {
|
||
stoppedAccounts[account] = reason
|
||
}
|
||
}
|
||
for product, reason := range outcome.stoppedProducts {
|
||
if stoppedProducts[product] == "" {
|
||
stoppedProducts[product] = reason
|
||
logger.LogEvent("抢购", "warn", "商品停止重试: %s，原因=%s", product, reason)
|
||
}
|
||
}
|
||
|
||
if len(outcome.successAccounts) > 0 {
|
||
logger.LogEvent("抢购", "success", "第%d轮成功: %d 个账号", round, len(outcome.successAccounts))
|
||
}
|
||
if len(stoppedAccounts) >= totalAccounts {
|
||
logger.LogEvent("抢购", "info", "所有账号都已成功或停止重试，停止抢购")
|
||
break
|
||
}
|
||
|
||
retryInterval := s.getRetryInterval(time.Since(startTime))
|
||
if retryInterval > 0 && time.Since(startTime)+retryInterval < seckillMaxDuration {
|
||
time.Sleep(jitterDuration(retryInterval))
|
||
}
|
||
}
|
||
|
||
logger.LogEvent("抢购", "info", "抢购任务结束: 成功=%d/%d，停止=%d，账号尝试=%d，耗时=%v",
|
||
len(successAccounts), totalAccounts, len(stoppedAccounts)-len(successAccounts), totalAttempts, utils.RoundDuration(time.Since(startTime)))
|
||
}
|
||
|
||
// filterPendingAttempts 过滤掉已终止账号和已确认无库存的商品。
|
||
func (s *Scheduler) filterPendingAttempts(allAttempts []types.ExchangeAttempt, stoppedAccounts, stoppedProducts map[string]string) []types.ExchangeAttempt {
|
||
var pending []types.ExchangeAttempt
|
||
for _, attempt := range allAttempts {
|
||
if stoppedAccounts[attempt.Account.Name] != "" {
|
||
continue
|
||
}
|
||
if stoppedProducts[attempt.Product.CommodityID] != "" {
|
||
continue
|
||
}
|
||
pending = append(pending, attempt)
|
||
}
|
||
return pending
|
||
}
|
||
|
||
// pickRoundAttempts 每轮每个账号最多执行一次，避免同账号并发过猛触发频控。
|
||
func (s *Scheduler) pickRoundAttempts(attempts []types.ExchangeAttempt, round int) []types.ExchangeAttempt {
|
||
if len(attempts) == 0 {
|
||
return nil
|
||
}
|
||
preferredIndex := 1
|
||
if s.attempts > 0 {
|
||
preferredIndex = ((round - 1) % s.attempts) + 1
|
||
}
|
||
|
||
byAccount := make(map[string]types.ExchangeAttempt)
|
||
fallback := make(map[string]types.ExchangeAttempt)
|
||
order := make([]string, 0, len(attempts))
|
||
for _, attempt := range attempts {
|
||
name := attempt.Account.Name
|
||
if _, ok := fallback[name]; !ok {
|
||
fallback[name] = attempt
|
||
order = append(order, name)
|
||
}
|
||
if attempt.Index == preferredIndex {
|
||
byAccount[name] = attempt
|
||
}
|
||
}
|
||
|
||
out := make([]types.ExchangeAttempt, 0, len(order))
|
||
for _, name := range order {
|
||
if attempt, ok := byAccount[name]; ok {
|
||
out = append(out, attempt)
|
||
} else {
|
||
out = append(out, fallback[name])
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// executeRound 执行一轮抢购，并返回成功和终止重试的账号。
|
||
func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt) roundOutcome {
|
||
outcome := roundOutcome{
|
||
successAccounts: make(map[string]bool),
|
||
stoppedAccounts: make(map[string]string),
|
||
stoppedProducts: make(map[string]string),
|
||
}
|
||
var mu sync.Mutex
|
||
|
||
limit := make(chan struct{}, s.concurrency)
|
||
var wg sync.WaitGroup
|
||
|
||
for _, attempt := range attempts {
|
||
wg.Add(1)
|
||
go func(attempt types.ExchangeAttempt) {
|
||
defer wg.Done()
|
||
limit <- struct{}{}
|
||
defer func() { <-limit }()
|
||
|
||
// 根据模式选择执行方法
|
||
var result types.ExchangeResult
|
||
if s.seckillMode == types.SeckillModeAggressive {
|
||
result = s.runAttemptFast(attempt)
|
||
} else {
|
||
result = s.runAttempt(attempt)
|
||
}
|
||
|
||
key := fmt.Sprintf("%s_第%d包", result.Account, result.Attempt)
|
||
s.setResult(key, result.Message)
|
||
|
||
mu.Lock()
|
||
outcome.attempts++
|
||
mu.Unlock()
|
||
|
||
if result.OK {
|
||
logger.LogEvent("抢购", "success", "成功: %s %s 第%d包 %s", result.Account, result.Product, result.Attempt, result.Message)
|
||
go image.SaveSuccessImage(attempt.Account, attempt.Product, result.RoleName, s.font, s.paths.SuccessDir)
|
||
|
||
mu.Lock()
|
||
outcome.successAccounts[attempt.Account.Name] = true
|
||
mu.Unlock()
|
||
} else {
|
||
logger.LogEvent("抢购", "info", "尝试: %s %s 第%d包 %s", result.Account, result.Product, result.Attempt, result.Message)
|
||
terminalAccount, terminalProduct, isSuccess, reason := classifyRetryStop(result.Message)
|
||
if terminalAccount || terminalProduct {
|
||
mu.Lock()
|
||
if terminalAccount {
|
||
outcome.stoppedAccounts[attempt.Account.Name] = reason
|
||
}
|
||
if terminalProduct {
|
||
outcome.stoppedProducts[attempt.Product.CommodityID] = reason
|
||
}
|
||
// 兑换上限也算成功
|
||
if isSuccess {
|
||
outcome.successAccounts[attempt.Account.Name] = true
|
||
logger.LogEvent("抢购", "success", "成功（兑换上限）: %s %s 第%d包", result.Account, result.Product, result.Attempt)
|
||
go image.SaveSuccessImage(attempt.Account, attempt.Product, result.RoleName, s.font, s.paths.SuccessDir)
|
||
}
|
||
mu.Unlock()
|
||
}
|
||
}
|
||
}(attempt)
|
||
}
|
||
|
||
wg.Wait()
|
||
return outcome
|
||
}
|
||
|
||
// getModeName 返回当前抢购模式的名称。
|
||
func (s *Scheduler) getModeName() string {
|
||
if s.seckillMode == types.SeckillModeAggressive {
|
||
return "激进"
|
||
}
|
||
return "标准"
|
||
}
|
||
|
||
// getRetryInterval 根据配置的重试阶段计算当前应该使用的重试间隔。
|
||
func (s *Scheduler) getRetryInterval(elapsed time.Duration) time.Duration {
|
||
elapsedSec := int(elapsed.Seconds())
|
||
|
||
// 使用配置的阶段
|
||
for _, stage := range s.retryStages {
|
||
if stage.MaxElapsed > 0 && elapsedSec < stage.MaxElapsed && stage.Interval > 0 {
|
||
return time.Duration(stage.Interval) * time.Millisecond
|
||
}
|
||
}
|
||
|
||
// 默认值（如果没有配置或配置无效）
|
||
return 900 * time.Millisecond
|
||
}
|
||
|
||
func jitterDuration(base time.Duration) time.Duration {
|
||
if base <= 0 {
|
||
return 0
|
||
}
|
||
// 加一点抖动，避免所有机器严格同频请求。
|
||
jitter := time.Duration(rand.Int63n(int64(base / 3)))
|
||
return base + jitter
|
||
}
|
||
|
||
func classifyRetryStop(message string) (account bool, product bool, success bool, reason string) {
|
||
switch {
|
||
case strings.Contains(message, "商品本次活动库存已耗完"):
|
||
// 库存耗尽不停止，继续重试（斗鱼可能会补货）
|
||
return false, false, false, ""
|
||
case strings.Contains(message, "您已达本次活动兑换上限"):
|
||
return true, false, true, "已抢到（兑换上限）"
|
||
case strings.Contains(message, "未绑定账号"):
|
||
return true, false, false, "账号未绑定角色"
|
||
case strings.Contains(message, "需要开通精英手册"):
|
||
return true, false, false, "账号不满足精英手册条件"
|
||
default:
|
||
return false, false, false, ""
|
||
}
|
||
}
|
||
|
||
func uniqueAttemptAccounts(attempts []types.ExchangeAttempt) int {
|
||
seen := make(map[string]struct{}, len(attempts))
|
||
for _, attempt := range attempts {
|
||
seen[attempt.Account.Name] = struct{}{}
|
||
}
|
||
return len(seen)
|
||
}
|
||
|
||
func (s *Scheduler) buildAttempts(accountProducts map[string]string) []types.ExchangeAttempt {
|
||
accounts := s.AccountSnapshot()
|
||
var attempts []types.ExchangeAttempt
|
||
for _, account := range accounts {
|
||
commodityID := accountProducts[account.Name]
|
||
if commodityID == "" {
|
||
continue
|
||
}
|
||
product := types.Product{CommodityID: commodityID, CommodityName: commodityID}
|
||
for i := 1; i <= s.attempts; i++ {
|
||
attempts = append(attempts, types.ExchangeAttempt{
|
||
Account: account,
|
||
Product: product,
|
||
Index: i,
|
||
})
|
||
}
|
||
}
|
||
return attempts
|
||
}
|
||
|
||
func warmupKey(attempt types.ExchangeAttempt) string {
|
||
return fmt.Sprintf("%s|%s|%d", attempt.Account.Name, attempt.Product.CommodityID, attempt.Index)
|
||
}
|
||
|
||
func (s *Scheduler) setWarmup(key string, data types.WarmupData) {
|
||
s.warmupsMu.Lock()
|
||
defer s.warmupsMu.Unlock()
|
||
if s.warmups == nil {
|
||
s.warmups = make(map[string]types.WarmupData)
|
||
}
|
||
s.warmups[key] = data
|
||
}
|
||
|
||
func (s *Scheduler) warmupFor(attempt types.ExchangeAttempt) (types.WarmupData, bool) {
|
||
s.warmupsMu.RLock()
|
||
data, ok := s.warmups[warmupKey(attempt)]
|
||
s.warmupsMu.RUnlock()
|
||
if !ok || data.CreatedAt.IsZero() || time.Since(data.CreatedAt) > warmupMaxAge {
|
||
return types.WarmupData{}, false
|
||
}
|
||
return data, true
|
||
}
|
||
|
||
func (s *Scheduler) clearWarmups() {
|
||
s.warmupsMu.Lock()
|
||
defer s.warmupsMu.Unlock()
|
||
s.warmups = make(map[string]types.WarmupData)
|
||
}
|
||
|
||
// ========== 激进模式缓存管理 ==========
|
||
|
||
func (s *Scheduler) getAttemptCache(accountName string) *types.AttemptCache {
|
||
s.attemptCacheMu.RLock()
|
||
defer s.attemptCacheMu.RUnlock()
|
||
return s.attemptCache[accountName]
|
||
}
|
||
|
||
// initCacheFromWarmup 从预热数据初始化激进模式缓存。
|
||
func (s *Scheduler) initCacheFromWarmup(attempts []types.ExchangeAttempt) {
|
||
for _, attempt := range attempts {
|
||
if warmup, ok := s.warmupFor(attempt); ok {
|
||
s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{
|
||
CSRF: warmup.CSRF,
|
||
RoleName: warmup.RoleName,
|
||
LastCheck: time.Now(),
|
||
})
|
||
logger.LogEvent("抢购", "info", "从预热初始化缓存: 账号=%s", attempt.Account.Name)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Scheduler) setAttemptCache(accountName string, cache *types.AttemptCache) {
|
||
s.attemptCacheMu.Lock()
|
||
defer s.attemptCacheMu.Unlock()
|
||
s.attemptCache[accountName] = cache
|
||
}
|
||
|
||
func (s *Scheduler) clearAttemptCache() {
|
||
s.attemptCacheMu.Lock()
|
||
defer s.attemptCacheMu.Unlock()
|
||
s.attemptCache = make(map[string]*types.AttemptCache)
|
||
}
|
||
|
||
func (s *Scheduler) getCacheCSRF(accountName string) string {
|
||
if cache := s.getAttemptCache(accountName); cache != nil && time.Since(cache.LastCheck) < 30*time.Second {
|
||
return cache.CSRF
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func (s *Scheduler) warmupAttempts(attempts []types.ExchangeAttempt) {
|
||
if len(attempts) == 0 {
|
||
return
|
||
}
|
||
s.clearWarmups()
|
||
logger.LogEvent("预热", "info", "开始预热: %d 个请求", len(attempts))
|
||
|
||
limit := make(chan struct{}, s.concurrency)
|
||
var wg sync.WaitGroup
|
||
for _, attempt := range attempts {
|
||
wg.Add(1)
|
||
go func(attempt types.ExchangeAttempt) {
|
||
defer wg.Done()
|
||
limit <- struct{}{}
|
||
defer func() { <-limit }()
|
||
|
||
start := time.Now()
|
||
csrf, err := s.client.GenerateCSRF(attempt.Account.Cookie)
|
||
if err != nil {
|
||
logger.LogEvent("预热", "warn", "CSRF预热失败: 账号=%s 商品=%s err=%v", attempt.Account.Name, attempt.Product.CommodityName, err)
|
||
return
|
||
}
|
||
roleName, err := s.client.RoleName(attempt.Account.Cookie)
|
||
if err != nil {
|
||
logger.LogEvent("预热", "warn", "角色预热失败: 账号=%s err=%v", attempt.Account.Name, err)
|
||
}
|
||
s.setWarmup(warmupKey(attempt), types.WarmupData{
|
||
CSRF: csrf,
|
||
RoleName: roleName,
|
||
CreatedAt: time.Now(),
|
||
})
|
||
logger.LogEvent("预热", "success", "预热完成: 账号=%s 商品=%s 耗时=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(start)))
|
||
}(attempt)
|
||
}
|
||
wg.Wait()
|
||
logger.LogEvent("预热", "info", "预热任务结束")
|
||
}
|
||
|
||
func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration {
|
||
calibration, err := s.client.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait)
|
||
if err != nil {
|
||
logger.LogEvent("时间", "warn", "%s校准降级: %v", label, err)
|
||
}
|
||
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(),
|
||
calibration.ErrorBound.Milliseconds(),
|
||
calibration.RTT.Milliseconds(),
|
||
calibration.Samples,
|
||
calibration.Source)
|
||
return calibration
|
||
}
|
||
|
||
func (s *Scheduler) replaceTimerLocked(timer **time.Timer, d time.Duration, fn func()) {
|
||
if *timer != nil {
|
||
(*timer).Stop()
|
||
}
|
||
*timer = time.AfterFunc(d, fn)
|
||
}
|
||
|
||
func (s *Scheduler) stopSchedulesLocked() {
|
||
if s.schedule != nil {
|
||
s.schedule.Stop()
|
||
s.schedule = nil
|
||
}
|
||
if s.warmupSchedule != nil {
|
||
s.warmupSchedule.Stop()
|
||
s.warmupSchedule = nil
|
||
}
|
||
if s.resyncSchedule != nil {
|
||
s.resyncSchedule.Stop()
|
||
s.resyncSchedule = nil
|
||
}
|
||
}
|
||
|
||
func (s *Scheduler) runAttempt(attempt types.ExchangeAttempt) types.ExchangeResult {
|
||
result := types.ExchangeResult{
|
||
Account: attempt.Account.Name,
|
||
Product: attempt.Product.CommodityName,
|
||
Attempt: attempt.Index,
|
||
}
|
||
|
||
logger.LogEvent("抢购", "info", "执行兑换链路: 账号=%s 商品=%s 第%d包", attempt.Account.Name, attempt.Product.CommodityName, attempt.Index)
|
||
var roleName string
|
||
csrf := ""
|
||
if warmup, ok := s.warmupFor(attempt); ok {
|
||
csrf = warmup.CSRF
|
||
roleName = warmup.RoleName
|
||
logger.LogEvent("抢购", "info", "使用预热CSRF: 账号=%s 商品=%s 预热距今=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(warmup.CreatedAt)))
|
||
}
|
||
if csrf == "" {
|
||
var err error
|
||
csrf, err = s.client.GenerateCSRF(attempt.Account.Cookie)
|
||
if err != nil {
|
||
result.Message = "失败|生成CSRF失败: " + err.Error()
|
||
return result
|
||
}
|
||
logger.LogEvent("抢购", "info", "CSRF已生成，本次兑换链路复用同一个 token: 账号=%s 商品=%s", attempt.Account.Name, attempt.Product.CommodityName)
|
||
}
|
||
|
||
limit, err := s.client.BatchExchangeLimit(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf)
|
||
if err != nil {
|
||
result.Message = "失败|批量限制检查失败: " + err.Error()
|
||
return result
|
||
}
|
||
if limit <= 0 {
|
||
result.Message = "失败|当前商品限制兑换"
|
||
return result
|
||
}
|
||
logger.LogEvent("抢购", "info", "批量限制通过: 账号=%s 商品=%s limit=%d", attempt.Account.Name, attempt.Product.CommodityName, limit)
|
||
|
||
if err := s.client.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf); err != nil {
|
||
result.Message = "失败|预兑换失败: " + err.Error()
|
||
return result
|
||
}
|
||
|
||
exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf)
|
||
if err != nil {
|
||
result.Message = "失败|" + err.Error()
|
||
return result
|
||
}
|
||
|
||
if roleName == "" {
|
||
roleName, err = s.client.RoleName(attempt.Account.Cookie)
|
||
if err != nil {
|
||
logger.LogEvent("抢购", "warn", "%s 角色名查询失败: %v", attempt.Account.Name, err)
|
||
}
|
||
}
|
||
|
||
result.OK = true
|
||
result.RoleName = roleName
|
||
if roleName != "" {
|
||
result.Message = "成功|" + roleName + "|" + exchangeInfo
|
||
} else {
|
||
result.Message = "成功|" + exchangeInfo
|
||
}
|
||
|
||
// 缓存 CSRF 和角色名（激进模式使用）
|
||
s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{
|
||
CSRF: csrf,
|
||
RoleName: roleName,
|
||
LastCheck: time.Now(),
|
||
})
|
||
|
||
return result
|
||
}
|
||
|
||
// runAttemptFast 激进模式的兑换逻辑，复用缓存的 CSRF，跳过 BatchExchangeLimit 和 PreExchange。
|
||
func (s *Scheduler) runAttemptFast(attempt types.ExchangeAttempt) types.ExchangeResult {
|
||
result := types.ExchangeResult{
|
||
Account: attempt.Account.Name,
|
||
Product: attempt.Product.CommodityName,
|
||
Attempt: attempt.Index,
|
||
}
|
||
|
||
// 获取缓存的 CSRF
|
||
csrf := s.getCacheCSRF(attempt.Account.Name)
|
||
if csrf == "" {
|
||
// 没有缓存，回退到标准模式
|
||
logger.LogEvent("抢购", "info", "激进模式回退标准: 账号=%s 原因=无缓存CSRF", attempt.Account.Name)
|
||
return s.runAttempt(attempt)
|
||
}
|
||
|
||
logger.LogEvent("抢购", "info", "激进模式兑换: 账号=%s 商品=%s 第%d包", attempt.Account.Name, attempt.Product.CommodityName, attempt.Index)
|
||
|
||
// 直接调用 Exchange
|
||
exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf)
|
||
if err != nil {
|
||
errMsg := err.Error()
|
||
// 检查是否是 CSRF 相关错误，如果是则回退到标准模式
|
||
if strings.Contains(errMsg, "csrf") || strings.Contains(errMsg, "CSRF") || strings.Contains(errMsg, "token") {
|
||
logger.LogEvent("抢购", "warn", "激进模式回退标准: 账号=%s 原因=CSRF错误", attempt.Account.Name)
|
||
s.setAttemptCache(attempt.Account.Name, nil) // 清空缓存
|
||
return s.runAttempt(attempt)
|
||
}
|
||
result.Message = "失败|" + errMsg
|
||
return result
|
||
}
|
||
|
||
// 获取角色名
|
||
cache := s.getAttemptCache(attempt.Account.Name)
|
||
roleName := ""
|
||
if cache != nil {
|
||
roleName = cache.RoleName
|
||
}
|
||
if roleName == "" {
|
||
var err error
|
||
roleName, err = s.client.RoleName(attempt.Account.Cookie)
|
||
if err != nil {
|
||
logger.LogEvent("抢购", "warn", "%s 角色名查询失败: %v", attempt.Account.Name, err)
|
||
}
|
||
// 更新缓存
|
||
if cache != nil {
|
||
cache.RoleName = roleName
|
||
}
|
||
}
|
||
|
||
result.OK = true
|
||
result.RoleName = roleName
|
||
if roleName != "" {
|
||
result.Message = "成功|" + roleName + "|" + exchangeInfo
|
||
} else {
|
||
result.Message = "成功|" + exchangeInfo
|
||
}
|
||
return result
|
||
}
|
||
|
||
// ScheduleRun 注册一次定时抢购。
|
||
func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error) {
|
||
if len(req.AccountProducts) == 0 {
|
||
return nil, errors.New("请选择账号和商品")
|
||
}
|
||
if strings.TrimSpace(req.TargetTime) == "" {
|
||
return nil, errors.New("请设置目标时间")
|
||
}
|
||
|
||
targetLocal, err := time.ParseInLocation("2006-01-02 15:04:05", req.TargetTime, constants.CST)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("时间格式错误: %w", err)
|
||
}
|
||
targetTime := targetLocal.UTC()
|
||
|
||
calibration := s.calibrateDouyuTime("定时")
|
||
serverTime := calibration.EstimatedServerNow()
|
||
|
||
advance := time.Duration(req.Advance) * time.Millisecond
|
||
triggerTime := targetTime.Add(-advance)
|
||
remain := triggerTime.Sub(serverTime)
|
||
if remain <= 0 {
|
||
return nil, errors.New("目标时间已过")
|
||
}
|
||
|
||
ap := utils.CloneStringMap(req.AccountProducts)
|
||
attempts := s.buildAttempts(ap)
|
||
s.clearWarmups()
|
||
s.scheduleMu.Lock()
|
||
s.stopSchedulesLocked()
|
||
s.scheduleMu.Unlock()
|
||
|
||
s.scheduleTimers(ap, attempts, targetTime, triggerTime, calibration, false)
|
||
|
||
resyncRemain := triggerTime.Sub(serverTime) - resyncBeforeTrigger
|
||
if resyncRemain < 0 {
|
||
resyncRemain = 0
|
||
}
|
||
|
||
s.scheduleMu.Lock()
|
||
s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() {
|
||
newCalibration := s.calibrateDouyuTime("临近")
|
||
s.scheduleTimers(ap, attempts, targetTime, triggerTime, newCalibration, true)
|
||
})
|
||
s.scheduleMu.Unlock()
|
||
|
||
warmupRemain := remain - warmupBeforeTrigger
|
||
if warmupRemain < 0 {
|
||
warmupRemain = 0
|
||
}
|
||
|
||
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"),
|
||
req.Advance,
|
||
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"),
|
||
time.Now().Add(resyncRemain).In(constants.CST).Format("15:04:05.000"),
|
||
)
|
||
|
||
return map[string]any{
|
||
"success": true,
|
||
"targetTime": targetTime.In(constants.CST).Format("2006-01-02 15:04:05"),
|
||
"triggerTime": triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||
"advance": req.Advance,
|
||
"remainSecond": int(remain.Seconds()),
|
||
"serverTime": serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
|
||
"offsetMs": calibration.Offset.Milliseconds(),
|
||
"errorBoundMs": calibration.ErrorBound.Milliseconds(),
|
||
"rttMs": calibration.RTT.Milliseconds(),
|
||
}, nil
|
||
}
|
||
|
||
func (s *Scheduler) scheduleTimers(accountProducts map[string]string, attempts []types.ExchangeAttempt, targetTime, triggerTime time.Time, calibration types.TimeCalibration, resync bool) {
|
||
serverTime := calibration.EstimatedServerNow()
|
||
remain := triggerTime.Sub(serverTime)
|
||
if remain < 0 {
|
||
remain = 0
|
||
}
|
||
warmupRemain := remain - warmupBeforeTrigger
|
||
if warmupRemain < 0 {
|
||
warmupRemain = 0
|
||
}
|
||
|
||
s.scheduleMu.Lock()
|
||
defer s.scheduleMu.Unlock()
|
||
s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() {
|
||
logger.LogEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(constants.CST).Format("15:04:05.000"))
|
||
s.warmupAttempts(attempts)
|
||
})
|
||
s.replaceTimerLocked(&s.schedule, remain, func() {
|
||
now := time.Now()
|
||
estimatedServer := now.Add(calibration.Offset)
|
||
logger.LogEvent("定时", "success", "定时触发: 本地=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms",
|
||
now.In(constants.CST).Format("15:04:05.000"),
|
||
estimatedServer.In(constants.CST).Format("15:04:05.000"),
|
||
calibration.Offset.Milliseconds(),
|
||
calibration.ErrorBound.Milliseconds())
|
||
if !s.TryStartRun() {
|
||
logger.LogEvent("抢购", "warn", "已有抢购任务正在运行，本次定时触发已跳过")
|
||
return
|
||
}
|
||
s.RunNow(accountProducts)
|
||
})
|
||
if resync {
|
||
logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%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"),
|
||
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"))
|
||
}
|
||
}
|
||
|
||
// CancelSchedule 停止尚未触发的定时任务。
|
||
func (s *Scheduler) CancelSchedule() bool {
|
||
s.scheduleMu.Lock()
|
||
defer s.scheduleMu.Unlock()
|
||
stopped := false
|
||
if s.schedule != nil {
|
||
stopped = s.schedule.Stop()
|
||
}
|
||
s.stopSchedulesLocked()
|
||
s.clearWarmups()
|
||
return stopped
|
||
}
|