package fwhQueue import ( "context" "encoding/json" "fmt" "fwh-toolbox-server/domain" "fwh-toolbox-server/jobs" "fwh-toolbox-server/repositories" "time" "github.com/hibiken/asynq" "go.uber.org/zap" ) // handleSiteAnalysis 是站点分析任务的 Asynq handler func handleSiteAnalysis(ctx context.Context, task *asynq.Task) error { var payload domain.TaskSiteAnalysis if err := json.Unmarshal(task.Payload(), &payload); err != nil { return fmt.Errorf("unmarshal SiteAnalysis payload: %w", err) } taskId := payload.Id.Hex() // 更新运行状态 if err := repositories.UpdateSiteAnalysisResult(payload.Id, domain.ProgressRunning); err != nil { zap.L().Error("[SiteAnalysis] 更新状态失败(Running)", zap.String("id", taskId)) return err } // 执行分析逻辑 err := executeSiteAnalysis(ctx, payload) if err != nil { zap.L().Error("[SiteAnalysis] 任务执行失败", zap.String("id", taskId), zap.String("err", err.Error())) repositories.UpdateSiteAnalysisResult(payload.Id, domain.ProgressError) return nil // 返回 nil 避免 Asynq 重试(业务错误不需要重试) } zap.L().Info("[SiteAnalysis] 任务执行成功", zap.String("id", taskId)) repositories.UpdateSiteAnalysisResult(payload.Id, domain.ProgressComplete) return nil } // executeSiteAnalysis 执行站点分析 func executeSiteAnalysis(ctx context.Context, payload domain.TaskSiteAnalysis) error { // 解析时间范围 startTime, err := time.Parse(time.DateOnly, payload.TimePeriod[0]) if err != nil { return fmt.Errorf("startTime 格式错误: %w", err) } endTime, err := time.Parse(time.DateOnly, payload.TimePeriod[1]) if err != nil { return fmt.Errorf("endTime 格式错误: %w", err) } endTime = endTime.Add(23*time.Hour + 59*time.Minute + 59*time.Second) // 创建记录数据库用于存储原始 MQTT 日志请求数据 queryRecord, err := repositories.CreateSitesAnalysisRecordTable(payload.Id.Hex()) if err != nil { return err } if payload.Options.EnableTimeline { // 查询 MQTT 日志 0 logList_0, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_0, startTime, endTime) if err != nil { return fmt.Errorf("查询MQTT日志0失败: %w", err) } for _, l := range logList_0 { if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil { return err } } // 查询 MQTT 日志 101 logList_101, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_101, startTime, endTime) if err != nil { return fmt.Errorf("查询MQTT日志101失败: %w", err) } for _, l := range logList_101 { if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil { return err } } // 查询事件触发记录 eventLog, err := repositories.QueryEventOrAlarmListOnTimeRange(payload.Sn, startTime, endTime, repositories.Event) if err != nil { return fmt.Errorf("查询事件日志失败: %w", err) } time.Sleep(time.Second) // 查询告警触发记录 alarmLog, err := repositories.QueryEventOrAlarmListOnTimeRange(payload.Sn, startTime, endTime, repositories.Alarm) if err != nil { return fmt.Errorf("查询告警日志失败: %w", err) } time.Sleep(time.Second) // 查询 MQTT 日志 357 logList_357, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_357, startTime, endTime) if err != nil { return fmt.Errorf("查询MQTT日志357失败: %w", err) } for _, l := range logList_357 { if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil { return err } } // 分析时间线数据 err = jobs.AnalysisSiteTimeline(payload.Id, jobs.SiteDataSet{ Log_0: logList_0, Log_101: logList_101, Log_357: logList_357, EventLog: eventLog, AlarmLog: alarmLog, }) if err != nil { return fmt.Errorf("站点时间线分析失败(sn=%s): %w", payload.Sn, err) } } if payload.Options.EnableRuntime { // 查询 MQTT 日志 201 logList_201, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_201, startTime, endTime) if err != nil { return fmt.Errorf("查询MQTT日志201失败: %w", err) } for _, l := range logList_201 { if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil { return err } } // 分析运行时数据 err = jobs.AnalysisSiteRuntime(payload.Id, jobs.SiteDataSet{ Log_201: logList_201, }) if err != nil { return fmt.Errorf("站点运行时数据分析失败(sn=%s): %w", payload.Sn, err) } } return nil }