yml/dygou_go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构: 按领域拆分大文件，清理死代码和空目录

Browse Source 
app.go (768行) 拆分为 6 个文件:
- app.go: App struct/辅助方法/NewApp/startup/shutdown/GetVersion
- app_products.go: 商品加载/账号查询
- app_seckill.go: 抢购控制/日志/时间
- app_verify.go: 积分查询/兑换记录复验
- app_export.go: 账号导出/整理
- app_activity.go: 活动切换/配置保存

scheduler.go (1141行) 拆分为 4 个文件:
- scheduler.go: struct/常量/设置/RunNow主循环/executeRound/辅助函数
- attempt.go: runAttempt/runAttemptFast/激进模式缓存
- warmup.go: 预热管理/warmupAttempts
- schedule_timer.go: 定时抢购/时间校准/定时器管理

其他清理:
- CollectFontCandidates 从 config 包移到 image 包（职责归属）
- 删除 logger 死代码: ParseLogLine/ParseLogMeta/DetectLogLevel/
  IsLogLevel/LogMeta/LogDebug/cutBracket 等反解析函数及对应测试
  （日志结构化后不再有生产调用方）
- 删除 static/ 空目录残留
This commit is contained in:
yml2213
2026-07-29 13:19:43 +08:00
parent b72a58fd57
commit 5631f579ea
14 changed files with 1226 additions and 1336 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.go
+1 -598
View File
@@ -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
}
app_activity.go
+90
View File
@@ -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}
}
app_export.go
+175
View File
@@ -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,
}
}
app_products.go
+82
View File
@@ -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
}
app_seckill.go
+74
View File
@@ -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}
}
app_verify.go
+225
View File
@@ -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
}
internal/config/config.go
+2 -40
View File
@@ -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
}
internal/image/font_paths.go
+44
View File
@@ -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
}
internal/logger/logger.go
-132
View File
@@ -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"
}
}
internal/logger/logger_test.go
-76
View File
@@ -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) {
internal/seckill/attempt.go
+229
View File
@@ -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)
}
internal/seckill/schedule_timer.go
+192
View File
@@ -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
}
}
internal/seckill/scheduler.go
+10 -490
View File
@@ -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
}
internal/seckill/warmup.go
+102
View File
@@ -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
}