package fwhQueue import ( "context" "encoding/json" "fmt" "fwh-toolbox-server/domain" "fwh-toolbox-server/jobs" "fwh-toolbox-server/repositories" "github.com/hibiken/asynq" "go.mongodb.org/mongo-driver/bson/primitive" "go.uber.org/zap" ) // handleShutdownStatistics 是 Shutdown 统计任务的 Asynq handler func handleShutdownStatistics(ctx context.Context, task *asynq.Task) error { var payload domain.TaskShutdownStatistics if err := json.Unmarshal(task.Payload(), &payload); err != nil { return fmt.Errorf("unmarshal ShutdownStatistics payload: %w", err) } zap.L().Info("[ShutdownStats] 开始执行", zap.String("date", payload.Date)) err := executeShutdownStatistics(ctx, payload) if err != nil { zap.L().Error("[ShutdownStats] 任务执行失败", zap.String("date", payload.Date), zap.String("err", err.Error())) return nil // 业务错误不重试 } zap.L().Info("[ShutdownStats] 任务执行成功", zap.String("date", payload.Date)) return nil } // executeShutdownStatistics 执行 Shutdown 统计 func executeShutdownStatistics(ctx context.Context, payload domain.TaskShutdownStatistics) error { records, err := repositories.QuerySiteShutdownList(payload.Date, 0) if err != nil { return fmt.Errorf("查询 Shutdown 列表失败: %w", err) } // 分析 Shutdown 统计数据 err = jobs.StatisticsShutdown(payload.Date, records, analysisShutdownSiteForNew) if err != nil { return fmt.Errorf("Shutdown 统计分析失败(date=%s): %w", payload.Date, err) } return nil } // analysisShutdownSiteForNew 为 Shutdown 统计触发站点分析 // 复用原有逻辑,通过新队列入队 func analysisShutdownSiteForNew(sn string, date string) (*primitive.ObjectID, error) { // 读取已完成或正在进行的站点分析 queryTimePeriod := [2]string{date, date} queryTrigger := domain.TriggerAuto res, err := repositories.GetSiteAnalysisManageList(1, 1000, &sn, &queryTimePeriod, nil, &queryTrigger) if err != nil { return nil, err } if res == nil { return nil, fmt.Errorf("GetSiteAnalysisManageList 查询结果为空") } if len(res.Data) > 0 { return &res.Data[0].Id, nil } // 创建新的分析任务 writeConfig := domain.SiteAnalysisConfigWrite{ Sn: sn, Description: fmt.Sprintf("Shutdown站点分析 %s", date), TimePeriod: [2]string{date, date}, Options: domain.SiteAnalysisOptions{ EnableTimeline: true, EnableRuntime: true, }, } id, err := repositories.InsertSiteAnalysisConfig(writeConfig, true) if err != nil { return nil, err } // 通过新队列入队站点分析任务 EnqueueSiteAnalysisTask(domain.TaskSiteAnalysis{ Id: id, Sn: writeConfig.Sn, TimePeriod: writeConfig.TimePeriod, Options: writeConfig.Options, }) return &id, nil }