exportask.go 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. // exportask
  2. package extract
  3. import (
  4. "fmt"
  5. db "jy/mongodbutil"
  6. ju "jy/util"
  7. "log"
  8. qu "qfw/util"
  9. "time"
  10. "github.com/tealeg/xlsx"
  11. )
  12. //执行导出定时任务
  13. func Export() {
  14. tk, _ := db.Mgo.Find("task_export", `{"state":0}`, nil, nil, false, -1, -1)
  15. for _, t := range *tk {
  16. extractAndExport("v1", t)
  17. extractAndExport("v2", t)
  18. time.Sleep(20 * time.Second)
  19. //生成excel
  20. filename := createFile(t)
  21. if filename != "" {
  22. db.Mgo.UpdateById("task_export", qu.BsonIdToSId(t["_id"]),
  23. map[string]interface{}{
  24. "$set": map[string]interface{}{
  25. "state": 1,
  26. "filename": filename,
  27. },
  28. })
  29. }
  30. }
  31. time.AfterFunc(30*time.Minute, Export)
  32. }
  33. func extractAndExport(v string, t map[string]interface{}) {
  34. db.Mgo.Del("task_export_result_"+v, `{}`) //清除历史数据
  35. e := &ExtractTask{}
  36. e.IsRun = true
  37. e.TaskInfo = &TaskInfo{
  38. Name: t["name"].(string),
  39. Version: t[v+"name"].(string),
  40. VersionId: t[v+"id"].(string),
  41. FromDbAddr: t["dbaddr"].(string),
  42. FromDB: t["dbname"].(string),
  43. FromColl: t["table"].(string),
  44. TestColl: "task_export_result_" + v,
  45. TrackColl: "task_export_tract_" + v,
  46. IsEtxLog: false,
  47. ProcessPool: make(chan bool, 5),
  48. }
  49. e.TaskInfo.FDB = db.MgoFactory(1, 3, 120, fmt.Sprint(t["dbaddr"]), fmt.Sprint(t["dbname"]))
  50. e.InitRulePres()
  51. e.InitRuleBacks()
  52. e.InitRuleCore()
  53. e.InitTag()
  54. e.InitClearFn()
  55. e.InfoTypeList()
  56. //品牌抽取是否开启
  57. ju.IsBrandGoods = ju.Config["brandgoods"].(bool)
  58. //附件抽取是否开启
  59. e.InitFile()
  60. query := t["query"]
  61. limit := qu.IntAll(t["limit"])
  62. list, _ := e.TaskInfo.FDB.Find(e.TaskInfo.FromColl, query, nil, Fields, false, 0, limit)
  63. for _, v := range *list {
  64. if qu.ObjToString(v["sensitive"]) != "" { //去除含敏感词数据
  65. continue
  66. }
  67. var j, jf *ju.Job
  68. if e.IsFileField && v["projectinfo"] != nil {
  69. v["isextFile"] = true
  70. j, jf = e.PreInfo(v)
  71. } else {
  72. j, _ = e.PreInfo(v)
  73. }
  74. e.TaskInfo.ProcessPool <- true
  75. go e.ExtractProcess(j, jf)
  76. }
  77. }
  78. func createFile(info map[string]interface{}) string {
  79. filename := ""
  80. if fields, ok := info["fields"].([]interface{}); ok {
  81. qfield := ""
  82. for k, v := range fields {
  83. if k == 0 {
  84. qfield += `"` + fmt.Sprint(v) + `":1`
  85. } else {
  86. qfield += `,"` + fmt.Sprint(v) + `":1`
  87. }
  88. }
  89. list_1, _ := db.Mgo.Find("task_export_result_v1", `{}`, nil, "{"+qfield+"}", false, -1, -1)
  90. result := map[string]map[string]interface{}{}
  91. for _, v := range *list_1 {
  92. id := qu.BsonIdToSId(v["_id"])
  93. tmp := map[string]interface{}{}
  94. for _, f := range fields {
  95. field := fmt.Sprint(f)
  96. tmp["v1_"+field] = qu.ObjToString(v[field])
  97. }
  98. result[id] = tmp
  99. }
  100. list_2, _ := db.Mgo.Find("task_export_result_v2", `{}`, nil, "{"+qfield+"}", false, -1, -1)
  101. for _, v := range *list_2 {
  102. id := qu.BsonIdToSId(v["_id"])
  103. tmp := map[string]interface{}{}
  104. if result[id] != nil {
  105. tmp = result[id]
  106. }
  107. for _, f := range fields {
  108. field := fmt.Sprint(f)
  109. tmp["v2_"+field] = qu.ObjToString(v[field])
  110. }
  111. result[id] = tmp
  112. }
  113. sheetmaps := map[string][]map[string]interface{}{}
  114. for _, f := range fields {
  115. field := fmt.Sprint(f)
  116. tmps := []map[string]interface{}{}
  117. for _, v := range result {
  118. tmp := map[string]interface{}{
  119. "v1": v["v1_"+field],
  120. "v2": v["v2_"+field],
  121. }
  122. tmps = append(tmps, tmp)
  123. }
  124. sheetmaps[field] = tmps
  125. }
  126. fx := xlsx.NewFile()
  127. for k, v := range sheetmaps {
  128. sheet := xlsx.Sheet{}
  129. title := sheet.AddRow()
  130. title.AddCell().SetString(fmt.Sprint(info["v1name"]))
  131. title.AddCell().SetString(fmt.Sprint(info["v2name"]))
  132. for _, d := range v {
  133. row := sheet.AddRow()
  134. row.AddCell().SetString(fmt.Sprint(d["v1"]))
  135. row.AddCell().SetString(fmt.Sprint(d["v2"]))
  136. }
  137. fx.AppendSheet(sheet, k)
  138. }
  139. t := time.Now()
  140. filename = fmt.Sprint(info["name"]) + "_" + t.Format("20060102150405") + ".xlsx"
  141. fx.Save("./web/res/down/" + filename)
  142. log.Println("任务完成", filename)
  143. }
  144. return filename
  145. }