FranklinWH-Toolbox/service/task/queue/fwhQueue/fwhTaskSiteAnalysis.go

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
}