jy-data-extract/src/jy/extract/exportask.go

// exportask
package extract

import (
	"fmt"
	db "jy/mongodbutil"
	ju "jy/util"
	"log"
	qu "qfw/util"
	"time"

	"github.com/tealeg/xlsx"
)

//执行导出定时任务
func Export() {
	tk, _ := db.Mgo.Find("task_export", `{"state":0}`, nil, nil, false, -1, -1)
	for _, t := range *tk {
		extractAndExport("v1", t)
		extractAndExport("v2", t)
		time.Sleep(20 * time.Second)
		//生成excel
		filename := createFile(t)
		if filename != "" {
			db.Mgo.UpdateById("task_export", qu.BsonIdToSId(t["_id"]),
				map[string]interface{}{
					"$set": map[string]interface{}{
						"state":    1,
						"filename": filename,
					},
				})
		}
	}
	time.AfterFunc(30*time.Minute, Export)
}

func extractAndExport(v string, t map[string]interface{}) {
	db.Mgo.Del("task_export_result_"+v, `{}`) //清除历史数据
	e := &ExtractTask{}
	e.IsRun = true
	e.TaskInfo = &TaskInfo{
		Name:        t["name"].(string),
		Version:     t[v+"name"].(string),
		VersionId:   t[v+"id"].(string),
		FromDbAddr:  t["dbaddr"].(string),
		FromDB:      t["dbname"].(string),
		FromColl:    t["table"].(string),
		TestColl:    "task_export_result_" + v,
		TrackColl:   "task_export_tract_" + v,
		IsEtxLog:    false,
		ProcessPool: make(chan bool, 5),
	}
	e.TaskInfo.FDB = db.MgoFactory(1, 3, 120, fmt.Sprint(t["dbaddr"]), fmt.Sprint(t["dbname"]))
	e.InitRulePres()
	e.InitRuleBacks()
	e.InitRuleCore()
	e.InitTag()
	e.InitClearFn()
	//品牌抽取是否开启
	ju.IsBrandGoods = ju.Config["brandgoods"].(bool)

	query := t["query"]
	limit := qu.IntAll(t["limit"])
	list, _ := e.TaskInfo.FDB.Find(e.TaskInfo.FromColl, query, nil, Fields, false, 0, limit)
	for _, v := range *list {
		if qu.ObjToString(v["sensitive"]) != "" { //去除含敏感词数据
			continue
		}
		var j, jf *ju.Job
		if e.IsFileField{
			if v["projectinfo"] != nil {
				v["isextFile"] = true
				j, jf = PreInfo(v)
			}
		}else {
			j, _ = PreInfo(v)
		}
		e.TaskInfo.ProcessPool <- true
		go e.ExtractProcess(j, jf)
	}
}

func createFile(info map[string]interface{}) string {
	filename := ""
	if fields, ok := info["fields"].([]interface{}); ok {
		qfield := ""
		for k, v := range fields {
			if k == 0 {
				qfield += `"` + fmt.Sprint(v) + `":1`
			} else {
				qfield += `,"` + fmt.Sprint(v) + `":1`
			}
		}
		list_1, _ := db.Mgo.Find("task_export_result_v1", `{}`, nil, "{"+qfield+"}", false, -1, -1)
		result := map[string]map[string]interface{}{}
		for _, v := range *list_1 {
			id := qu.BsonIdToSId(v["_id"])
			tmp := map[string]interface{}{}
			for _, f := range fields {
				field := fmt.Sprint(f)
				tmp["v1_"+field] = qu.ObjToString(v[field])
			}
			result[id] = tmp
		}
		list_2, _ := db.Mgo.Find("task_export_result_v2", `{}`, nil, "{"+qfield+"}", false, -1, -1)
		for _, v := range *list_2 {
			id := qu.BsonIdToSId(v["_id"])
			tmp := map[string]interface{}{}
			if result[id] != nil {
				tmp = result[id]
			}
			for _, f := range fields {
				field := fmt.Sprint(f)
				tmp["v2_"+field] = qu.ObjToString(v[field])
			}
			result[id] = tmp
		}
		sheetmaps := map[string][]map[string]interface{}{}
		for _, f := range fields {
			field := fmt.Sprint(f)
			tmps := []map[string]interface{}{}
			for _, v := range result {
				tmp := map[string]interface{}{
					"v1": v["v1_"+field],
					"v2": v["v2_"+field],
				}
				tmps = append(tmps, tmp)
			}
			sheetmaps[field] = tmps
		}
		fx := xlsx.NewFile()
		for k, v := range sheetmaps {
			sheet := xlsx.Sheet{}
			title := sheet.AddRow()
			title.AddCell().SetString(fmt.Sprint(info["v1name"]))
			title.AddCell().SetString(fmt.Sprint(info["v2name"]))
			for _, d := range v {
				row := sheet.AddRow()
				row.AddCell().SetString(fmt.Sprint(d["v1"]))
				row.AddCell().SetString(fmt.Sprint(d["v2"]))
			}
			fx.AppendSheet(sheet, k)
		}
		t := time.Now()
		filename = fmt.Sprint(info["name"]) + "_" + t.Format("20060102150405") + ".xlsx"
		fx.Save("./web/res/down/" + filename)
		log.Println("任务完成", filename)
	}
	return filename
}