callFwhFile.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698
  1. package jobs
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "fwh-toolbox-server/cache"
  8. "fwh-toolbox-server/domain"
  9. "fwh-toolbox-server/logger"
  10. "fwh-toolbox-server/repositories"
  11. "path"
  12. "strconv"
  13. "strings"
  14. "sync"
  15. "time"
  16. "github.com/redis/go-redis/v9"
  17. "go.mongodb.org/mongo-driver/bson/primitive"
  18. "go.uber.org/zap"
  19. )
  20. type TaskQueueMutex struct {
  21. running bool
  22. mu sync.Mutex
  23. }
  24. type EmsLogType = string
  25. var (
  26. EmsRunLog = "EmsRunLog" // 4-IBG运行日志
  27. EmsBattLog = "EmsBattLog" // 7-电池日志
  28. EmsMonitorLog = "EmsMonitorLog" // 9-设备系统资源监控日志
  29. Ems4gLog = "Ems4gLog" // 10-4G日志
  30. EmsAwsLog = "EmsAwsLog" // 11-AWS日志
  31. EmsMainLog = "EmsMainLog" // 12-主进程日志
  32. )
  33. // EMS日志类型map
  34. var mapEmsLogType = map[string]int{
  35. "EmsRunLog": 4,
  36. "EmsBattLog": 7,
  37. "EmsMonitorLog": 9,
  38. "Ems4gLog": 10,
  39. "EmsAwsLog": 11,
  40. "EmsMainLog": 12,
  41. }
  42. var (
  43. callFwhFileCtx = context.Background()
  44. streamNamePrefix = "fwhtoolbox:CallFwhFile:"
  45. groupName = "task_queue"
  46. consumerID = "worker-taskQueue"
  47. // 防止队列任务函数重复启动
  48. taskQueueMutex = map[string]*TaskQueueMutex{}
  49. taskQueueMutexMu sync.RWMutex // 用于保护 map
  50. // 最大召唤次数
  51. maxCount = 3
  52. )
  53. // 召唤文件类型
  54. var (
  55. FileTypeEmsLog = "EmsLog"
  56. FileTypeFaultRecord = "FaultRecord"
  57. )
  58. type CallFileTaskStreamInfo struct {
  59. Ctx context.Context
  60. StreamName string
  61. GroupName string
  62. ID string
  63. }
  64. type CallFileTaskInfo struct {
  65. MainTaskId primitive.ObjectID `bson:"mainTaskId" json:"mainTaskId"`
  66. }
  67. func CreateCallFwhFileTask(mainTaskId primitive.ObjectID, taskList []domain.BatchDownloadFileTaskInfo) error {
  68. // 筛选SN,不同SN可并行执行
  69. sortTaskList := map[string][]domain.BatchDownloadFileTaskInfo{}
  70. for _, task := range taskList {
  71. sortTaskList[task.Sn] = append(sortTaskList[task.Sn], task)
  72. }
  73. taskRecordList := map[string]domain.BatchDownloadFileTaskInfo{}
  74. for sn, payloadList := range sortTaskList {
  75. for _, payload := range payloadList {
  76. // 确保消费组存在(忽略已存在错误)
  77. _ = cache.XGroupCreate(callFwhFileCtx, streamNamePrefix+sn, groupName, "$")
  78. id, err := enqueueTask(sn, CallFileTaskInfo{
  79. MainTaskId: mainTaskId,
  80. }, payload)
  81. if err != nil {
  82. zap.L().Error("创建召唤FWH文件任务队列失败")
  83. return err
  84. }
  85. taskRecordList[id] = payload
  86. }
  87. }
  88. if err := repositories.CreateBatchDownloadCallFileTaskList(mainTaskId.Hex(), taskRecordList); err != nil {
  89. zap.L().Error("插入召唤FWH文件任务队列记录列表失败")
  90. return err
  91. }
  92. // 提取SN list
  93. snList := make([]string, 0, len(sortTaskList))
  94. for sn := range sortTaskList {
  95. snList = append(snList, sn)
  96. }
  97. // 启动任务队列
  98. for _, sn := range snList {
  99. go startTaskQueueWorker(sn)
  100. time.Sleep(time.Second) // 任务队列启动延时,避免过多请求同时发出,将任务错开运行
  101. }
  102. return nil
  103. }
  104. func callFwhFile(mainTaskId primitive.ObjectID, taskId string, info domain.BatchDownloadFileTaskInfo) {
  105. // var reason string
  106. streamName := streamNamePrefix + info.Sn
  107. var success bool
  108. var failReason string
  109. // ✅ defer 确保无论走哪条路径,最终结果一定被更新
  110. defer func() {
  111. if success {
  112. repositories.UpdateBatchDownloadCallFileResult(mainTaskId.Hex(), taskId, 1, "")
  113. // ack stream
  114. ackTask(CallFileTaskStreamInfo{
  115. Ctx: callFwhFileCtx,
  116. StreamName: streamName,
  117. GroupName: groupName,
  118. ID: taskId,
  119. }, true, nil)
  120. } else {
  121. repositories.UpdateBatchDownloadCallFileResult(mainTaskId.Hex(), taskId, -1, failReason)
  122. // ack stream
  123. ackTask(CallFileTaskStreamInfo{
  124. Ctx: callFwhFileCtx,
  125. StreamName: streamName,
  126. GroupName: groupName,
  127. ID: taskId,
  128. }, false, errors.New(failReason))
  129. }
  130. }()
  131. var emsDeadline time.Time
  132. var emsSlot *EmsLogCallSlot
  133. var emsCtx context.Context
  134. if info.FileType == FileTypeEmsLog {
  135. slot, err := AcquireEmsLogCallSlot(callFwhFileCtx, info.Sn)
  136. if err != nil {
  137. failReason = fmt.Sprintf("等待 EMS 日志召唤名额失败: %s", err)
  138. return
  139. }
  140. emsSlot = slot
  141. defer slot.Release()
  142. emsDeadline = time.Now().Add(EmsLogCallTimeout)
  143. var cancel context.CancelFunc
  144. emsCtx, cancel = context.WithDeadline(slot.Context(), emsDeadline)
  145. defer cancel()
  146. }
  147. for count := 0; count < maxCount; count++ {
  148. // 请求召唤
  149. switch info.FileType {
  150. case FileTypeEmsLog: // 召唤EMS日志
  151. if emsSlot.Context().Err() != nil {
  152. failReason = ErrEmsLogCallSlotLost.Error()
  153. return
  154. }
  155. if time.Until(emsDeadline) <= 0 {
  156. failReason = ErrEmsLogCallTimeout.Error()
  157. return
  158. }
  159. logType, ok := mapEmsLogType[info.FileSupType]
  160. if !ok {
  161. failReason = fmt.Sprintf("未知 EMS 日志类型: %s", info.FileSupType)
  162. return
  163. }
  164. baselineID, baselineErr := repositories.LatestEmsLogCallRecordIDContext(emsCtx, info.Sn, info.Date, logType)
  165. if time.Until(emsDeadline) <= 0 {
  166. failReason = ErrEmsLogCallTimeout.Error()
  167. return
  168. }
  169. if baselineErr != nil {
  170. failReason = fmt.Sprintf("查询 EMS 日志召唤基线失败: %s", baselineErr)
  171. return
  172. }
  173. fmt.Printf("callEmsLog SN-%s Date-%s Type-%s\n", info.Sn, info.Date, info.FileSupType)
  174. err := callEmsLog(emsCtx, info.Sn, info.Date, info.FileSupType)
  175. if time.Until(emsDeadline) <= 0 {
  176. failReason = ErrEmsLogCallTimeout.Error()
  177. return
  178. }
  179. if err != nil {
  180. success = false
  181. failReason = fmt.Sprintf("handle task CallEmsLog error - %s", err.Error())
  182. return
  183. }
  184. if emsSlot.Context().Err() != nil {
  185. failReason = ErrEmsLogCallSlotLost.Error()
  186. return
  187. }
  188. ok, err = waitUntilDone(func() (bool, error) {
  189. // 查询召唤结果
  190. for count := 0; count < maxCount; count++ {
  191. if emsSlot.Context().Err() != nil {
  192. return false, ErrEmsLogCallSlotLost
  193. }
  194. if time.Until(emsDeadline) <= 0 {
  195. return false, ErrEmsLogCallTimeout
  196. }
  197. fmt.Printf("getEmsLogUrl SN-%s Date-%s Type-%s\n", info.Sn, info.Date, info.FileSupType)
  198. recordInfo, err := getEmsLogUrlAfter(emsCtx, info.Sn, info.Date, info.FileSupType, baselineID)
  199. if err != nil {
  200. zap.L().Error("getEmsLogUrl读取EMS日志地址失败")
  201. continue
  202. }
  203. if recordInfo.Status == -1 {
  204. zap.L().Error("EMS日志召唤失败")
  205. return false, fmt.Errorf("EMS日志召唤失败")
  206. } else if recordInfo.Status == 1 {
  207. // 下载文件
  208. var downloadCount int = 0
  209. for downloadCount = 0; downloadCount < maxCount; downloadCount++ {
  210. dataBytes, err := repositories.QueryDownloadEmsLogFileContext(emsCtx, recordInfo.FileUrl)
  211. if err != nil {
  212. zap.L().Error("EMS日志文件下载失败", zap.String("err", err.Error()))
  213. continue
  214. }
  215. // 提取文件后缀
  216. baseStr := path.Base(recordInfo.FileUrl) // 取最后的文件名部分
  217. if baseStr == "." || baseStr == "/" {
  218. baseStr = ""
  219. }
  220. // 加载加州时区(Pacific Time)
  221. loc, err := time.LoadLocation("America/Los_Angeles")
  222. if err != nil {
  223. zap.L().Error("Ems日志LoadLocation设置失败", zap.String("callTime", recordInfo.CallTime), zap.String("err", err.Error()))
  224. return false, err
  225. }
  226. // 按指定时区解析时间,生效召唤时间的时间戳
  227. callTime, err := time.ParseInLocation(time.DateTime, recordInfo.CallTime, loc)
  228. if err != nil {
  229. zap.L().Error("Ems日志CallTime解析失败", zap.String("callTime", recordInfo.CallTime), zap.String("err", err.Error()))
  230. return false, err
  231. }
  232. // 保存数据到文件服务器
  233. objName, err := SaveBatchDownloadEmsLog(info.Sn, dataBytes, baseStr, info.FileSupType, info.Date, callTime)
  234. if err != nil {
  235. zap.L().Error("Ems日志文件保存失败", zap.String("devType", strconv.Itoa(info.DevType)), zap.String("sn", info.Sn), zap.String("err", err.Error()))
  236. return false, fmt.Errorf("Ems日志文件保存失败")
  237. }
  238. if err := repositories.InsertBatchDownloadPathEmsLogs(mainTaskId, info.Sn, info.Date, info.FileSupType, objName); err != nil {
  239. zap.L().Error("BatchDownload - EmsLog文件地址写入失败", zap.String("sn", info.Sn), zap.String("date", info.Date), zap.String("objName", objName), zap.String("err", err.Error()))
  240. return false, fmt.Errorf("BatchDownload - EmsLog文件地址写入失败")
  241. }
  242. break
  243. }
  244. // 超过最大尝试次数
  245. if downloadCount >= maxCount {
  246. return false, fmt.Errorf("EMS日志文件下载失败,超过最大尝试次数:%d", maxCount)
  247. }
  248. return true, nil
  249. }
  250. return false, nil
  251. }
  252. return false, fmt.Errorf("读取EMS日志地址请求失败")
  253. }, time.Minute, time.Until(emsDeadline)) // 单个文件共用 30 分钟期限
  254. if err != nil {
  255. success = false
  256. failReason = err.Error() // 保存失败原因
  257. }
  258. if time.Until(emsDeadline) <= 0 {
  259. failReason = ErrEmsLogCallTimeout.Error()
  260. return // 超时后处理同一 SN 的下一个文件,不再召唤本文件
  261. }
  262. if emsSlot.Context().Err() != nil {
  263. failReason = ErrEmsLogCallSlotLost.Error()
  264. return
  265. }
  266. if ok {
  267. // 召唤成功
  268. success = true
  269. failReason = ""
  270. return
  271. } else {
  272. zap.L().Warn("Call FWH file失败, 再次尝试", zap.String("count", strconv.Itoa(count)))
  273. }
  274. case FileTypeFaultRecord: // 召唤故障录波
  275. timeRange := strings.Split(info.Date, ",")
  276. if len(timeRange) != 2 {
  277. // 召唤失败
  278. success = false
  279. failReason = fmt.Sprintf("handle task CallFaultRecord timeRange error - timeRange: %s", info.Date)
  280. return
  281. }
  282. // 提取下载记录的indexList
  283. indexList := []int{}
  284. indexStrList := strings.Split(info.FileSupType, ",")
  285. for _, indexStr := range indexStrList {
  286. idx, err := strconv.Atoi(indexStr)
  287. if err != nil {
  288. // 召唤失败
  289. success = false
  290. failReason = fmt.Sprintf("handle task CallFaultRecord indexStrList error - indexStrList: %s - err: %s", info.FileSupType, err.Error())
  291. return
  292. }
  293. indexList = append(indexList, idx)
  294. }
  295. // 召唤故障录波数据
  296. err := callFaultRecord(info.DevType, info.Sn, timeRange[0], timeRange[1], indexList)
  297. if err != nil {
  298. // 召唤失败
  299. success = false
  300. failReason = fmt.Sprintf("handle task CallFaultRecord error - err: %s", err.Error())
  301. return
  302. }
  303. ok, err := waitUntilDone(func() (bool, error) {
  304. // 查询召唤结果
  305. for count := 0; count < maxCount; count++ {
  306. status, faultId, reasonStr, err := getFaultRecordUrl(info.DevType, info.Sn, timeRange[0], timeRange[1])
  307. if err != nil {
  308. zap.L().Error("getFaultRecordUrl读取故障录波地址失败", zap.String("err", err.Error()))
  309. continue
  310. }
  311. if status == -1 {
  312. zap.L().Error("故障录波召唤失败", zap.String("reason", reasonStr))
  313. return false, fmt.Errorf("故障录波召唤失败 - %s", reasonStr)
  314. } else if status == 2 {
  315. // 查询子记录
  316. subRecordList, err := repositories.QueryFaultRecordSubList(faultId)
  317. if err != nil {
  318. zap.L().Error("故障录波子记录查询失败", zap.String("err", err.Error()))
  319. return false, fmt.Errorf("故障录波子记录查询失败 - %w", err)
  320. }
  321. // 下载所有自己录中的文件
  322. for _, subRecord := range subRecordList {
  323. var downloadCount int = 0
  324. for downloadCount = 0; downloadCount < maxCount; downloadCount++ {
  325. _, dataBytes, err := repositories.QueryDownloadFaultRecordData(faultId, subRecord.Id)
  326. if err != nil {
  327. zap.L().Error("故障录波记录文件下载失败", zap.String("devType", strconv.Itoa(subRecord.DeviceType)), zap.String("sn", subRecord.GatewayId), zap.String("faultId", strconv.FormatInt(subRecord.FaultId, 10)), zap.String("id", strconv.FormatInt(subRecord.Id, 10)), zap.String("err", err.Error()))
  328. continue
  329. }
  330. // 保存数据到文件服务器
  331. objName, err := SaveBatchDownloadFaultRecord(subRecord.GatewayId, dataBytes, subRecord.FileName+".xls")
  332. if err != nil {
  333. zap.L().Error("故障录波记录文件保存失败", zap.String("devType", strconv.Itoa(subRecord.DeviceType)), zap.String("sn", subRecord.GatewayId), zap.String("faultId", strconv.FormatInt(subRecord.FaultId, 10)), zap.String("id", strconv.FormatInt(subRecord.Id, 10)), zap.String("err", err.Error()))
  334. return false, fmt.Errorf("故障录波记录文件保存失败")
  335. }
  336. if err := repositories.InsertBatchDownloadPathFaultRecord(mainTaskId, subRecord.GatewayId, subRecord.FileName, objName); err != nil {
  337. zap.L().Error("BatchDownload - FaultRecord文件地址写入失败", zap.String("sn", subRecord.GatewayId), zap.String("fileName", subRecord.FileName), zap.String("objName", objName), zap.String("err", err.Error()))
  338. return false, fmt.Errorf("BatchDownload - FaultRecord文件地址写入失败")
  339. }
  340. break
  341. }
  342. // 超过最大尝试次数
  343. if downloadCount >= maxCount {
  344. return false, fmt.Errorf("故障录波记录文件下载失败,超过最大尝试次数:%d", maxCount)
  345. }
  346. }
  347. return true, nil
  348. }
  349. return false, nil
  350. }
  351. return false, fmt.Errorf("读取故障录波地址请求失败")
  352. }, time.Minute, 30*time.Minute) // 每分钟查看,半小时超时
  353. if err != nil {
  354. success = false
  355. failReason = err.Error() // 保存失败原因
  356. }
  357. if ok {
  358. // 召唤成功
  359. success = true
  360. failReason = ""
  361. return
  362. } else {
  363. zap.L().Warn("Call FWH file失败, 再次尝试", zap.String("count", strconv.Itoa(count)))
  364. }
  365. default:
  366. logger.Std.Println("Unknown call file type:", info.FileType)
  367. }
  368. }
  369. }
  370. /************** 任务轮询 **************/
  371. func waitUntilDone(check func() (bool, error), interval time.Duration, timeout time.Duration) (bool, error) {
  372. ticker := time.NewTicker(interval)
  373. defer ticker.Stop()
  374. timeoutChan := time.After(timeout)
  375. for {
  376. select {
  377. case <-ticker.C:
  378. ok, err := check()
  379. if err != nil {
  380. zap.L().Debug("❌ 轮询任务失败", zap.String("err", err.Error()))
  381. return false, err
  382. } else if ok {
  383. zap.L().Debug("✅ 轮询任务完成")
  384. return true, nil
  385. } else {
  386. zap.L().Debug("⏳ 任务轮询中...")
  387. }
  388. case <-timeoutChan:
  389. zap.L().Debug("❌ 轮询超时")
  390. return false, fmt.Errorf("召唤结果轮询超时")
  391. }
  392. }
  393. }
  394. /************** 召唤文件任务队列 **************/
  395. func startTaskQueueWorker(sn string) {
  396. taskQueueMutexMu.Lock()
  397. taskMutex, ok := taskQueueMutex[sn]
  398. if !ok || taskMutex == nil {
  399. taskMutex = &TaskQueueMutex{
  400. running: false,
  401. mu: sync.Mutex{},
  402. }
  403. taskQueueMutex[sn] = taskMutex
  404. }
  405. taskQueueMutexMu.Unlock()
  406. taskMutex.mu.Lock()
  407. if taskMutex.running {
  408. taskMutex.mu.Unlock()
  409. zap.L().Info("任务已在运行,跳过执行")
  410. return
  411. }
  412. taskMutex.running = true
  413. taskMutex.mu.Unlock()
  414. streamName := streamNamePrefix + sn
  415. streamsMap := make(map[string]string)
  416. streamsMap[streamName] = ">"
  417. for {
  418. // 处理新任务
  419. res, err := cache.XReadGroup(callFwhFileCtx, groupName, consumerID, streamsMap, 1) // 等待任务
  420. if err != nil {
  421. if err == redis.Nil {
  422. // 未读取到新消息 - 退出
  423. logger.Std.Println("未读取到新消息: ", sn)
  424. taskMutex.mu.Lock()
  425. taskMutex.running = false
  426. taskMutex.mu.Unlock()
  427. cache.Delete(callFwhFileCtx, streamNamePrefix+sn)
  428. return
  429. }
  430. // Redis异常 - 退出
  431. logger.Std.Println("XReadGroup error:", err)
  432. taskMutex.mu.Lock()
  433. taskMutex.running = false
  434. taskMutex.mu.Unlock()
  435. return
  436. }
  437. for _, stream := range res {
  438. for _, message := range stream.Messages {
  439. logger.Std.Println("XReadGroup read message:", message)
  440. handleTaskQueueMsg(sn, message)
  441. }
  442. }
  443. }
  444. }
  445. // 处理单条消息
  446. func handleTaskQueueMsg(sn string, message redis.XMessage) {
  447. id := message.ID
  448. rawPayload := message.Values["payload"].(string)
  449. rawTaskInfo := message.Values["taskInfo"].(string)
  450. streamName := streamNamePrefix + sn
  451. // 反序列化payload
  452. var payload domain.BatchDownloadFileTaskInfo
  453. if err := json.Unmarshal([]byte(rawPayload), &payload); err != nil {
  454. ackTask(CallFileTaskStreamInfo{
  455. Ctx: callFwhFileCtx,
  456. StreamName: streamName,
  457. GroupName: groupName,
  458. ID: id,
  459. }, false, fmt.Errorf("unmarshal payload error - %s", err.Error()))
  460. return
  461. }
  462. // 反序列化taskInfo
  463. var taskInfo CallFileTaskInfo
  464. if err := json.Unmarshal([]byte(rawTaskInfo), &taskInfo); err != nil {
  465. ackTask(CallFileTaskStreamInfo{
  466. Ctx: callFwhFileCtx,
  467. StreamName: streamName,
  468. GroupName: groupName,
  469. ID: id,
  470. }, false, fmt.Errorf("unmarshal taskInfo error - %s", err.Error()))
  471. return
  472. }
  473. logger.Std.Printf("Processing task %s: %+v\n", id, payload)
  474. // 异步执行召唤文件任务
  475. callFwhFile(taskInfo.MainTaskId, id, payload)
  476. // switch payload.FileType {
  477. // case FileTypeEmsLog: // 召唤EMS日志
  478. // if err := handleTaskSiteAnalysis(ctx, payload.Payload); err != nil {
  479. // ackTask(CallFileTaskStreamInfo{
  480. // Ctx: callFwhFileCtx,
  481. // StreamName: streamName,
  482. // GroupName: groupName,
  483. // ID: id,
  484. // }, false, fmt.Errorf("handle task SiteAnalysis error - %s", err.Error()))
  485. // return
  486. // }
  487. // case FileTypeFaultRecord: // 召唤故障录波
  488. // if err := handleTaskShutdownStatistics(ctx, payload.Payload); err != nil {
  489. // ackTask(CallFileTaskStreamInfo{
  490. // Ctx: callFwhFileCtx,
  491. // StreamName: streamName,
  492. // GroupName: groupName,
  493. // ID: id,
  494. // }, false, fmt.Errorf("handle task SiteAnalysis error - %s", err.Error()))
  495. // return
  496. // }
  497. // default:
  498. // logger.Std.Println("Unknown task type:", payload.FileType)
  499. // }
  500. // 任务完成
  501. // ackTask(CallFileTaskStreamInfo{
  502. // Ctx: callFwhFileCtx,
  503. // StreamName: streamName,
  504. // GroupName: groupName,
  505. // ID: id,
  506. // }, true, nil)
  507. }
  508. func ackTask(streamInfo CallFileTaskStreamInfo, success bool, err error) {
  509. // 删除任务
  510. if delErr := cache.XAckDel(streamInfo.Ctx, streamInfo.StreamName, streamInfo.GroupName, streamInfo.ID); delErr != nil {
  511. logger.Std.Println("XAck error:", delErr)
  512. }
  513. if success {
  514. // TODO 确认任务完成
  515. } else {
  516. // TODO 任务失败
  517. logger.Std.Println("Task queue execution failed, error:", err)
  518. }
  519. if err != nil {
  520. zap.L().Error("taskQueue ackTask error.", zap.String("err", err.Error()), zap.String("StreamName", streamInfo.StreamName), zap.String("GroupName", streamInfo.GroupName), zap.String("ID", streamInfo.ID))
  521. }
  522. }
  523. func enqueueTask(sn string, taskInfo CallFileTaskInfo, payload domain.BatchDownloadFileTaskInfo) (string, error) {
  524. payloadBytes, err := json.Marshal(payload)
  525. if err != nil {
  526. return "", err
  527. }
  528. taskInfoBytes, err := json.Marshal(taskInfo)
  529. if err != nil {
  530. return "", err
  531. }
  532. return cache.XAdd(callFwhFileCtx, streamNamePrefix+sn, map[string]interface{}{"payload": string(payloadBytes), "taskInfo": string(taskInfoBytes)})
  533. }
  534. /************* 文件召唤&查询&下载接口 *************/
  535. // 召唤EMS日志
  536. func callEmsLog(ctx context.Context, sn string, date string, logType string) error {
  537. logTypeNum, ok := mapEmsLogType[logType]
  538. if !ok {
  539. zap.L().Error("logType不存在", zap.String("logType", logType))
  540. return fmt.Errorf("logType不存在,logType=%s", logType)
  541. }
  542. err := repositories.QueryCallEmsLogContext(ctx, sn, date, logTypeNum)
  543. if err != nil {
  544. zap.L().Error("QueryCallEmsLog请求失败", zap.String("err", err.Error()))
  545. return err
  546. }
  547. return nil
  548. }
  549. // 获取本次召唤的日志下载地址
  550. func getEmsLogUrlAfter(ctx context.Context, sn string, date string, logType string, afterID int64) (*domain.EmsLogRecordInfo, error) {
  551. logTypeNum, ok := mapEmsLogType[logType]
  552. if !ok {
  553. zap.L().Error("logType不存在", zap.String("logType", logType))
  554. return nil, fmt.Errorf("logType不存在,logType=%s", logType)
  555. }
  556. recordInfo, err := repositories.QueryEmsLogCallRecordAfterContext(ctx, sn, date, logTypeNum, afterID)
  557. if err != nil {
  558. zap.L().Error("GetEmsLogUrl请求失败", zap.String("err", err.Error()))
  559. return nil, err
  560. }
  561. return recordInfo, nil
  562. }
  563. // 召唤故障录波
  564. func callFaultRecord(devType int, sn string, startTime string, endTime string, indexList []int) error {
  565. // 查询故障录波列表
  566. err := repositories.QueryCellFaultRecord(devType, sn, indexList)
  567. if err != nil {
  568. return err
  569. }
  570. return nil
  571. }
  572. // 获取日志下载地址
  573. // return (status, faultId, faultReason, error)
  574. // status: -1-失败 0-召唤中 2-成功
  575. func getFaultRecordUrl(devType int, sn string, startTime string, endTime string) (int, int64, string, error) {
  576. // 查询召唤记录
  577. recordList, err := repositories.QueryFaultRecordCallRecordList()
  578. if err != nil {
  579. return 0, 0, "", err
  580. }
  581. for _, record := range recordList {
  582. // // 解析时间
  583. // t, err := time.Parse(time.DateTime, record.CellTime)
  584. // if err != nil {
  585. // return 0, nil, err
  586. // }
  587. // // 加 8 小时
  588. // newTime := t.Add(8 * time.Hour)
  589. // // 转回字符串
  590. // callTime := newTime.Format(time.DateTime)
  591. // // 判断记录是否当天召唤
  592. // currTime := time.Now().Format(time.DateOnly)
  593. // if !strings.HasPrefix(callTime, currTime) {
  594. // // 之后的非当天召唤的数据
  595. // break
  596. // }
  597. // 匹配符合条件的第一个数据
  598. if record.DeviceType == devType {
  599. var devSn string = ""
  600. switch devType {
  601. case 0: // aPower
  602. devSn = *record.ApowerId
  603. case 1: // aGate
  604. devSn = record.GatewayId
  605. case 2: // MPPT
  606. devSn = *record.ApowerId
  607. case 3: // aHub
  608. devSn = *record.PowerlinkId
  609. default:
  610. continue
  611. }
  612. if devSn == sn {
  613. // 匹配到的站点
  614. if *record.Status == -1 {
  615. return *record.Status, record.Id, *record.FaultReason, nil
  616. }
  617. return *record.Status, record.Id, "", nil
  618. }
  619. }
  620. }
  621. return 0, 0, "", fmt.Errorf("未找到召唤匹配记录(devType=%d,sn=%s,startTime=%s,endTime=%s)", devType, sn, startTime, endTime)
  622. }