diff --git a/app.go b/app.go index 2b6a430..7202134 100644 --- a/app.go +++ b/app.go @@ -1,13 +1,8 @@ package main import ( - "bufio" "context" - "fmt" "os" - "path/filepath" - "strconv" - "strings" "sync" "time" @@ -22,10 +17,10 @@ import ( "douyu_go/internal/seckill" "douyu_go/internal/trace" "douyu_go/internal/types" - "douyu_go/internal/utils" ) // App 应用主结构体，所有导出方法可被前端通过 Wails 绑定调用。 +// 绑定方法按领域拆分到 app_*.go 文件。 type App struct { ctx context.Context @@ -48,8 +43,6 @@ type App struct { closeTrace func() } -const exchangeLookupConcurrency = 8 - // getClient 返回当前 client 指针快照（读锁保护）。 func (a *App) getClient() *client.DouyuClient { a.clientMu.RLock() @@ -88,13 +81,6 @@ func cloneConfig(c types.Config) types.Config { return cp } -type exchangeLookupResult struct { - Account types.Account - Records []types.ExchangeRecord - Err error - EmptyCookie bool -} - // NewApp 创建 App 实例。 func NewApp() *App { return &App{} @@ -158,473 +144,6 @@ func (a *App) shutdown(ctx context.Context) { } } -// ==================== 前端可调用方法 ==================== - -// LoadProducts 刷新商品列表。refresh=true 时从斗鱼实时拉取。 -func (a *App) LoadProducts(refresh bool) types.ProductPayload { - if refresh { - products, err := a.getClient().Products() - if err != nil { - logger.LogEvent("商品", "error", "商品刷新失败: %v", err) - return types.ProductPayload{Success: false} - } - fetchedAt := time.Now() - a.setProducts(products, fetchedAt) - logger.LogEvent("商品", "success", "商品刷新完成: %d 个，刷新时间=%s", len(products), fetchedAt.In(constants.CST).Format("15:04:05.000")) - return a.productPayload("live") - } - return a.productPayload("cache") -} - -// GetAccounts 返回账号列表。 -func (a *App) GetAccounts() []types.AccountView { - accounts := a.scheduler.AccountSnapshot() - list := make([]types.AccountView, 0, len(accounts)) - for i, account := range accounts { - list = append(list, types.AccountView{ - Index: i, - Name: account.Name, - Cookie: utils.MaskCookie(account.Cookie), - }) - } - return list -} - -// ReloadAccounts 重新加载账号文件。 -func (a *App) ReloadAccounts() types.ReloadAccountsResult { - a.scheduler.LoadAccounts() - return types.ReloadAccountsResult{Success: true, Count: a.scheduler.AccountCount()} -} - -// StartSeckill 立即启动抢购。 -func (a *App) StartSeckill(accountProducts map[string]string) types.MessageResult { - if len(accountProducts) == 0 { - return types.MessageResult{Error: "请选择账号和商品"} - } - if !a.scheduler.TryStartRun() { - return types.MessageResult{Error: "抢购任务正在运行，请勿重复提交"} - } - go a.scheduler.RunNow(utils.CloneStringMap(accountProducts)) - return types.MessageResult{Success: true, Msg: "抢购已启动"} -} - -// SetSchedule 设置定时抢购。 -func (a *App) SetSchedule(req types.SeckillRequest) (types.ScheduleResult, error) { - return a.scheduler.ScheduleRun(req) -} - -// CancelSchedule 取消未触发的定时抢购。 -func (a *App) CancelSchedule() types.CancelScheduleResult { - stopped := a.scheduler.CancelSchedule() - return types.CancelScheduleResult{Success: true, Stopped: stopped} -} - -// GetResults 返回当前抢购结果。 -func (a *App) GetResults() map[string]string { - return a.scheduler.Results() -} - -// GetLogs 返回后端内存日志，支持增量获取。 -func (a *App) GetLogs(since int64) types.LogsResult { - entries := a.logStore.Since(since) - nextSince := since - if len(entries) > 0 { - nextSince = entries[len(entries)-1].ID - } - return types.LogsResult{ - Success: true, - Logs: entries, - Since: nextSince, - File: a.logStore.FilePath(), - } -} - -// GetTime 返回服务端时间和本地时间。 -func (a *App) GetTime() types.TimeResult { - serverTime, err := a.getClient().ServerTime() - source := "currentTime" - if err != nil { - source = "localFallback" - logger.LogEvent("时间", "warn", "时间接口降级: %v", err) - } - return types.TimeResult{ - ServerTime: serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), - LocalTime: time.Now().In(constants.CST).Format("2006-01-02 15:04:05.000"), - Source: source, - } -} - -// ClearResults 清空抢购结果。 -func (a *App) ClearResults() types.SimpleResult { - a.scheduler.ClearResults() - return types.SimpleResult{Success: true} -} - -// QueryScore 查询单个账号的积分。 -func (a *App) QueryScore(accountName string) types.ScoreResult { - if accountName == "" { - return types.ScoreResult{Score: "缺少账号名"} - } - cookie, ok := a.scheduler.AccountCookieByName(accountName) - if !ok || cookie == "" { - return types.ScoreResult{Score: fmt.Sprintf("账号不存在或 Cookie 为空: %s", accountName)} - } - score, err := a.getClient().Score(cookie) - if err != nil { - return types.ScoreResult{Score: "查询失败: " + err.Error()} - } - return types.ScoreResult{Score: strconv.Itoa(score)} -} - -// VerifyExchange 查询兑换记录，用于复验抢购结果。 -func (a *App) VerifyExchange(accountName string) types.ExchangeRecordsResult { - if accountName == "" { - return types.ExchangeRecordsResult{Error: "缺少账号名"} - } - cookie, ok := a.scheduler.AccountCookieByName(accountName) - if !ok || cookie == "" { - return types.ExchangeRecordsResult{Error: fmt.Sprintf("账号不存在或 Cookie 为空: %s", accountName)} - } - - records, err := a.getClient().ExchangeList(cookie) - if err != nil { - return types.ExchangeRecordsResult{Error: "查询失败: " + err.Error()} - } - matched := filterExchangeRecords(records) - - return types.ExchangeRecordsResult{ - Success: true, - Records: matched, - Count: len(matched), - } -} - -// VerifyAllExchange 批量查询兑换记录，使用有限并发减少等待时间。 -func (a *App) VerifyAllExchange() types.VerifyAllExchangeResult { - accounts := a.scheduler.AccountSnapshot() - results := make(map[string]types.AccountExchangeResult) - lookups := a.lookupExchangeRecords("复验", accounts) - - for _, lookup := range lookups { - if lookup.EmptyCookie { - results[lookup.Account.Name] = types.AccountExchangeResult{Error: "Cookie 为空"} - continue - } - - if lookup.Err != nil { - results[lookup.Account.Name] = types.AccountExchangeResult{Error: "查询失败: " + lookup.Err.Error()} - continue - } - - matched := filterExchangeRecords(lookup.Records) - results[lookup.Account.Name] = types.AccountExchangeResult{ - Success: true, - Records: matched, - Count: len(matched), - } - } - - return types.VerifyAllExchangeResult{Results: results} -} - -// VerifyAndExport 复验并导出结果。 -func (a *App) VerifyAndExport(commodityID string) types.VerifyAndExportResult { - accounts := a.scheduler.AccountSnapshot() - results := make(map[string]types.AccountHitResult) - hitAccounts := make([]string, 0) - missAccounts := make([]string, 0) - lookups := a.lookupExchangeRecords("复验", accounts) - - for _, lookup := range lookups { - if lookup.EmptyCookie { - missAccounts = append(missAccounts, lookup.Account.Name) - results[lookup.Account.Name] = types.AccountHitResult{Error: "Cookie 为空"} - continue - } - - if lookup.Err != nil { - missAccounts = append(missAccounts, lookup.Account.Name) - results[lookup.Account.Name] = types.AccountHitResult{Error: "查询失败: " + lookup.Err.Error()} - continue - } - - // 检查是否抢到了指定商品 - hit := hasCommodityRecord(lookup.Records, commodityID, 0, 0) - if hit { - hitAccounts = append(hitAccounts, lookup.Account.Name) - results[lookup.Account.Name] = types.AccountHitResult{Success: true, Hit: true} - } else { - missAccounts = append(missAccounts, lookup.Account.Name) - results[lookup.Account.Name] = types.AccountHitResult{Success: true} - } - } - - return types.VerifyAndExportResult{ - Success: true, - HitCount: len(hitAccounts), - MissCount: len(missAccounts), - HitAccounts: hitAccounts, - MissAccounts: missAccounts, - Details: results, - } -} - -func (a *App) lookupExchangeRecords(category string, accounts []types.Account) []exchangeLookupResult { - results := make([]exchangeLookupResult, len(accounts)) - if len(accounts) == 0 { - return results - } - - workerCount := exchangeLookupConcurrency - if len(accounts) < workerCount { - workerCount = len(accounts) - } - start := time.Now() - logger.LogEvent(category, "info", "兑换记录批量查询开始: 账号=%d 并发=%d", len(accounts), workerCount) - - jobs := make(chan int) - var wg sync.WaitGroup - for worker := 0; worker < workerCount; worker++ { - wg.Add(1) - go func() { - defer wg.Done() - for idx := range jobs { - account := accounts[idx] - lookup := exchangeLookupResult{Account: account} - if strings.TrimSpace(account.Cookie) == "" { - lookup.EmptyCookie = true - results[idx] = lookup - continue - } - records, err := a.getClient().ExchangeList(account.Cookie) - lookup.Records = records - lookup.Err = err - results[idx] = lookup - } - }() - } - for idx := range accounts { - jobs <- idx - } - close(jobs) - wg.Wait() - - errors := 0 - empty := 0 - for _, result := range results { - if result.EmptyCookie { - empty++ - } - if result.Err != nil { - errors++ - } - } - level := "success" - if errors > 0 { - level = "warn" - } - logger.LogEvent(category, level, "兑换记录批量查询完成: 账号=%d 空CK=%d 失败=%d 耗时=%s", - len(accounts), empty, errors, time.Since(start).Round(time.Millisecond)) - return results -} - -func filterExchangeRecords(records []types.ExchangeRecord) []types.ExchangeRecord { - matched := make([]types.ExchangeRecord, 0, len(records)) - for _, r := range records { - if r.CommodityID != "" { - matched = append(matched, r) - } - } - return matched -} - -func hasCommodityRecord(records []types.ExchangeRecord, commodityID string, startTime, endTime int64) bool { - if commodityID == "" { - return false - } - for _, r := range records { - if r.CommodityID != commodityID { - continue - } - if startTime == 0 && endTime == 0 { - return true - } - recordTime, err := strconv.ParseInt(r.Time, 10, 64) - if err != nil { - continue - } - if startTime > 0 && recordTime < startTime { - continue - } - if endTime > 0 && recordTime > endTime { - continue - } - return true - } - return false -} - -// accountLine 账号文件的一行解析结果。 -type accountLine struct { - Raw string // 原始行文本 - Name string // 解析出的账号名（注释/空行为空） - Comment bool // 是否为注释或空行 -} - -// parseAccountFile 读取账号文件并按行解析账号名。 -func parseAccountFile(path string) ([]accountLine, error) { - data, err := os.ReadFile(path) - if err != nil { - return nil, err - } - var lines []accountLine - scanner := bufio.NewScanner(strings.NewReader(string(data))) - for scanner.Scan() { - raw := scanner.Text() - trimmed := strings.TrimSpace(raw) - if trimmed == "" || strings.HasPrefix(trimmed, "#") { - lines = append(lines, accountLine{Raw: raw, Comment: true}) - continue - } - name := "" - if parts := strings.SplitN(trimmed, "----", 3); len(parts) >= 1 { - name = strings.TrimSpace(parts[0]) - } - lines = append(lines, accountLine{Raw: raw, Name: name}) - } - return lines, scanner.Err() -} - -// ExportAccounts 导出指定账号到文件。 -func (a *App) ExportAccounts(accountNames []string, filename string) types.ExportAccountsResult { - if len(accountNames) == 0 { - return types.ExportAccountsResult{Error: "没有要导出的账号"} - } - - accountFile := a.scheduler.AccountFilePath() - lines, err := parseAccountFile(accountFile) - if err != nil { - return types.ExportAccountsResult{Error: "读取账号文件失败: " + err.Error()} - } - - nameSet := make(map[string]bool) - for _, name := range accountNames { - nameSet[name] = true - } - - var matchedLines []string - for _, line := range lines { - if line.Comment { - continue - } - if nameSet[line.Name] { - matchedLines = append(matchedLines, line.Raw) - } - } - - if len(matchedLines) == 0 { - return types.ExportAccountsResult{Error: "没有找到匹配的账号行"} - } - - // 生成导出文件路径 - exportPath := filepath.Join(filepath.Dir(accountFile), filename) - content := strings.Join(matchedLines, "\n") + "\n" - - if err := os.WriteFile(exportPath, []byte(content), 0644); err != nil { - return types.ExportAccountsResult{Error: "写入文件失败: " + err.Error()} - } - - logger.LogEvent("导出", "success", "已导出 %d 个账号到: %s", len(matchedLines), exportPath) - - return types.ExportAccountsResult{ - Success: true, - Count: len(matchedLines), - FilePath: exportPath, - } -} - -// ExportAccountsWithProduct 导出指定账号到文件，自动使用中文商品名和时间戳命名。 -func (a *App) ExportAccountsWithProduct(accountNames []string, commodityID, commodityName string) types.ExportAccountsResult { - // 生成文件名: user_已抢到_商品中文名_20260626_150405.txt - now := time.Now() - timestamp := now.Format("20060102_150405") - // 清理商品名中的特殊字符（复用 utils.SafeFilename，避免与 utils 包重复实现） - safeName := utils.SafeFilename(commodityName) - filename := fmt.Sprintf("user_已抢到_%s_%s.txt", safeName, timestamp) - - return a.ExportAccounts(accountNames, filename) -} - -// OrganizeAccounts 整理账号，把已抢到指定商品的账号分离出来。 -func (a *App) OrganizeAccounts(commodityID, commodityName string, startTime, endTime int64) types.OrganizeAccountsResult { - accounts := a.scheduler.AccountSnapshot() - hitAccounts := make([]string, 0) - missAccounts := make([]string, 0) - lookups := a.lookupExchangeRecords("整理", accounts) - - for _, lookup := range lookups { - if lookup.EmptyCookie || lookup.Err != nil { - missAccounts = append(missAccounts, lookup.Account.Name) - continue - } - - if hasCommodityRecord(lookup.Records, commodityID, startTime, endTime) { - hitAccounts = append(hitAccounts, lookup.Account.Name) - } else { - missAccounts = append(missAccounts, lookup.Account.Name) - } - } - - // 导出已抢到的账号 - var exportResult types.ExportAccountsResult - if len(hitAccounts) > 0 { - exportResult = a.ExportAccountsWithProduct(hitAccounts, commodityID, commodityName) - } - - // 更新原文件，只保留未抢到的 - accountFile := a.scheduler.AccountFilePath() - lines, err := parseAccountFile(accountFile) - if err != nil { - return types.OrganizeAccountsResult{Error: "读取账号文件失败: " + err.Error()} - } - - missSet := make(map[string]bool) - for _, name := range missAccounts { - missSet[name] = true - } - - var keepLines []string - for _, line := range lines { - if line.Comment { - keepLines = append(keepLines, line.Raw) - continue - } - if missSet[line.Name] { - keepLines = append(keepLines, line.Raw) - } - } - - // 写回原文件 - newContent := strings.Join(keepLines, "\n") + "\n" - if err := os.WriteFile(accountFile, []byte(newContent), 0644); err != nil { - return types.OrganizeAccountsResult{Error: "更新账号文件失败: " + err.Error()} - } - - // 重新加载账号 - a.scheduler.LoadAccounts() - - logger.LogEvent("整理", "success", "整理完成: 已抢到=%d, 未抢到=%d", len(hitAccounts), len(missAccounts)) - - return types.OrganizeAccountsResult{ - Success: true, - HitCount: len(hitAccounts), - MissCount: len(missAccounts), - HitAccounts: hitAccounts, - MissAccounts: missAccounts, - ExportResult: exportResult, - } -} - // GetVersion 返回应用版本号。 func (a *App) GetVersion() string { if constants.BuildTime != "unknown" { @@ -632,119 +151,3 @@ func (a *App) GetVersion() string { } return constants.Version } - -// GetActivities 返回所有已注册的手册活动，供前端下拉选择。 -func (a *App) GetActivities() []types.ActivityInfo { - list := activity.List() - out := make([]types.ActivityInfo, 0, len(list)) - for _, act := range list { - out = append(out, types.ActivityInfo{ID: act.ID, DisplayName: act.DisplayName}) - } - return out -} - -// GetCurrentActivity 返回当前生效的活动。 -func (a *App) GetCurrentActivity() types.ActivityInfo { - cli := a.getClient() - act := cli.Activity - return types.ActivityInfo{ID: act.ID, DisplayName: act.DisplayName} -} - -// SwitchActivity 切换当前活动并持久化到配置；切换后重建客户端与时间基准。 -func (a *App) SwitchActivity(id string) types.SwitchActivityResult { - act := activity.Get(id) - if act == nil { - return types.SwitchActivityResult{Error: "未知活动: " + id} - } - cfg := a.getCfg() - cfg.Activity = act.ID - // 仅当用户未自定义时间校准接口时才回退默认值，避免切换活动静默覆盖用户在设置里填的 timeApiUrl。 - if cfg.TimeAPIURL == "" { - cfg.TimeAPIURL = constants.DefaultTimeAPI - } - if err := config.SaveConfig(a.paths.ConfigFile, cfg); err != nil { - logger.LogEvent("活动", "error", "切换活动配置保存失败: %v", err) - return types.SwitchActivityResult{Error: "配置保存失败: " + err.Error()} - } - a.setCfg(cfg) - newClient := client.New(act, cfg.TimeAPIURL, a.trace) - a.setClient(newClient) - if !a.scheduler.UpdateClient(newClient) { - logger.LogEvent("活动", "warn", "抢购任务正在运行，调度器客户端将在下次抢购时生效") - } - if sample, err := newClient.ServerTimeSample(); err == nil { - logger.SetDouyuOffset(sample.Offset) - } - logger.LogEvent("活动", "success", "已切换到活动: %s", act.DisplayName) - return types.SwitchActivityResult{Success: true, Activity: act.ID, DisplayName: act.DisplayName} -} - -// GetSettings 返回当前配置。 -func (a *App) GetSettings() types.Config { - return a.getCfg() -} - -// SaveSettings 保存配置并运行时生效。 -func (a *App) SaveSettings(cfg types.Config) types.SimpleResult { - // 校验 - if cfg.Concurrency <= 0 { - cfg.Concurrency = constants.DefaultConcurrency - } - if cfg.Attempts <= 0 { - cfg.Attempts = constants.DefaultAttempts - } - if cfg.LogLevel == "" { - cfg.LogLevel = "info" - } - - // 写入文件 - if err := config.SaveConfig(a.paths.ConfigFile, cfg); err != nil { - logger.LogEvent("配置", "error", "配置保存失败: %v", err) - return types.SimpleResult{} - } - - // 运行时生效 - a.setCfg(cfg) - logger.SetMinLevel(cfg.LogLevel) - a.scheduler.UpdateSettings(cfg.Concurrency, cfg.Attempts, cfg.RetryStages, cfg.SeckillMode) - - logger.LogEvent("配置", "success", "配置已保存并生效: 日志等级=%s 并发=%d 重试=%d 重试阶段=%d 模式=%d trace=%v", - cfg.LogLevel, cfg.Concurrency, cfg.Attempts, len(cfg.RetryStages), cfg.SeckillMode, cfg.TraceEnabled) - return types.SimpleResult{Success: true} -} - -// ==================== 内部辅助方法 ==================== - -func (a *App) setProducts(products []types.Product, fetchedAt time.Time) { - a.productsMu.Lock() - a.products = products - a.productsAt = fetchedAt - a.productsMu.Unlock() - if a.scheduler != nil { - a.scheduler.UpdateProducts(products) - } -} - -func (a *App) productPayload(source string) types.ProductPayload { - a.productsMu.RLock() - defer a.productsMu.RUnlock() - - payload := types.ProductPayload{ - Success: true, - Source: source, - Products: a.products, - Count: len(a.products), - } - if !a.productsAt.IsZero() { - age := time.Since(a.productsAt) - if age < 0 { - age = 0 - } - payload.FetchedAt = a.productsAt.In(constants.CST).Format("2006-01-02 15:04:05.000") - payload.AgeSecond = int(age.Seconds()) - payload.Stale = age > constants.ProductCacheTTL - } else { - payload.Stale = true - } - return payload -} diff --git a/app_activity.go b/app_activity.go new file mode 100644 index 0000000..d2971ff --- /dev/null +++ b/app_activity.go @@ -0,0 +1,90 @@ +package main + +import ( + "douyu_go/internal/activity" + "douyu_go/internal/client" + "douyu_go/internal/config" + "douyu_go/internal/constants" + "douyu_go/internal/logger" + "douyu_go/internal/types" +) + +// GetActivities 返回所有已注册的手册活动，供前端下拉选择。 +func (a *App) GetActivities() []types.ActivityInfo { + list := activity.List() + out := make([]types.ActivityInfo, 0, len(list)) + for _, act := range list { + out = append(out, types.ActivityInfo{ID: act.ID, DisplayName: act.DisplayName}) + } + return out +} + +// GetCurrentActivity 返回当前生效的活动。 +func (a *App) GetCurrentActivity() types.ActivityInfo { + cli := a.getClient() + act := cli.Activity + return types.ActivityInfo{ID: act.ID, DisplayName: act.DisplayName} +} + +// SwitchActivity 切换当前活动并持久化到配置；切换后重建客户端与时间基准。 +func (a *App) SwitchActivity(id string) types.SwitchActivityResult { + act := activity.Get(id) + if act == nil { + return types.SwitchActivityResult{Error: "未知活动: " + id} + } + cfg := a.getCfg() + cfg.Activity = act.ID + // 仅当用户未自定义时间校准接口时才回退默认值，避免切换活动静默覆盖用户在设置里填的 timeApiUrl。 + if cfg.TimeAPIURL == "" { + cfg.TimeAPIURL = constants.DefaultTimeAPI + } + if err := config.SaveConfig(a.paths.ConfigFile, cfg); err != nil { + logger.LogEvent("活动", "error", "切换活动配置保存失败: %v", err) + return types.SwitchActivityResult{Error: "配置保存失败: " + err.Error()} + } + a.setCfg(cfg) + newClient := client.New(act, cfg.TimeAPIURL, a.trace) + a.setClient(newClient) + if !a.scheduler.UpdateClient(newClient) { + logger.LogEvent("活动", "warn", "抢购任务正在运行，调度器客户端将在下次抢购时生效") + } + if sample, err := newClient.ServerTimeSample(); err == nil { + logger.SetDouyuOffset(sample.Offset) + } + logger.LogEvent("活动", "success", "已切换到活动: %s", act.DisplayName) + return types.SwitchActivityResult{Success: true, Activity: act.ID, DisplayName: act.DisplayName} +} + +// GetSettings 返回当前配置。 +func (a *App) GetSettings() types.Config { + return a.getCfg() +} + +// SaveSettings 保存配置并运行时生效。 +func (a *App) SaveSettings(cfg types.Config) types.SimpleResult { + // 校验 + if cfg.Concurrency <= 0 { + cfg.Concurrency = constants.DefaultConcurrency + } + if cfg.Attempts <= 0 { + cfg.Attempts = constants.DefaultAttempts + } + if cfg.LogLevel == "" { + cfg.LogLevel = "info" + } + + // 写入文件 + if err := config.SaveConfig(a.paths.ConfigFile, cfg); err != nil { + logger.LogEvent("配置", "error", "配置保存失败: %v", err) + return types.SimpleResult{} + } + + // 运行时生效 + a.setCfg(cfg) + logger.SetMinLevel(cfg.LogLevel) + a.scheduler.UpdateSettings(cfg.Concurrency, cfg.Attempts, cfg.RetryStages, cfg.SeckillMode) + + logger.LogEvent("配置", "success", "配置已保存并生效: 日志等级=%s 并发=%d 重试=%d 重试阶段=%d 模式=%d trace=%v", + cfg.LogLevel, cfg.Concurrency, cfg.Attempts, len(cfg.RetryStages), cfg.SeckillMode, cfg.TraceEnabled) + return types.SimpleResult{Success: true} +} diff --git a/app_export.go b/app_export.go new file mode 100644 index 0000000..ea30f7b --- /dev/null +++ b/app_export.go @@ -0,0 +1,175 @@ +package main + +import ( + "bufio" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "douyu_go/internal/logger" + "douyu_go/internal/types" + "douyu_go/internal/utils" +) + +// accountLine 账号文件的一行解析结果。 +type accountLine struct { + Raw string // 原始行文本 + Name string // 解析出的账号名（注释/空行为空） + Comment bool // 是否为注释或空行 +} + +// parseAccountFile 读取账号文件并按行解析账号名。 +func parseAccountFile(path string) ([]accountLine, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var lines []accountLine + scanner := bufio.NewScanner(strings.NewReader(string(data))) + for scanner.Scan() { + raw := scanner.Text() + trimmed := strings.TrimSpace(raw) + if trimmed == "" || strings.HasPrefix(trimmed, "#") { + lines = append(lines, accountLine{Raw: raw, Comment: true}) + continue + } + name := "" + if parts := strings.SplitN(trimmed, "----", 3); len(parts) >= 1 { + name = strings.TrimSpace(parts[0]) + } + lines = append(lines, accountLine{Raw: raw, Name: name}) + } + return lines, scanner.Err() +} + +// ExportAccounts 导出指定账号到文件。 +func (a *App) ExportAccounts(accountNames []string, filename string) types.ExportAccountsResult { + if len(accountNames) == 0 { + return types.ExportAccountsResult{Error: "没有要导出的账号"} + } + + accountFile := a.scheduler.AccountFilePath() + lines, err := parseAccountFile(accountFile) + if err != nil { + return types.ExportAccountsResult{Error: "读取账号文件失败: " + err.Error()} + } + + nameSet := make(map[string]bool) + for _, name := range accountNames { + nameSet[name] = true + } + + var matchedLines []string + for _, line := range lines { + if line.Comment { + continue + } + if nameSet[line.Name] { + matchedLines = append(matchedLines, line.Raw) + } + } + + if len(matchedLines) == 0 { + return types.ExportAccountsResult{Error: "没有找到匹配的账号行"} + } + + // 生成导出文件路径 + exportPath := filepath.Join(filepath.Dir(accountFile), filename) + content := strings.Join(matchedLines, "\n") + "\n" + + if err := os.WriteFile(exportPath, []byte(content), 0644); err != nil { + return types.ExportAccountsResult{Error: "写入文件失败: " + err.Error()} + } + + logger.LogEvent("导出", "success", "已导出 %d 个账号到: %s", len(matchedLines), exportPath) + + return types.ExportAccountsResult{ + Success: true, + Count: len(matchedLines), + FilePath: exportPath, + } +} + +// ExportAccountsWithProduct 导出指定账号到文件，自动使用中文商品名和时间戳命名。 +func (a *App) ExportAccountsWithProduct(accountNames []string, commodityID, commodityName string) types.ExportAccountsResult { + // 生成文件名: user_已抢到_商品中文名_20260626_150405.txt + now := time.Now() + timestamp := now.Format("20060102_150405") + // 清理商品名中的特殊字符（复用 utils.SafeFilename，避免与 utils 包重复实现） + safeName := utils.SafeFilename(commodityName) + filename := fmt.Sprintf("user_已抢到_%s_%s.txt", safeName, timestamp) + + return a.ExportAccounts(accountNames, filename) +} + +// OrganizeAccounts 整理账号，把已抢到指定商品的账号分离出来。 +func (a *App) OrganizeAccounts(commodityID, commodityName string, startTime, endTime int64) types.OrganizeAccountsResult { + accounts := a.scheduler.AccountSnapshot() + hitAccounts := make([]string, 0) + missAccounts := make([]string, 0) + lookups := a.lookupExchangeRecords("整理", accounts) + + for _, lookup := range lookups { + if lookup.EmptyCookie || lookup.Err != nil { + missAccounts = append(missAccounts, lookup.Account.Name) + continue + } + + if hasCommodityRecord(lookup.Records, commodityID, startTime, endTime) { + hitAccounts = append(hitAccounts, lookup.Account.Name) + } else { + missAccounts = append(missAccounts, lookup.Account.Name) + } + } + + // 导出已抢到的账号 + var exportResult types.ExportAccountsResult + if len(hitAccounts) > 0 { + exportResult = a.ExportAccountsWithProduct(hitAccounts, commodityID, commodityName) + } + + // 更新原文件，只保留未抢到的 + accountFile := a.scheduler.AccountFilePath() + lines, err := parseAccountFile(accountFile) + if err != nil { + return types.OrganizeAccountsResult{Error: "读取账号文件失败: " + err.Error()} + } + + missSet := make(map[string]bool) + for _, name := range missAccounts { + missSet[name] = true + } + + var keepLines []string + for _, line := range lines { + if line.Comment { + keepLines = append(keepLines, line.Raw) + continue + } + if missSet[line.Name] { + keepLines = append(keepLines, line.Raw) + } + } + + // 写回原文件 + newContent := strings.Join(keepLines, "\n") + "\n" + if err := os.WriteFile(accountFile, []byte(newContent), 0644); err != nil { + return types.OrganizeAccountsResult{Error: "更新账号文件失败: " + err.Error()} + } + + // 重新加载账号 + a.scheduler.LoadAccounts() + + logger.LogEvent("整理", "success", "整理完成: 已抢到=%d, 未抢到=%d", len(hitAccounts), len(missAccounts)) + + return types.OrganizeAccountsResult{ + Success: true, + HitCount: len(hitAccounts), + MissCount: len(missAccounts), + HitAccounts: hitAccounts, + MissAccounts: missAccounts, + ExportResult: exportResult, + } +} diff --git a/app_products.go b/app_products.go new file mode 100644 index 0000000..5eac365 --- /dev/null +++ b/app_products.go @@ -0,0 +1,82 @@ +package main + +import ( + "time" + + "douyu_go/internal/constants" + "douyu_go/internal/logger" + "douyu_go/internal/types" + "douyu_go/internal/utils" +) + +// LoadProducts 刷新商品列表。refresh=true 时从斗鱼实时拉取。 +func (a *App) LoadProducts(refresh bool) types.ProductPayload { + if refresh { + products, err := a.getClient().Products() + if err != nil { + logger.LogEvent("商品", "error", "商品刷新失败: %v", err) + return types.ProductPayload{Success: false} + } + fetchedAt := time.Now() + a.setProducts(products, fetchedAt) + logger.LogEvent("商品", "success", "商品刷新完成: %d 个，刷新时间=%s", len(products), fetchedAt.In(constants.CST).Format("15:04:05.000")) + return a.productPayload("live") + } + return a.productPayload("cache") +} + +// GetAccounts 返回账号列表。 +func (a *App) GetAccounts() []types.AccountView { + accounts := a.scheduler.AccountSnapshot() + list := make([]types.AccountView, 0, len(accounts)) + for i, account := range accounts { + list = append(list, types.AccountView{ + Index: i, + Name: account.Name, + Cookie: utils.MaskCookie(account.Cookie), + }) + } + return list +} + +// ReloadAccounts 重新加载账号文件。 +func (a *App) ReloadAccounts() types.ReloadAccountsResult { + a.scheduler.LoadAccounts() + return types.ReloadAccountsResult{Success: true, Count: a.scheduler.AccountCount()} +} + +// ==================== 内部辅助方法 ==================== + +func (a *App) setProducts(products []types.Product, fetchedAt time.Time) { + a.productsMu.Lock() + a.products = products + a.productsAt = fetchedAt + a.productsMu.Unlock() + if a.scheduler != nil { + a.scheduler.UpdateProducts(products) + } +} + +func (a *App) productPayload(source string) types.ProductPayload { + a.productsMu.RLock() + defer a.productsMu.RUnlock() + + payload := types.ProductPayload{ + Success: true, + Source: source, + Products: a.products, + Count: len(a.products), + } + if !a.productsAt.IsZero() { + age := time.Since(a.productsAt) + if age < 0 { + age = 0 + } + payload.FetchedAt = a.productsAt.In(constants.CST).Format("2006-01-02 15:04:05.000") + payload.AgeSecond = int(age.Seconds()) + payload.Stale = age > constants.ProductCacheTTL + } else { + payload.Stale = true + } + return payload +} diff --git a/app_seckill.go b/app_seckill.go new file mode 100644 index 0000000..e93de69 --- /dev/null +++ b/app_seckill.go @@ -0,0 +1,74 @@ +package main + +import ( + "time" + + "douyu_go/internal/constants" + "douyu_go/internal/logger" + "douyu_go/internal/types" + "douyu_go/internal/utils" +) + +// StartSeckill 立即启动抢购。 +func (a *App) StartSeckill(accountProducts map[string]string) types.MessageResult { + if len(accountProducts) == 0 { + return types.MessageResult{Error: "请选择账号和商品"} + } + if !a.scheduler.TryStartRun() { + return types.MessageResult{Error: "抢购任务正在运行，请勿重复提交"} + } + go a.scheduler.RunNow(utils.CloneStringMap(accountProducts)) + return types.MessageResult{Success: true, Msg: "抢购已启动"} +} + +// SetSchedule 设置定时抢购。 +func (a *App) SetSchedule(req types.SeckillRequest) (types.ScheduleResult, error) { + return a.scheduler.ScheduleRun(req) +} + +// CancelSchedule 取消未触发的定时抢购。 +func (a *App) CancelSchedule() types.CancelScheduleResult { + stopped := a.scheduler.CancelSchedule() + return types.CancelScheduleResult{Success: true, Stopped: stopped} +} + +// GetResults 返回当前抢购结果。 +func (a *App) GetResults() map[string]string { + return a.scheduler.Results() +} + +// GetLogs 返回后端内存日志，支持增量获取。 +func (a *App) GetLogs(since int64) types.LogsResult { + entries := a.logStore.Since(since) + nextSince := since + if len(entries) > 0 { + nextSince = entries[len(entries)-1].ID + } + return types.LogsResult{ + Success: true, + Logs: entries, + Since: nextSince, + File: a.logStore.FilePath(), + } +} + +// GetTime 返回服务端时间和本地时间。 +func (a *App) GetTime() types.TimeResult { + serverTime, err := a.getClient().ServerTime() + source := "currentTime" + if err != nil { + source = "localFallback" + logger.LogEvent("时间", "warn", "时间接口降级: %v", err) + } + return types.TimeResult{ + ServerTime: serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"), + LocalTime: time.Now().In(constants.CST).Format("2006-01-02 15:04:05.000"), + Source: source, + } +} + +// ClearResults 清空抢购结果。 +func (a *App) ClearResults() types.SimpleResult { + a.scheduler.ClearResults() + return types.SimpleResult{Success: true} +} diff --git a/app_verify.go b/app_verify.go new file mode 100644 index 0000000..758ec38 --- /dev/null +++ b/app_verify.go @@ -0,0 +1,225 @@ +package main + +import ( + "fmt" + "strconv" + "strings" + "sync" + "time" + + "douyu_go/internal/logger" + "douyu_go/internal/types" +) + +const exchangeLookupConcurrency = 8 + +type exchangeLookupResult struct { + Account types.Account + Records []types.ExchangeRecord + Err error + EmptyCookie bool +} + +// QueryScore 查询单个账号的积分。 +func (a *App) QueryScore(accountName string) types.ScoreResult { + if accountName == "" { + return types.ScoreResult{Score: "缺少账号名"} + } + cookie, ok := a.scheduler.AccountCookieByName(accountName) + if !ok || cookie == "" { + return types.ScoreResult{Score: fmt.Sprintf("账号不存在或 Cookie 为空: %s", accountName)} + } + score, err := a.getClient().Score(cookie) + if err != nil { + return types.ScoreResult{Score: "查询失败: " + err.Error()} + } + return types.ScoreResult{Score: strconv.Itoa(score)} +} + +// VerifyExchange 查询兑换记录，用于复验抢购结果。 +func (a *App) VerifyExchange(accountName string) types.ExchangeRecordsResult { + if accountName == "" { + return types.ExchangeRecordsResult{Error: "缺少账号名"} + } + cookie, ok := a.scheduler.AccountCookieByName(accountName) + if !ok || cookie == "" { + return types.ExchangeRecordsResult{Error: fmt.Sprintf("账号不存在或 Cookie 为空: %s", accountName)} + } + + records, err := a.getClient().ExchangeList(cookie) + if err != nil { + return types.ExchangeRecordsResult{Error: "查询失败: " + err.Error()} + } + matched := filterExchangeRecords(records) + + return types.ExchangeRecordsResult{ + Success: true, + Records: matched, + Count: len(matched), + } +} + +// VerifyAllExchange 批量查询兑换记录，使用有限并发减少等待时间。 +func (a *App) VerifyAllExchange() types.VerifyAllExchangeResult { + accounts := a.scheduler.AccountSnapshot() + results := make(map[string]types.AccountExchangeResult) + lookups := a.lookupExchangeRecords("复验", accounts) + + for _, lookup := range lookups { + if lookup.EmptyCookie { + results[lookup.Account.Name] = types.AccountExchangeResult{Error: "Cookie 为空"} + continue + } + + if lookup.Err != nil { + results[lookup.Account.Name] = types.AccountExchangeResult{Error: "查询失败: " + lookup.Err.Error()} + continue + } + + matched := filterExchangeRecords(lookup.Records) + results[lookup.Account.Name] = types.AccountExchangeResult{ + Success: true, + Records: matched, + Count: len(matched), + } + } + + return types.VerifyAllExchangeResult{Results: results} +} + +// VerifyAndExport 复验并导出结果。 +func (a *App) VerifyAndExport(commodityID string) types.VerifyAndExportResult { + accounts := a.scheduler.AccountSnapshot() + results := make(map[string]types.AccountHitResult) + hitAccounts := make([]string, 0) + missAccounts := make([]string, 0) + lookups := a.lookupExchangeRecords("复验", accounts) + + for _, lookup := range lookups { + if lookup.EmptyCookie { + missAccounts = append(missAccounts, lookup.Account.Name) + results[lookup.Account.Name] = types.AccountHitResult{Error: "Cookie 为空"} + continue + } + + if lookup.Err != nil { + missAccounts = append(missAccounts, lookup.Account.Name) + results[lookup.Account.Name] = types.AccountHitResult{Error: "查询失败: " + lookup.Err.Error()} + continue + } + + // 检查是否抢到了指定商品 + hit := hasCommodityRecord(lookup.Records, commodityID, 0, 0) + if hit { + hitAccounts = append(hitAccounts, lookup.Account.Name) + results[lookup.Account.Name] = types.AccountHitResult{Success: true, Hit: true} + } else { + missAccounts = append(missAccounts, lookup.Account.Name) + results[lookup.Account.Name] = types.AccountHitResult{Success: true} + } + } + + return types.VerifyAndExportResult{ + Success: true, + HitCount: len(hitAccounts), + MissCount: len(missAccounts), + HitAccounts: hitAccounts, + MissAccounts: missAccounts, + Details: results, + } +} + +func (a *App) lookupExchangeRecords(category string, accounts []types.Account) []exchangeLookupResult { + results := make([]exchangeLookupResult, len(accounts)) + if len(accounts) == 0 { + return results + } + + workerCount := exchangeLookupConcurrency + if len(accounts) < workerCount { + workerCount = len(accounts) + } + start := time.Now() + logger.LogEvent(category, "info", "兑换记录批量查询开始: 账号=%d 并发=%d", len(accounts), workerCount) + + jobs := make(chan int) + var wg sync.WaitGroup + for worker := 0; worker < workerCount; worker++ { + wg.Add(1) + go func() { + defer wg.Done() + for idx := range jobs { + account := accounts[idx] + lookup := exchangeLookupResult{Account: account} + if strings.TrimSpace(account.Cookie) == "" { + lookup.EmptyCookie = true + results[idx] = lookup + continue + } + records, err := a.getClient().ExchangeList(account.Cookie) + lookup.Records = records + lookup.Err = err + results[idx] = lookup + } + }() + } + for idx := range accounts { + jobs <- idx + } + close(jobs) + wg.Wait() + + errors := 0 + empty := 0 + for _, result := range results { + if result.EmptyCookie { + empty++ + } + if result.Err != nil { + errors++ + } + } + level := "success" + if errors > 0 { + level = "warn" + } + logger.LogEvent(category, level, "兑换记录批量查询完成: 账号=%d 空CK=%d 失败=%d 耗时=%s", + len(accounts), empty, errors, time.Since(start).Round(time.Millisecond)) + return results +} + +func filterExchangeRecords(records []types.ExchangeRecord) []types.ExchangeRecord { + matched := make([]types.ExchangeRecord, 0, len(records)) + for _, r := range records { + if r.CommodityID != "" { + matched = append(matched, r) + } + } + return matched +} + +func hasCommodityRecord(records []types.ExchangeRecord, commodityID string, startTime, endTime int64) bool { + if commodityID == "" { + return false + } + for _, r := range records { + if r.CommodityID != commodityID { + continue + } + if startTime == 0 && endTime == 0 { + return true + } + recordTime, err := strconv.ParseInt(r.Time, 10, 64) + if err != nil { + continue + } + if startTime > 0 && recordTime < startTime { + continue + } + if endTime > 0 && recordTime > endTime { + continue + } + return true + } + return false +} diff --git a/internal/config/config.go b/internal/config/config.go index 9e8035d..59ac0ce 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -5,11 +5,11 @@ import ( "errors" "os" "path/filepath" - "runtime" "strconv" "douyu_go/internal/activity" "douyu_go/internal/constants" + imgpkg "douyu_go/internal/image" "douyu_go/internal/logger" "douyu_go/internal/types" ) @@ -22,7 +22,7 @@ func DefaultAppPaths() types.AppPaths { AccountFile: resolvePath(envOrDefault("DOUYU_USERS", constants.DefaultAccountFile)), LogDir: resolvePath(envOrDefault("DOUYU_LOG_DIR", constants.DefaultLogDir)), SuccessDir: resolvePath(envOrDefault("DOUYU_SUCCESS_DIR", constants.DefaultSuccessDir)), - FontFiles: CollectFontCandidates(), + FontFiles: imgpkg.CollectFontCandidates(), } } @@ -153,41 +153,3 @@ func SaveConfig(path string, cfg types.Config) error { } return os.WriteFile(path, data, 0644) } - -// CollectFontCandidates 收集常见中文字体路径，允许用 DOUYU_FONT 指定自定义字体。 -func CollectFontCandidates() []string { - candidates := []string{} - if custom := os.Getenv("DOUYU_FONT"); custom != "" { - candidates = append(candidates, custom) - } - candidates = append(candidates, - "Microsoft Yahei.ttf", - "Microsoft YaHei.ttf", - "NotoSansCJKsc-Regular.otf", - "NotoSansCJK-Regular.ttc", - ) - - switch runtime.GOOS { - case "darwin": - candidates = append(candidates, - "/System/Library/Fonts/PingFang.ttc", - "/System/Library/Fonts/Supplemental/Arial Unicode.ttf", - "/System/Library/Fonts/Supplemental/Songti.ttc", - ) - case "windows": - candidates = append(candidates, - `C:\Windows\Fonts\msyh.ttc`, - `C:\Windows\Fonts\simhei.ttf`, - `C:\Windows\Fonts\simsun.ttc`, - ) - default: - candidates = append(candidates, - "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", - "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.otf", - "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc", - "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", - "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", - ) - } - return candidates -} diff --git a/internal/image/font_paths.go b/internal/image/font_paths.go new file mode 100644 index 0000000..a7c527d --- /dev/null +++ b/internal/image/font_paths.go @@ -0,0 +1,44 @@ +package image + +import ( + "os" + "runtime" +) + +// CollectFontCandidates 收集常见中文字体路径，允许用 DOUYU_FONT 指定自定义字体。 +func CollectFontCandidates() []string { + candidates := []string{} + if custom := os.Getenv("DOUYU_FONT"); custom != "" { + candidates = append(candidates, custom) + } + candidates = append(candidates, + "Microsoft Yahei.ttf", + "Microsoft YaHei.ttf", + "NotoSansCJKsc-Regular.otf", + "NotoSansCJK-Regular.ttc", + ) + + switch runtime.GOOS { + case "darwin": + candidates = append(candidates, + "/System/Library/Fonts/PingFang.ttc", + "/System/Library/Fonts/Supplemental/Arial Unicode.ttf", + "/System/Library/Fonts/Supplemental/Songti.ttc", + ) + case "windows": + candidates = append(candidates, + `C:\Windows\Fonts\msyh.ttc`, + `C:\Windows\Fonts\simhei.ttf`, + `C:\Windows\Fonts\simsun.ttc`, + ) + default: + candidates = append(candidates, + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.otf", + "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + ) + } + return candidates +} diff --git a/internal/logger/logger.go b/internal/logger/logger.go index 48d75ba..4880c23 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -226,113 +226,6 @@ func formatLogLine(now time.Time, category, level, account, message string) stri return fmt.Sprintf("%s[%s][%s] %s", prefix, category, level, message) } -// LogMeta 是从日志文本中解析出的结构化信息。 -type LogMeta struct { - Time string - Category string - Level string - Account string -} - -// ParseLogLine 从标准日志行中提取毫秒时间、分类、级别和账号。 -func ParseLogLine(line string) LogMeta { - meta := LogMeta{ - Time: parseLogTime(line), - } - - idx := strings.Index(line, "[") - if idx < 0 { - meta.Account = parseAccountFromText(line) - return meta - } - category, rest, ok := cutBracket(line[idx:]) - if !ok { - meta.Account = parseAccountFromText(line) - return meta - } - level, rest, ok := cutBracket(rest) - if !ok || !IsLogLevel(level) { - meta.Category = category - meta.Account = parseAccountFromText(line) - return meta - } - - meta.Category = category - meta.Level = level - rest = strings.TrimLeft(rest, " \t") - tag, _, ok := cutBracket(rest) - if ok && strings.HasPrefix(tag, "账号:") { - meta.Account = strings.TrimSpace(strings.TrimPrefix(tag, "账号:")) - } - if meta.Account == "" { - meta.Account = parseAccountFromText(line) - } - return meta -} - -func parseLogTime(line string) string { - if len(line) < len(logTimeLayout) { - return "" - } - raw := line[:len(logTimeLayout)] - t, err := time.ParseInLocation(logTimeLayout, raw, constants.CST) - if err != nil { - return "" - } - return t.Format(logEntryTimeLayout) -} - -func parseAccountFromText(line string) string { - for _, marker := range []string{"账号=", "账号:"} { - idx := strings.Index(line, marker) - if idx < 0 { - continue - } - start := idx + len(marker) - rest := strings.TrimSpace(line[start:]) - end := strings.IndexFunc(rest, func(r rune) bool { - return r == ' ' || r == '\t' || r == ']' || r == ',' || r == '，' || r == '|' || r == ';' || r == '；' - }) - if end >= 0 { - rest = rest[:end] - } - return strings.TrimSpace(rest) - } - return "" -} - -// ParseLogMeta 从标准日志行中提取类似 [网络][warn] 的分类和级别。 -func ParseLogMeta(line string) (string, string) { - meta := ParseLogLine(line) - return meta.Category, meta.Level -} - -func cutBracket(s string) (string, string, bool) { - if !strings.HasPrefix(s, "[") { - return "", s, false - } - end := strings.Index(s, "]") - if end <= 1 { - return "", s, false - } - return s[1:end], s[end+1:], true -} - -// IsLogLevel 判断是否为有效日志级别。 -func IsLogLevel(level string) bool { - switch strings.ToLower(level) { - case "success", "error", "warn", "info", "debug": - return true - default: - return false - } -} - -// LogDebug 输出 debug 级别日志，仅在日志等级设为 debug 时可见。 -func LogDebug(category, format string, args ...any) { - LogEvent(category, "debug", format, args...) -} - // resolveAbsolutePath 将相对路径解析为绝对路径。 // 如果路径已经是绝对路径则直接返回；否则基于可执行文件所在目录解析。 func resolveAbsolutePath(path string) string { @@ -349,28 +242,3 @@ func resolveAbsolutePath(path string) string { } return path } - -// DetectLogLevel 根据日志文本粗略分类，方便前端染色。 -func DetectLogLevel(line string) string { - lower := strings.ToLower(line) - switch { - case strings.Contains(lower, "[debug]"): - return "debug" - case strings.Contains(lower, "[success]"): - return "success" - case strings.Contains(lower, "[error]"): - return "error" - case strings.Contains(lower, "[warn]"): - return "warn" - case strings.Contains(lower, "[info]"): - return "info" - case strings.Contains(line, "成功"): - return "success" - case strings.Contains(line, "失败") || strings.Contains(line, "错误") || strings.Contains(lower, "error") || strings.Contains(lower, "panic"): - return "error" - case strings.Contains(line, "降级") || strings.Contains(line, "跳过") || strings.Contains(line, "不存在") || strings.Contains(lower, "warn"): - return "warn" - default: - return "info" - } -} diff --git a/internal/logger/logger_test.go b/internal/logger/logger_test.go index 8d47d56..b71949d 100644 --- a/internal/logger/logger_test.go +++ b/internal/logger/logger_test.go @@ -5,82 +5,6 @@ import ( "testing" ) -func TestIsLogLevel(t *testing.T) { - valid := []string{"success", "error", "warn", "info", "debug", "SUCCESS", "Warn"} - for _, l := range valid { - if !IsLogLevel(l) { - t.Errorf("IsLogLevel(%q) = false, want true", l) - } - } - invalid := []string{"fatal", "trace", "", "warning"} - for _, l := range invalid { - if IsLogLevel(l) { - t.Errorf("IsLogLevel(%q) = true, want false", l) - } - } -} - -func TestDetectLogLevel(t *testing.T) { - cases := []struct { - line string - want string - }{ - {"[2026/01/01 00:00:00.000][网络][debug] 调试", "debug"}, - {"[2026/01/01 00:00:00.000][抢购][success] 兑换成功", "success"}, - {"[2026/01/01 00:00:00.000][系统][error] 失败", "error"}, - {"[2026/01/01 00:00:00.000][系统][warn] 跳过", "warn"}, - {"[2026/01/01 00:00:00.000][系统][info] 信息", "info"}, - {"兑换成功完成", "success"}, - {"网络请求失败 error 发生", "error"}, - {"配置降级处理 warn", "warn"}, - {"普通信息日志", "info"}, - } - for _, c := range cases { - if got := DetectLogLevel(c.line); got != c.want { - t.Errorf("DetectLogLevel(%q) = %q, want %q", c.line, got, c.want) - } - } -} - -func TestParseLogLine(t *testing.T) { - line := "2026/01/01 12:00:00.123 [抢购][success] 账号:testuser 兑换成功" - meta := ParseLogLine(line) - if meta.Time != "2026-01-01 12:00:00.123" { - t.Errorf("ParseLogLine Time = %q, want '2026-01-01 12:00:00.123'", meta.Time) - } - if meta.Category != "抢购" { - t.Errorf("ParseLogLine Category = %q, want '抢购'", meta.Category) - } - if meta.Level != "success" { - t.Errorf("ParseLogLine Level = %q, want 'success'", meta.Level) - } - if meta.Account != "testuser" { - t.Errorf("ParseLogLine Account = %q, want 'testuser'", meta.Account) - } -} - -func TestParseLogLineShort(t *testing.T) { - // 不含时间的行，ParseLogLine 不应 panic，Time 应为空 - line := "[系统][info] 配置已加载" - meta := ParseLogLine(line) - if meta.Category != "系统" { - t.Errorf("ParseLogLine Category = %q, want '系统'", meta.Category) - } - if meta.Level != "info" { - t.Errorf("ParseLogLine Level = %q, want 'info'", meta.Level) - } - if meta.Time != "" { - t.Errorf("ParseLogLine Time = %q, want empty", meta.Time) - } -} - -func TestParseLogMeta(t *testing.T) { - cat, lvl := ParseLogMeta("[网络][warn] 降级") - if cat != "网络" || lvl != "warn" { - t.Errorf("ParseLogMeta = (%q, %q), want ('网络', 'warn')", cat, lvl) - } -} - // TestStructuredLogging 验证 LogEvent→sink→MemoryLogStore 结构化链路： // LogEntry 的 category/level/account 应直接由调用方填入，而非字符串反解析。 func TestStructuredLogging(t *testing.T) { diff --git a/internal/seckill/attempt.go b/internal/seckill/attempt.go new file mode 100644 index 0000000..c7893e5 --- /dev/null +++ b/internal/seckill/attempt.go @@ -0,0 +1,229 @@ +package seckill + +import ( + "strings" + "time" + + "douyu_go/internal/logger" + "douyu_go/internal/types" + "douyu_go/internal/utils" +) + +// runAttempt 标准模式兑换：完整执行 CSRF→BatchLimit→PreExchange→Exchange 链路。 +func (s *Scheduler) runAttempt(attempt types.ExchangeAttempt) types.ExchangeResult { + cli := s.clientSnapshot() + result := types.ExchangeResult{ + Account: attempt.Account.Name, + Product: attempt.Product.CommodityName, + Attempt: attempt.Index, + } + + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "执行资格预检链路: 账号=%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.LogAccountEvent("抢购", "info", attempt.Account.Name, "使用预热CSRF: 账号=%s 商品=%s 预热距今=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(warmup.CreatedAt))) + } + if csrf == "" { + var err error + csrf, err = cli.GenerateCSRF(attempt.Account.Cookie) + if err != nil { + result.Message = "失败|生成CSRF失败: " + err.Error() + return result + } + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "CSRF已生成，本次兑换链路复用同一个 token: 账号=%s 商品=%s", attempt.Account.Name, attempt.Product.CommodityName) + } + + limit, err := cli.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.LogAccountEvent("抢购", "info", attempt.Account.Name, "批量限制通过: 账号=%s 商品=%s limit=%d", attempt.Account.Name, attempt.Product.CommodityName, limit) + + if err := cli.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf); err != nil { + result.Message = "失败|预兑换失败: " + err.Error() + return result + } + s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ + CSRF: csrf, + RoleName: roleName, + Qualified: true, + LastCheck: time.Now(), + }) + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "资格预检通过: 账号=%s 商品=%s，后续可直兑", attempt.Account.Name, attempt.Product.CommodityName) + + exchangeInfo, err := cli.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf) + if err != nil { + s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ + CSRF: csrf, + RoleName: roleName, + Qualified: true, + LastCheck: time.Now(), + }) + result.RoleName = roleName + result.Message = "失败|" + err.Error() + return result + } + + if roleName == "" { + roleName, err = cli.RoleName(attempt.Account.Cookie) + if err != nil { + logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "角色名查询失败: 账号=%s err=%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, + Qualified: true, + LastCheck: time.Now(), + }) + + return result +} + +// runAttemptFast 激进模式的兑换逻辑，复用缓存的 CSRF，跳过 BatchExchangeLimit 和 PreExchange。 +func (s *Scheduler) runAttemptFast(attempt types.ExchangeAttempt) types.ExchangeResult { + cli := s.clientSnapshot() + result := types.ExchangeResult{ + Account: attempt.Account.Name, + Product: attempt.Product.CommodityName, + Attempt: attempt.Index, + } + + cache := s.getAttemptCache(attempt.Account.Name) + if cache == nil || cache.CSRF == "" || time.Since(cache.LastCheck) >= qualifiedMaxAge { + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "激进模式执行资格预检: 账号=%s 原因=无缓存CSRF", attempt.Account.Name) + return s.runAttempt(attempt) + } + if !cache.Qualified { + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "激进模式执行资格预检: 账号=%s 原因=未完成资格预检", attempt.Account.Name) + return s.runAttempt(attempt) + } + + csrf := cache.CSRF + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "激进模式直兑: 账号=%s 商品=%s 第%d包", attempt.Account.Name, attempt.Product.CommodityName, attempt.Index) + + // 直接调用 Exchange + exchangeInfo, err := cli.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf) + if err != nil { + s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ + CSRF: csrf, + RoleName: cache.RoleName, + Qualified: true, + LastCheck: time.Now(), + }) + result.RoleName = cache.RoleName + errMsg := err.Error() + if shouldFallbackStandard(errMsg) { + logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "激进模式回退标准: 账号=%s 原因=%s", attempt.Account.Name, utils.TrimForLog(errMsg, 80)) + s.setAttemptCache(attempt.Account.Name, nil) // 清空缓存 + return s.runAttempt(attempt) + } + result.Message = "失败|" + errMsg + return result + } + + // 获取角色名 + roleName := "" + if cache != nil { + roleName = cache.RoleName + } + if roleName == "" { + var err error + roleName, err = cli.RoleName(attempt.Account.Cookie) + if err != nil { + logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "角色名查询失败: 账号=%s err=%v", attempt.Account.Name, err) + } + } + s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ + CSRF: csrf, + RoleName: roleName, + Qualified: true, + LastCheck: time.Now(), + }) + + result.OK = true + result.RoleName = roleName + if roleName != "" { + result.Message = "成功|" + roleName + "|" + exchangeInfo + } else { + result.Message = "成功|" + exchangeInfo + } + return result +} + +func shouldFallbackStandard(errMsg string) bool { + lower := strings.ToLower(errMsg) + switch { + case strings.Contains(lower, "csrf"): + return true + case strings.Contains(lower, "cvl_csrf_token"): + return true + case strings.Contains(lower, "preexchange"): + return true + case strings.Contains(lower, "limit"): + return true + case strings.Contains(errMsg, "预兑换"): + return true + case strings.Contains(errMsg, "限制检查"): + return true + default: + return false + } +} + +// ========== 激进模式缓存管理 ========== + +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, + Qualified: warmup.Qualified, + LastCheck: time.Now(), + }) + logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "从预热初始化缓存: 账号=%s", attempt.Account.Name) + } + } +} + +func (s *Scheduler) setAttemptCache(accountName string, cache *types.AttemptCache) { + s.attemptCacheMu.Lock() + defer s.attemptCacheMu.Unlock() + if cache == nil { + delete(s.attemptCache, accountName) + return + } + s.attemptCache[accountName] = cache +} + +func (s *Scheduler) clearAttemptCache() { + s.attemptCacheMu.Lock() + defer s.attemptCacheMu.Unlock() + s.attemptCache = make(map[string]*types.AttemptCache) +} diff --git a/internal/seckill/schedule_timer.go b/internal/seckill/schedule_timer.go new file mode 100644 index 0000000..94b1403 --- /dev/null +++ b/internal/seckill/schedule_timer.go @@ -0,0 +1,192 @@ +package seckill + +import ( + "errors" + "fmt" + "sync/atomic" + "time" + + "douyu_go/internal/constants" + "douyu_go/internal/logger" + "douyu_go/internal/types" + "douyu_go/internal/utils" +) + +// ScheduleRun 注册一次定时抢购。 +func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, error) { + if len(req.AccountProducts) == 0 { + return types.ScheduleResult{}, errors.New("请选择账号和商品") + } + if req.TargetTime == "" { + return types.ScheduleResult{}, errors.New("请设置目标时间") + } + + targetLocal, err := time.ParseInLocation("2006-01-02 15:04:05", req.TargetTime, constants.CST) + if err != nil { + return types.ScheduleResult{}, fmt.Errorf("时间格式错误: %w", err) + } + targetTime := targetLocal.UTC() + + calibration := s.calibrateDouyuTime("定时") + serverTime := calibration.EstimatedServerNow() + + triggerTime := targetTime.Add(-internalAdvance) + remain := triggerTime.Sub(serverTime) + if remain <= 0 { + return types.ScheduleResult{}, errors.New("目标时间已过") + } + + ap := utils.CloneStringMap(req.AccountProducts) + s.scheduleMu.Lock() + s.stopSchedulesLocked() + s.clearWarmups() + s.scheduleMu.Unlock() + + s.scheduleTimers(ap, 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, 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"), + internalAdvance.Milliseconds(), + 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"), + ) + + remainSecond := int(triggerTime.Sub(serverTime).Seconds()) + if remainSecond < 0 { + remainSecond = 0 + } + + return types.ScheduleResult{ + 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"), + RemainSecond: remainSecond, + OffsetMs: calibration.Offset.Milliseconds(), + ErrorBoundMs: calibration.ErrorBound.Milliseconds(), + RttMs: calibration.RTT.Milliseconds(), + }, nil +} + +func (s *Scheduler) scheduleTimers(accountProducts map[string]string, 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() + gen := atomic.AddInt64(&s.warmupGen, 1) + s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() { + if atomic.LoadInt64(&s.warmupGen) != gen { + 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) + 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 +} + +func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration { + cli := s.clientSnapshot() + calibration, err := cli.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait) + 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", + 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 + } +} diff --git a/internal/seckill/scheduler.go b/internal/seckill/scheduler.go index 96ab4ff..10ffe33 100644 --- a/internal/seckill/scheduler.go +++ b/internal/seckill/scheduler.go @@ -1,12 +1,10 @@ package seckill import ( - "errors" "fmt" "math/rand" "strings" "sync" - "sync/atomic" "time" "github.com/golang/freetype/truetype" @@ -49,6 +47,8 @@ type runSettings struct { } // Scheduler 抢购调度器。 +// 具体职责拆分到同包文件：attempt.go（兑换执行）、warmup.go（预热）、 +// schedule_timer.go（定时抢购）、account_store.go（账号管理）。 type Scheduler struct { clientMu sync.RWMutex client *client.DouyuClient @@ -221,6 +221,8 @@ func normalizeSeckillMode(mode types.SeckillMode) types.SeckillMode { } } +// ==================== 账号转发（AccountStore） ==================== + // LoadAccounts 从账号文件加载账号。 func (s *Scheduler) LoadAccounts() { s.accounts.Load() @@ -246,6 +248,8 @@ func (s *Scheduler) AccountCookieByName(name string) (string, bool) { return s.accounts.CookieByName(name) } +// ==================== 结果管理 ==================== + // Results 返回当前抢购结果快照。 func (s *Scheduler) Results() map[string]string { s.resultsMu.RLock() @@ -270,6 +274,8 @@ func (s *Scheduler) setResult(key, value string) { s.results[key] = value } +// ==================== 运行控制 ==================== + // TryStartRun 尝试开始抢购，返回 false 表示已有任务在运行。 func (s *Scheduler) TryStartRun() bool { s.runMu.Lock() @@ -537,6 +543,8 @@ func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt, settings runS return outcome } +// ==================== 辅助函数 ==================== + // modeName 返回抢购模式名称。 func modeName(mode types.SeckillMode) string { if mode == types.SeckillModeAggressive { @@ -651,491 +659,3 @@ func (s *Scheduler) enqueueSuccessImage(account types.Account, product types.Pro } s.successImages.Enqueue(account, product, roleName) } - -func warmupKey(attempt types.ExchangeAttempt) string { - return fmt.Sprintf("%s|%s", attempt.Account.Name, attempt.Product.CommodityID) -} - -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, - Qualified: warmup.Qualified, - LastCheck: time.Now(), - }) - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "从预热初始化缓存: 账号=%s", attempt.Account.Name) - } - } -} - -func (s *Scheduler) setAttemptCache(accountName string, cache *types.AttemptCache) { - s.attemptCacheMu.Lock() - defer s.attemptCacheMu.Unlock() - if cache == nil { - delete(s.attemptCache, accountName) - return - } - 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) warmupAttempts(attempts []types.ExchangeAttempt) { - s.warmupRunMu.Lock() - defer s.warmupRunMu.Unlock() - - attempts = uniqueWarmupAttempts(attempts) - if len(attempts) == 0 { - return - } - s.clearWarmups() - logger.LogEvent("预热", "info", "开始预热: %d 个请求", len(attempts)) - - settings := s.settingsSnapshot() - cli := s.clientSnapshot() - limit := make(chan struct{}, settings.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 := cli.GenerateCSRF(attempt.Account.Cookie) - if err != nil { - logger.LogAccountEvent("预热", "warn", attempt.Account.Name, "CSRF预热失败: 账号=%s 商品=%s err=%v", attempt.Account.Name, attempt.Product.CommodityName, err) - return - } - roleName, err := cli.RoleName(attempt.Account.Cookie) - if err != nil { - logger.LogAccountEvent("预热", "warn", attempt.Account.Name, "角色预热失败: 账号=%s err=%v", attempt.Account.Name, err) - } - s.setWarmup(warmupKey(attempt), types.WarmupData{ - CSRF: csrf, - RoleName: roleName, - Qualified: true, - CreatedAt: time.Now(), - }) - logger.LogAccountEvent("预热", "success", attempt.Account.Name, "预热完成: 账号=%s 商品=%s 耗时=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(start))) - }(attempt) - } - wg.Wait() - logger.LogEvent("预热", "info", "预热任务结束") -} - -func uniqueWarmupAttempts(attempts []types.ExchangeAttempt) []types.ExchangeAttempt { - if len(attempts) <= 1 { - return attempts - } - seen := make(map[string]struct{}, len(attempts)) - out := make([]types.ExchangeAttempt, 0, len(attempts)) - for _, attempt := range attempts { - key := warmupKey(attempt) - if _, ok := seen[key]; ok { - continue - } - seen[key] = struct{}{} - out = append(out, attempt) - } - return out -} - -func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration { - cli := s.clientSnapshot() - calibration, err := cli.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait) - 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", - 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 { - cli := s.clientSnapshot() - result := types.ExchangeResult{ - Account: attempt.Account.Name, - Product: attempt.Product.CommodityName, - Attempt: attempt.Index, - } - - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "执行资格预检链路: 账号=%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.LogAccountEvent("抢购", "info", attempt.Account.Name, "使用预热CSRF: 账号=%s 商品=%s 预热距今=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(warmup.CreatedAt))) - } - if csrf == "" { - var err error - csrf, err = cli.GenerateCSRF(attempt.Account.Cookie) - if err != nil { - result.Message = "失败|生成CSRF失败: " + err.Error() - return result - } - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "CSRF已生成，本次兑换链路复用同一个 token: 账号=%s 商品=%s", attempt.Account.Name, attempt.Product.CommodityName) - } - - limit, err := cli.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.LogAccountEvent("抢购", "info", attempt.Account.Name, "批量限制通过: 账号=%s 商品=%s limit=%d", attempt.Account.Name, attempt.Product.CommodityName, limit) - - if err := cli.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf); err != nil { - result.Message = "失败|预兑换失败: " + err.Error() - return result - } - s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ - CSRF: csrf, - RoleName: roleName, - Qualified: true, - LastCheck: time.Now(), - }) - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "资格预检通过: 账号=%s 商品=%s，后续可直兑", attempt.Account.Name, attempt.Product.CommodityName) - - exchangeInfo, err := cli.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf) - if err != nil { - s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ - CSRF: csrf, - RoleName: roleName, - Qualified: true, - LastCheck: time.Now(), - }) - result.RoleName = roleName - result.Message = "失败|" + err.Error() - return result - } - - if roleName == "" { - roleName, err = cli.RoleName(attempt.Account.Cookie) - if err != nil { - logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "角色名查询失败: 账号=%s err=%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, - Qualified: true, - LastCheck: time.Now(), - }) - - return result -} - -// runAttemptFast 激进模式的兑换逻辑，复用缓存的 CSRF，跳过 BatchExchangeLimit 和 PreExchange。 -func (s *Scheduler) runAttemptFast(attempt types.ExchangeAttempt) types.ExchangeResult { - cli := s.clientSnapshot() - result := types.ExchangeResult{ - Account: attempt.Account.Name, - Product: attempt.Product.CommodityName, - Attempt: attempt.Index, - } - - cache := s.getAttemptCache(attempt.Account.Name) - if cache == nil || cache.CSRF == "" || time.Since(cache.LastCheck) >= qualifiedMaxAge { - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "激进模式执行资格预检: 账号=%s 原因=无缓存CSRF", attempt.Account.Name) - return s.runAttempt(attempt) - } - if !cache.Qualified { - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "激进模式执行资格预检: 账号=%s 原因=未完成资格预检", attempt.Account.Name) - return s.runAttempt(attempt) - } - - csrf := cache.CSRF - logger.LogAccountEvent("抢购", "info", attempt.Account.Name, "激进模式直兑: 账号=%s 商品=%s 第%d包", attempt.Account.Name, attempt.Product.CommodityName, attempt.Index) - - // 直接调用 Exchange - exchangeInfo, err := cli.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf) - if err != nil { - s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ - CSRF: csrf, - RoleName: cache.RoleName, - Qualified: true, - LastCheck: time.Now(), - }) - result.RoleName = cache.RoleName - errMsg := err.Error() - if shouldFallbackStandard(errMsg) { - logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "激进模式回退标准: 账号=%s 原因=%s", attempt.Account.Name, utils.TrimForLog(errMsg, 80)) - s.setAttemptCache(attempt.Account.Name, nil) // 清空缓存 - return s.runAttempt(attempt) - } - result.Message = "失败|" + errMsg - return result - } - - // 获取角色名 - roleName := "" - if cache != nil { - roleName = cache.RoleName - } - if roleName == "" { - var err error - roleName, err = cli.RoleName(attempt.Account.Cookie) - if err != nil { - logger.LogAccountEvent("抢购", "warn", attempt.Account.Name, "角色名查询失败: 账号=%s err=%v", attempt.Account.Name, err) - } - } - s.setAttemptCache(attempt.Account.Name, &types.AttemptCache{ - CSRF: csrf, - RoleName: roleName, - Qualified: true, - LastCheck: time.Now(), - }) - - result.OK = true - result.RoleName = roleName - if roleName != "" { - result.Message = "成功|" + roleName + "|" + exchangeInfo - } else { - result.Message = "成功|" + exchangeInfo - } - return result -} - -func shouldFallbackStandard(errMsg string) bool { - lower := strings.ToLower(errMsg) - switch { - case strings.Contains(lower, "csrf"): - return true - case strings.Contains(lower, "cvl_csrf_token"): - return true - case strings.Contains(lower, "preexchange"): - return true - case strings.Contains(lower, "limit"): - return true - case strings.Contains(errMsg, "预兑换"): - return true - case strings.Contains(errMsg, "限制检查"): - return true - default: - return false - } -} - -// ScheduleRun 注册一次定时抢购。 -func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (types.ScheduleResult, error) { - if len(req.AccountProducts) == 0 { - return types.ScheduleResult{}, errors.New("请选择账号和商品") - } - if strings.TrimSpace(req.TargetTime) == "" { - return types.ScheduleResult{}, errors.New("请设置目标时间") - } - - targetLocal, err := time.ParseInLocation("2006-01-02 15:04:05", req.TargetTime, constants.CST) - if err != nil { - return types.ScheduleResult{}, fmt.Errorf("时间格式错误: %w", err) - } - targetTime := targetLocal.UTC() - - calibration := s.calibrateDouyuTime("定时") - serverTime := calibration.EstimatedServerNow() - - triggerTime := targetTime.Add(-internalAdvance) - remain := triggerTime.Sub(serverTime) - if remain <= 0 { - return types.ScheduleResult{}, errors.New("目标时间已过") - } - - ap := utils.CloneStringMap(req.AccountProducts) - s.scheduleMu.Lock() - s.stopSchedulesLocked() - s.clearWarmups() - s.scheduleMu.Unlock() - - s.scheduleTimers(ap, 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, 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"), - internalAdvance.Milliseconds(), - 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"), - ) - - remainSecond := int(triggerTime.Sub(serverTime).Seconds()) - if remainSecond < 0 { - remainSecond = 0 - } - - return types.ScheduleResult{ - 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"), - RemainSecond: remainSecond, - OffsetMs: calibration.Offset.Milliseconds(), - ErrorBoundMs: calibration.ErrorBound.Milliseconds(), - RttMs: calibration.RTT.Milliseconds(), - }, nil -} - -func (s *Scheduler) scheduleTimers(accountProducts map[string]string, 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() - gen := atomic.AddInt64(&s.warmupGen, 1) - s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() { - if atomic.LoadInt64(&s.warmupGen) != gen { - 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) - 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 -} diff --git a/internal/seckill/warmup.go b/internal/seckill/warmup.go new file mode 100644 index 0000000..2d4e383 --- /dev/null +++ b/internal/seckill/warmup.go @@ -0,0 +1,102 @@ +package seckill + +import ( + "fmt" + "sync" + "time" + + "douyu_go/internal/logger" + "douyu_go/internal/types" + "douyu_go/internal/utils" +) + +func warmupKey(attempt types.ExchangeAttempt) string { + return fmt.Sprintf("%s|%s", attempt.Account.Name, attempt.Product.CommodityID) +} + +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) warmupAttempts(attempts []types.ExchangeAttempt) { + s.warmupRunMu.Lock() + defer s.warmupRunMu.Unlock() + + attempts = uniqueWarmupAttempts(attempts) + if len(attempts) == 0 { + return + } + s.clearWarmups() + logger.LogEvent("预热", "info", "开始预热: %d 个请求", len(attempts)) + + settings := s.settingsSnapshot() + cli := s.clientSnapshot() + limit := make(chan struct{}, settings.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 := cli.GenerateCSRF(attempt.Account.Cookie) + if err != nil { + logger.LogAccountEvent("预热", "warn", attempt.Account.Name, "CSRF预热失败: 账号=%s 商品=%s err=%v", attempt.Account.Name, attempt.Product.CommodityName, err) + return + } + roleName, err := cli.RoleName(attempt.Account.Cookie) + if err != nil { + logger.LogAccountEvent("预热", "warn", attempt.Account.Name, "角色预热失败: 账号=%s err=%v", attempt.Account.Name, err) + } + s.setWarmup(warmupKey(attempt), types.WarmupData{ + CSRF: csrf, + RoleName: roleName, + Qualified: true, + CreatedAt: time.Now(), + }) + logger.LogAccountEvent("预热", "success", attempt.Account.Name, "预热完成: 账号=%s 商品=%s 耗时=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(start))) + }(attempt) + } + wg.Wait() + logger.LogEvent("预热", "info", "预热任务结束") +} + +func uniqueWarmupAttempts(attempts []types.ExchangeAttempt) []types.ExchangeAttempt { + if len(attempts) <= 1 { + return attempts + } + seen := make(map[string]struct{}, len(attempts)) + out := make([]types.ExchangeAttempt, 0, len(attempts)) + for _, attempt := range attempts { + key := warmupKey(attempt) + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + out = append(out, attempt) + } + return out +}