package jobs import ( "context" "encoding/json" "errors" "fmt" "fwh-toolbox-server/controllers" "fwh-toolbox-server/ext/api/fwh_pro" function "fwh-toolbox-server/function/SnapFile" "fwh-toolbox-server/logger" "fwh-toolbox-server/models/model_pgsql/model_alarm_snapshot" "fwh-toolbox-server/models/model_pgsql/query_alarm_snapshot" "fwh-toolbox-server/types" "fwh-toolbox-server/utils/files" "os" "path/filepath" "regexp" "strconv" "strings" "time" "go.uber.org/zap" "gorm.io/gorm" "gorm.io/gorm/clause" ) type AlarmSnapshotInfo struct { Id int64 `json:"id"` CodeType int32 `json:"codeType"` // 作用未知 FileName string `json:"fileName"` FileUrl string `json:"fileUrl"` LinkId string `json:"linkId"` // site SN LogTime string `json:"logTime"` // format YYYY-MM-DD SerialNumber string `json:"serialNumber"` StartTime string `json:"startTime"` // format YYYY-MM-DD HH:mm:SS EndTime string `json:"endTime"` // format YYYY-MM-DD HH:mm:SS Status int `json:"status"` // 1-完成 CompleteTime string `json:"completeTime"` // format YYYY-MM-DD HH:mm:SS } type AlarmSnapshotListResponse = types.FwhPageResponse[AlarmSnapshotInfo] type FileRecord = model_alarm_snapshot.SiteAlarmSnapshotFileRecord type SnapshotData = model_alarm_snapshot.SiteAlarmSnapshotData // 解析 func handleAlarmSnapshot(sn string, logTime string, page int) (uint64, []AlarmSnapshotInfo, error) { // 查询告警快照数据 resByte, err := fwh_pro.QueryAlarmSnapshotInfo(sn, logTime, page) if err != nil { return 0, nil, fmt.Errorf("[handleAlarmSnapshot] %s", err.Error()) } var alarmSnapshotRes AlarmSnapshotListResponse // 反序列化(unmarshal) jsonErr := json.Unmarshal(resByte, &alarmSnapshotRes) if jsonErr != nil { return 0, nil, fmt.Errorf("[handleAlarmSnapshot] %s", jsonErr) } return alarmSnapshotRes.Total, alarmSnapshotRes.Result, nil } // 获取告警快照 // 从FHW后端请求快照文件后,处理并放入数据库 // logTime 格式 YYYY-MM-DD func GetAlarmSnapshot(ctx context.Context, sn, logTime string) error { // 请求快照信息列表 var page int = 1 snapshotList := make([]AlarmSnapshotInfo, 0) total, dataList, err := handleAlarmSnapshot(sn, logTime, page) if err != nil { return fmt.Errorf("[GetAlarmSnapshot] %s", err.Error()) } snapshotList = append(snapshotList, dataList...) for uint64(len(snapshotList)) < total { page = page + 1 _, dataList, err = handleAlarmSnapshot(sn, logTime, page) if err != nil { return fmt.Errorf("[GetAlarmSnapshot] %s", err.Error()) } snapshotList = append(snapshotList, dataList...) } // 创建临时目录存储下载和解压数据 tempDir, err := os.MkdirTemp("", "fwhtb-alarmSnapshot-*") if err != nil { logger.Std.Printf("创建临时目录失败: %v\n", err) return err } defer os.RemoveAll(tempDir) // 程序结束时清理临时目录 for _, info := range snapshotList { fileRecordModel := query_alarm_snapshot.SiteAlarmSnapshotFileRecord // 判断文件是否已经导入 _, err := fileRecordModel.WithContext(ctx). Where(fileRecordModel.FileName.Eq(info.FileName)). First() if errors.Is(err, gorm.ErrRecordNotFound) { // 不存在 - 继续运行 } else if err != nil { // 查询出错 logger.Std.Printf("数据库快照文件记录查询失败: %v\n", err) return err } else { // 存在 record continue } // 获取站点信息,获取设备时区信息 res, err := controllers.QuerySiteInfoList(controllers.SiteInfoPara{ SN: &info.LinkId, PageSize: 1, PageNum: 1, }) if err != nil { zap.L().Error("站点信息查询失败。", zap.String("err", err.Error())) return err } // 提取ZoneInfo if len(res.Result) != 1 { zap.L().Error("站点信息查询失败,未查询待站点数据。", zap.Any("res", res)) return fmt.Errorf("站点信息查询失败,未查询待站点数据。") } zoneInfo := res.Result[0].ZoneInfo snapFileName := filepath.Join(tempDir, info.FileName+".tgz") // 下载快照文件 fileUrl := "https://energy.franklinwh.com/hes-upload/v1/file/download/" + info.FileUrl if err = files.DownloadFile(fileUrl, snapFileName); err != nil { zap.L().Error("快照文件下载失败。", zap.String("err", err.Error())) return err } // 创建解压目标文件夹 snapDir := filepath.Join(tempDir, fmt.Sprintf("AlarmSnapshot_%d", info.Id)) if err = files.CheckAndCreateDir(snapDir); err != nil { zap.L().Error("文件夹创建失败", zap.String("err", err.Error())) return err } // 解压包文件 if err = files.AutoExtract(snapFileName, snapDir); err != nil { zap.L().Error("解压快照包失败", zap.String("err", err.Error())) return err } startTime, err := time.Parse(time.DateTime, info.StartTime) if err != nil { zap.L().Error("StartTime格式错误", zap.String("err", err.Error())) return err } endTime, err := time.Parse(time.DateTime, info.EndTime) if err != nil { zap.L().Error("EndTime格式错误", zap.String("err", err.Error())) return err } completeTime, err := time.Parse(time.DateTime, info.CompleteTime) if err != nil { zap.L().Error("CompleteTime格式错误", zap.String("err", err.Error())) return err } logTime, err := time.Parse(time.DateOnly, info.LogTime) if err != nil { zap.L().Error("LogTime格式错误", zap.String("err", err.Error())) return err } fileRecord := FileRecord{ CodeType: info.CodeType, FileName: info.FileName, FileURL: info.FileUrl, Sn: info.LinkId, LogTime: logTime, StartTime: startTime, EndTime: endTime, CompleteTime: completeTime, State: 0, } // 创建文件解析记录(冲突时覆盖) err = fileRecordModel.WithContext(ctx).Clauses(clause.OnConflict{ UpdateAll: true, }).Create(&fileRecord) if err != nil { zap.L().Error("文件解析记录数据库创建失败", zap.String("err", err.Error())) return err } // 遍历文件 var fileList []string if fileList, err = files.ListFiles(snapDir, true); err != nil { zap.L().Error("遍历文件失败", zap.String("err", err.Error())) return err } // 包文件有两个版本: // 1 - 包中文件内容为字符串,直接读取即可 // 2 - 包中数据为16进制数据,须配合VID/PID表解析 // 解析文件并写入数据库 if err = handleAlarmSnapshotFiles(ctx, fileRecord, fileList, zoneInfo); err != nil { zap.L().Error("解析文件或写入数据库失败", zap.String("err", err.Error())) return err } } return nil } // handleAlarmSnapshotFiles 遍历文件列表并处理符合格式的文件 func handleAlarmSnapshotFiles(ctx context.Context, record FileRecord, fileList []string, zoneInfo string) error { // 正则规则 reDigits10 := regexp.MustCompile(`^\d{10}$`) reZip := regexp.MustCompile(`^\d{8}_\d{6}_\d{3}\.zip$`) for _, file := range fileList { // 获取文件名(去掉路径) name := filepath.Base(file) // 处理纯数字长度为10 if reDigits10.MatchString(name) { if err := parserOldAlarmSnapshotFile(ctx, record, file, zoneInfo); err != nil { zap.L().Error("解析OLD告警快照文件失败", zap.String("err", err.Error())) } continue } // 处理 zip 包 (数字+下划线组合) if strings.HasSuffix(name, ".zip") && reZip.MatchString(name) { parserNewAlarmSnapshotFile(ctx, record, file, zoneInfo) continue } // 其他文件忽略 zap.L().Warn("忽略文件", zap.String("fileName", name), zap.String("filePath", file)) } return nil } func parserOldAlarmSnapshotFile(ctx context.Context, record FileRecord, filePath string, zoneInfo string) error { // 获取文件名(去掉路径) name := filepath.Base(filePath) // 读取文件内容 data, err := os.ReadFile(filePath) if err != nil { return fmt.Errorf("读取文件失败 %s: %s", filePath, err.Error()) } if len(data) == 0 { return fmt.Errorf("读取文件内容为空") } // 校验 JSON 格式 var js json.RawMessage if err := json.Unmarshal(data, &js); err != nil { return fmt.Errorf("解析 JSON 失败 %s: %s", filePath, err.Error()) } // 获取设备时区,计算日志记录时间 tsInt, err := strconv.ParseInt(name, 10, 64) if err != nil { return fmt.Errorf("解析 LogTime 失败 %s: %s", filePath, err.Error()) } t := time.Unix(tsInt, 0) // 秒级 // 存入数据库 siteAlarmSnapshotData := SnapshotData{ Sn: record.Sn, FileRecordID: record.ID, LogTime: t, JSONData: string(js), ZoneInfo: zoneInfo, } SnapshotDataModel := query_alarm_snapshot.SiteAlarmSnapshotData err = SnapshotDataModel.WithContext(ctx).Clauses(clause.OnConflict{ UpdateAll: true, }).Create(&siteAlarmSnapshotData) if err != nil { zap.L().Error("文件解析记录数据库创建失败", zap.String("err", err.Error())) return err } return nil } func parserNewAlarmSnapshotFile(ctx context.Context, record FileRecord, filePath string, zoneInfo string) error { // 获取文件名(去掉路径) name := filepath.Base(filePath) // 生成文件夹路径 destPath := strings.TrimSuffix(filePath, filepath.Ext(name)) if err := files.Unzip(filePath, destPath); err != nil { zap.L().Error("zip文件解压失败", zap.String("err", err.Error())) return err } // 遍历文件 fileList, err := files.ListFiles(destPath, true) if err != nil { zap.L().Error("遍历文件失败", zap.String("err", err.Error())) return err } if len(fileList) != 1 { zap.L().Error("zip文件内容不正确", zap.String("err", err.Error())) return err } snapFilePath := fileList[0] parser := function.NewParserAlarmSnapshot(snapFilePath) if err := parser.ParserSnapFile(); err != nil { zap.L().Error("告警快照文件解析失败", zap.String("err", err.Error()), zap.String("filePath", snapFilePath)) return err } for _, d := range parser.SnapJson { // 转 JSON 字符串 dataStr, err := json.Marshal(d) if err != nil { return fmt.Errorf("告警快照数据转 JSON 失败 %s: %s", filePath, err.Error()) } // 存入数据库 siteAlarmSnapshotData := SnapshotData{ Sn: record.Sn, FileRecordID: record.ID, LogTime: time.Unix(int64(d.TimeStamp), 0), JSONData: string(dataStr), ZoneInfo: zoneInfo, } SnapshotDataModel := query_alarm_snapshot.SiteAlarmSnapshotData err = SnapshotDataModel.WithContext(ctx).Clauses(clause.OnConflict{ UpdateAll: true, }).Create(&siteAlarmSnapshotData) if err != nil { zap.L().Error("文件解析记录数据库创建失败", zap.String("err", err.Error())) return err } } return nil }