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

package fwhQueue

import (
	"context"
	"fmt"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/jobs"
	"fwh-toolbox-server/logger"
	"fwh-toolbox-server/repositories"
	"time"

	"go.uber.org/zap"
)

var (
	TaskTypeSiteAnalysis = "task_SiteAnalysis"
)

// type TaskSiteAnalysis = types.TaskSiteAnalysis

func handleTaskSiteAnalysis(ctx context.Context, payload domain.TaskSiteAnalysis) error {
	// 更新运行状态
	fmt.Println("更新任务运行状态")
	if repoErr := repositories.UpdateSiteAnalysisResult(payload.Id, domain.ProgressRunning); repoErr != nil {
		zap.L().Error("更新站点分析结果状态失败(Running)", zap.String("id", payload.Id.String()))
		return repoErr
	}

	fmt.Printf("开始执行任务 %v\n", payload)
	err := executeTaskSiteAnalysis(ctx, payload)
	// 记录任务执行结果
	if err != nil {
		fmt.Printf("任务执行失败 %s\n", err.Error())
		if repoErr := repositories.UpdateSiteAnalysisResult(payload.Id, domain.ProgressError); repoErr != nil {
			zap.L().Error("更新站点分析结果状态失败(Error)", zap.String("id", payload.Id.String()))
			return repoErr
		}
	} else {
		fmt.Println("任务执行成功")
		if repoErr := repositories.UpdateSiteAnalysisResult(payload.Id, domain.ProgressComplete); repoErr != nil {
			zap.L().Error("更新站点分析结果状态失败(Complete)", zap.String("id", payload.Id.String()), zap.String("err", repoErr.Error()))
			return repoErr
		}
	}

	return err
}

var executeTaskSiteAnalysis = func(ctx context.Context, payload domain.TaskSiteAnalysis) error {
	// 转换时间范围，只取用日期部分
	startTime, err := time.Parse(time.DateOnly, payload.TimePeriod[0])
	if err != nil {
		zap.L().Error("StartTime 格式错误", zap.String("time", payload.TimePeriod[0]), zap.String("err", err.Error()))
		return err
	}
	endTime, err := time.Parse(time.DateOnly, payload.TimePeriod[1])
	if err != nil {
		zap.L().Error("EndTime 格式错误", zap.String("time", payload.TimePeriod[1]), zap.String("err", err.Error()))
		return err
	}

	// 将 endTime 调整到当天的 23:59:59
	endTime = endTime.Add(23*time.Hour + 59*time.Minute + 59*time.Second)

	fmt.Println(payload.Options)

	// 创建记录数据库用于存储原始MQTT日志请求数据
	queryRecord, err := repositories.CreateSitesAnalysisRecordTable(string(payload.Id.Hex()))
	if err != nil {
		return err
	}

	if payload.Options.EnableTimeline {
		// 时间线展示功能
		// Site 时间线生成逻辑：
		// 1、分析MQTT日志0和101，获取离线上线状态
		// 2、分析事件日志，获取站点的事件触发记录
		// 3、分析告警日志，获取站点的告警开始-结束触发记录
		// 4、分析MQTT日志357，获取站点模式切换记录
		// 最终处理：获取事件时间点的MQTT日志201数据，关联到事件点。
		// if payload.SiteAnalysisOptions.EnableTimeline

		// 查询MQTT日志0和101
		fmt.Println("查询MQTT日志0")
		logList_0, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_0, startTime, endTime)
		if err != nil {
			zap.L().Error("查询MQTT日志0失败", zap.String("err", err.Error()))
			return err
		}
		for _, l := range logList_0 {
			if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil {
				return err
			}
		}

		fmt.Println("查询MQTT日志101")
		logList_101, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_101, startTime, endTime)
		if err != nil {
			zap.L().Error("查询MQTT日志101失败", zap.String("err", err.Error()))
			return err
		}
		for _, l := range logList_101 {
			if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil {
				return err
			}
		}

		// 查询站点的事件触发记录
		fmt.Println("查询站点的事件触发记录")
		eventLog, err := repositories.QueryEventOrAlarmListOnTimeRange(payload.Sn, startTime, endTime, repositories.Event)
		if err != nil {
			zap.L().Error("查询事件日志失败", zap.String("err", err.Error()))
			return err
		}

		time.Sleep(time.Second) // 延时1s

		// 查询站点的告警开始-结束触发记录
		fmt.Println("查询站点的告警开始-结束触发记录")
		alarmLog, err := repositories.QueryEventOrAlarmListOnTimeRange(payload.Sn, startTime, endTime, repositories.Alarm)
		if err != nil {
			zap.L().Error("查询事件日志失败", zap.String("err", err.Error()))
			return err
		}

		time.Sleep(time.Second) // 延时1s

		// 查询MQTT日志357
		logList_357, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_357, startTime, endTime)
		if err != nil {
			zap.L().Error("查询MQTT日志357失败", zap.String("err", err.Error()))
			return err
		}
		for _, l := range logList_357 {
			if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil {
				return err
			}
		}

		// 分析数据，并将结果写入数据库
		fmt.Println("分析数据，并将结果写入数据库")
		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 {
			zap.L().Error("站点时间线数据分析失败", zap.String("sn", payload.Sn), zap.String("err", err.Error()))
			return err
		}
	}

	if payload.Options.EnableRuntime {
		// 实时数据展示

		// 查询MQTT日志201
		fmt.Println("查询MQTT日志201")
		logList_201, err := repositories.QueryMqttLogListOnTimeRange(payload.Sn, domain.MQTT_LOG_201, startTime, endTime)
		if err != nil {
			zap.L().Error("查询MQTT日志201失败", zap.String("err", err.Error()))
			return err
		}
		for _, l := range logList_201 {
			if err := queryRecord.InsertSitesAnalysisRecord(l); err != nil {
				return err
			}
		}

		// 分析数据，并将结果写入数据库
		fmt.Println("分析数据，并将结果写入数据库")
		err = jobs.AnalysisSiteRuntime(payload.Id, jobs.SiteDataSet{
			Log_201: logList_201,
		})
		if err != nil {
			zap.L().Error("站点运行时数据分析失败", zap.String("sn", payload.Sn), zap.String("err", err.Error()))
			return err
		}
	}

	return nil
}

// 插入站点分析队列
// payload - 任务参数
// auto - 是否是自动插入
func EnqueueSiteAnalysisTask(payload domain.TaskSiteAnalysis, auto bool) error {
	p := TaskPayload[domain.TaskSiteAnalysis]{
		TaskPayloadBase: TaskPayloadBase{
			Type: TaskTypeSiteAnalysis,
		},
		Payload: payload,
	}
	if err := EnqueueTask[domain.TaskSiteAnalysis](p); err != nil {
		logger.Std.Printf("EnqueueSiteAnalysisTask - %s : %s\n", payload.Id, err.Error())
		return err
	}
	return nil
}