batchDownloadRepo.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459
  1. package mongo
  2. import (
  3. "context"
  4. "fwh-toolbox-server/database"
  5. "fwh-toolbox-server/domain"
  6. "fwh-toolbox-server/dto"
  7. "time"
  8. "go.mongodb.org/mongo-driver/bson"
  9. "go.mongodb.org/mongo-driver/bson/primitive"
  10. "go.mongodb.org/mongo-driver/mongo/options"
  11. )
  12. // 从数据库获取批量下载数据列表
  13. func GetBatchDownloadManageList(page, size uint, sn, progress *string) (*dto.ResPage[domain.BatchDownloadInfo], error) {
  14. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  15. defer cancel()
  16. collection := database.GetCollection("BatchDownloadConfig")
  17. filter := bson.M{}
  18. // 条件
  19. if sn != nil {
  20. filter["sn"] = *sn
  21. }
  22. if progress != nil {
  23. filter["progress"] = *progress
  24. }
  25. findOptions := options.Find().
  26. SetProjection(bson.M{
  27. "_id": 1,
  28. "sn": 1,
  29. "description": 1,
  30. "timePeriod": 1,
  31. "options": 1,
  32. "progress": 1,
  33. "failReasons": 1,
  34. }).
  35. SetSkip(int64((page - 1) * size)).
  36. SetLimit(int64(size)).
  37. SetSort(bson.D{
  38. {Key: "_id", Value: -1}, // 倒序
  39. })
  40. cursor, err := collection.Find(ctx, filter, findOptions)
  41. if err != nil {
  42. return nil, err
  43. }
  44. defer cursor.Close(ctx)
  45. var configList []domain.BatchDownloadInfo
  46. for cursor.Next(ctx) {
  47. var cfg domain.BatchDownloadInfo
  48. if err := cursor.Decode(&cfg); err != nil {
  49. return nil, err
  50. }
  51. configList = append(configList, cfg)
  52. }
  53. // 查询总数
  54. total, err := collection.CountDocuments(ctx, filter)
  55. if err != nil {
  56. return nil, err
  57. }
  58. res := dto.ResPage[domain.BatchDownloadInfo]{
  59. Data: configList,
  60. PageSize: size,
  61. PageNum: page,
  62. Total: uint(total),
  63. }
  64. return &res, nil
  65. }
  66. // 从数据库获取一条批量下载数据配置
  67. func GetBatchDownloadConfig(id primitive.ObjectID) (*domain.BatchDownloadConfig, error) {
  68. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  69. defer cancel()
  70. collection := database.GetCollection("BatchDownloadConfig")
  71. var cfg domain.BatchDownloadConfig
  72. err := collection.FindOne(ctx, bson.M{"_id": id}).Decode(&cfg)
  73. if err != nil {
  74. return nil, err
  75. }
  76. return &cfg, nil
  77. }
  78. // 插入一条批量下载配置到数据库
  79. func InsertBatchDownloadConfig(writeConfig domain.BatchDownloadConfigWrite) (primitive.ObjectID, error) {
  80. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  81. defer cancel()
  82. config := domain.BatchDownloadConfig{
  83. Sn: writeConfig.Sn,
  84. Description: writeConfig.Description,
  85. TimePeriod: writeConfig.TimePeriod,
  86. Options: writeConfig.Options,
  87. Progress: domain.ProgressPending,
  88. Data: map[string]domain.BatchDownloadData{},
  89. FailReasons: []string{},
  90. }
  91. collection := database.GetCollection("BatchDownloadConfig")
  92. res, err := collection.InsertOne(ctx, config)
  93. if err != nil {
  94. return primitive.ObjectID{}, err
  95. }
  96. return res.InsertedID.(primitive.ObjectID), nil
  97. }
  98. // 更新一条批量下载描述描述到数据库
  99. func UpdateBatchDownloadDescription(id primitive.ObjectID, desc string) error {
  100. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  101. defer cancel()
  102. collection := database.GetCollection("BatchDownloadConfig")
  103. // _, err := collection.InsertOne(ctx, config)
  104. filter := bson.M{"_id": id} // 按ID匹配
  105. update := bson.M{
  106. // 如果存在就更新全部字段
  107. "$set": bson.M{
  108. "description": desc,
  109. },
  110. }
  111. opts := options.Update().SetUpsert(true)
  112. _, err := collection.UpdateOne(ctx, filter, update, opts)
  113. if err != nil {
  114. return err
  115. }
  116. return nil
  117. }
  118. // 更新一条批量下载数据owner到数据库
  119. func UpdateBatchDownloadOwner(id primitive.ObjectID, owner []string) error {
  120. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  121. defer cancel()
  122. collection := database.GetCollection("BatchDownloadConfig")
  123. // _, err := collection.InsertOne(ctx, config)
  124. filter := bson.M{"_id": id} // 按ID匹配
  125. update := bson.M{
  126. // 如果存在就更新全部字段
  127. "$set": bson.M{
  128. "owner": owner,
  129. },
  130. }
  131. opts := options.Update().SetUpsert(true)
  132. _, err := collection.UpdateOne(ctx, filter, update, opts)
  133. if err != nil {
  134. return err
  135. }
  136. return nil
  137. }
  138. // 更新一条批量下载结果到数据库
  139. func UpdateBatchDownloadResult(id primitive.ObjectID, result domain.Progress, reasons []string) error {
  140. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  141. defer cancel()
  142. collection := database.GetCollection("BatchDownloadConfig")
  143. // _, err := collection.InsertOne(ctx, config)
  144. filter := bson.M{"_id": id} // 按ID匹配
  145. update := bson.M{
  146. // 如果存在就更新非progress字段
  147. "$set": bson.M{
  148. "progress": result,
  149. "failReasons": reasons,
  150. },
  151. }
  152. opts := options.Update().SetUpsert(true)
  153. _, err := collection.UpdateOne(ctx, filter, update, opts)
  154. if err != nil {
  155. return err
  156. }
  157. return nil
  158. }
  159. // 删除一条批量下载配置
  160. func DeleteBatchDownloadConfig(id primitive.ObjectID) error {
  161. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  162. defer cancel()
  163. filter := bson.M{"_id": id} // 删除条件
  164. collection := database.GetCollection("BatchDownloadConfig")
  165. _, err := collection.DeleteOne(ctx, filter)
  166. if err != nil {
  167. return err
  168. }
  169. return nil
  170. }
  171. // 插入批量下载结果数据
  172. func InsertBatchDownloadData(id primitive.ObjectID, data domain.BatchDownloadData) error {
  173. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  174. defer cancel()
  175. collection := database.GetCollection("BatchDownloadConfig")
  176. // _, err := collection.InsertOne(ctx, config)
  177. filter := bson.M{"_id": id} // 按ID匹配
  178. update := bson.M{
  179. // 如果存在就更新data字段
  180. "$set": bson.M{
  181. "data": data,
  182. },
  183. }
  184. opts := options.Update().SetUpsert(true)
  185. _, err := collection.UpdateOne(ctx, filter, update, opts)
  186. if err != nil {
  187. return err
  188. }
  189. return nil
  190. }
  191. // 获取下载数据包路径
  192. func GetBatchDownloadData(id primitive.ObjectID) (map[string]domain.BatchDownloadData, error) {
  193. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  194. defer cancel()
  195. collection := database.GetCollection("BatchDownloadConfig")
  196. var cfg domain.BatchDownloadConfig
  197. err := collection.FindOne(ctx, bson.M{"_id": id}).Decode(&cfg)
  198. if err != nil {
  199. return nil, err
  200. }
  201. return cfg.Data, nil
  202. }
  203. // 插入批量下载结果数据 - PowerFlow
  204. func InsertBatchDownloadPathPowerFlow(id primitive.ObjectID, sn string, date string, path string) error {
  205. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  206. defer cancel()
  207. collection := database.GetCollection("BatchDownloadConfig")
  208. // _, err := collection.InsertOne(ctx, config)
  209. filter := bson.M{"_id": id} // 按ID匹配
  210. update := bson.M{
  211. // 如果存在就更新data字段
  212. "$set": bson.M{
  213. "data." + sn + ".powerFlow." + date: path,
  214. },
  215. }
  216. opts := options.Update().SetUpsert(true)
  217. _, err := collection.UpdateOne(ctx, filter, update, opts)
  218. if err != nil {
  219. return err
  220. }
  221. return nil
  222. }
  223. // 插入批量下载结果数据 - MqttLogs
  224. func InsertBatchDownloadPathMqttLogs(id primitive.ObjectID, sn string, date string, logType string, path string) error {
  225. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  226. defer cancel()
  227. collection := database.GetCollection("BatchDownloadConfig")
  228. // _, err := collection.InsertOne(ctx, config)
  229. filter := bson.M{"_id": id} // 按ID匹配
  230. update := bson.M{
  231. // 如果存在就更新data字段
  232. "$set": bson.M{
  233. "data." + sn + ".mqttLogs." + date + "." + logType: path,
  234. },
  235. }
  236. opts := options.Update().SetUpsert(true)
  237. _, err := collection.UpdateOne(ctx, filter, update, opts)
  238. if err != nil {
  239. return err
  240. }
  241. return nil
  242. }
  243. // 插入批量下载结果数据 - EventAlarm
  244. func InsertBatchDownloadPathEventAlarm(id primitive.ObjectID, sn string, date string, path string) error {
  245. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  246. defer cancel()
  247. collection := database.GetCollection("BatchDownloadConfig")
  248. // _, err := collection.InsertOne(ctx, config)
  249. filter := bson.M{"_id": id} // 按ID匹配
  250. update := bson.M{
  251. // 如果存在就更新data字段
  252. "$set": bson.M{
  253. "data." + sn + ".eventAlarm." + date: path,
  254. },
  255. }
  256. opts := options.Update().SetUpsert(true)
  257. _, err := collection.UpdateOne(ctx, filter, update, opts)
  258. if err != nil {
  259. return err
  260. }
  261. return nil
  262. }
  263. // 插入批量下载结果数据 - AlarmSnapshot
  264. func InsertBatchDownloadPathAlarmSnapshot(id primitive.ObjectID, sn string, date string, path string) error {
  265. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  266. defer cancel()
  267. collection := database.GetCollection("BatchDownloadConfig")
  268. // _, err := collection.InsertOne(ctx, config)
  269. filter := bson.M{"_id": id} // 按ID匹配
  270. update := bson.M{
  271. // 如果存在就更新data字段
  272. "$set": bson.M{
  273. "data." + sn + ".alarmSnapshot." + date: path,
  274. },
  275. }
  276. opts := options.Update().SetUpsert(true)
  277. _, err := collection.UpdateOne(ctx, filter, update, opts)
  278. if err != nil {
  279. return err
  280. }
  281. return nil
  282. }
  283. // 插入批量下载结果数据 - EmsLogs
  284. func InsertBatchDownloadPathEmsLogs(id primitive.ObjectID, sn string, date string, logType string, path string) error {
  285. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  286. defer cancel()
  287. collection := database.GetCollection("BatchDownloadConfig")
  288. // _, err := collection.InsertOne(ctx, config)
  289. filter := bson.M{"_id": id} // 按ID匹配
  290. update := bson.M{
  291. // 如果存在就更新data字段
  292. "$set": bson.M{
  293. "data." + sn + ".emsLogs." + date + "." + logType: path,
  294. },
  295. }
  296. opts := options.Update().SetUpsert(true)
  297. _, err := collection.UpdateOne(ctx, filter, update, opts)
  298. if err != nil {
  299. return err
  300. }
  301. return nil
  302. }
  303. // 插入批量下载结果数据 - FaultRecord
  304. func InsertBatchDownloadPathFaultRecord(id primitive.ObjectID, sn string, fileName string, path string) error {
  305. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  306. defer cancel()
  307. collection := database.GetCollection("BatchDownloadConfig")
  308. // _, err := collection.InsertOne(ctx, config)
  309. filter := bson.M{"_id": id} // 按ID匹配
  310. update := bson.M{
  311. // 如果存在就更新data字段
  312. "$set": bson.M{
  313. "data." + sn + ".faultRecord." + fileName: path,
  314. },
  315. }
  316. opts := options.Update().SetUpsert(true)
  317. _, err := collection.UpdateOne(ctx, filter, update, opts)
  318. if err != nil {
  319. return err
  320. }
  321. return nil
  322. }
  323. // 插入批量下载结果数据 - UpdateLog
  324. func InsertBatchDownloadPathUpdateLog(id primitive.ObjectID, sn string, path string) error {
  325. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  326. defer cancel()
  327. collection := database.GetCollection("BatchDownloadConfig")
  328. // _, err := collection.InsertOne(ctx, config)
  329. filter := bson.M{"_id": id} // 按ID匹配
  330. update := bson.M{
  331. // 如果存在就更新data字段
  332. "$set": bson.M{
  333. "data." + sn + ".updateLog": path,
  334. },
  335. }
  336. opts := options.Update().SetUpsert(true)
  337. _, err := collection.UpdateOne(ctx, filter, update, opts)
  338. if err != nil {
  339. return err
  340. }
  341. return nil
  342. }
  343. // 插入批量下载结果数据 - Settings
  344. func InsertBatchDownloadPathSettings(id primitive.ObjectID, sn string, path string) error {
  345. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  346. defer cancel()
  347. collection := database.GetCollection("BatchDownloadConfig")
  348. // _, err := collection.InsertOne(ctx, config)
  349. filter := bson.M{"_id": id} // 按ID匹配
  350. update := bson.M{
  351. // 如果存在就更新data字段
  352. "$set": bson.M{
  353. "data." + sn + ".settings": path,
  354. },
  355. }
  356. opts := options.Update().SetUpsert(true)
  357. _, err := collection.UpdateOne(ctx, filter, update, opts)
  358. if err != nil {
  359. return err
  360. }
  361. return nil
  362. }