data_validation/src/front/project.go

package front

import (
	"encoding/json"
	"fmt"
	"github.com/tealeg/xlsx"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"io/ioutil"
	"math"
	"mime/multipart"
	"mongodb"
	qu "qfw/util"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"util"
)

// ProjectList 项目列表
func (f *Front) ProjectList() {
	defer qu.Catch()
	if f.Method() == "POST" {
		start, _ := f.GetInteger("start")
		limit, _ := f.GetInteger("length")
		draw, _ := f.GetInteger("draw")
		status := f.GetString("status")
		searchStr := f.GetString("search[value]")
		search := strings.TrimSpace(searchStr)
		//data := util.GetPostForm(f.Request)
		query := map[string]interface{}{}
		if status != "-1" {
			query["s_status"] = status
		}
		if search != "" {
			query["$or"] = []interface{}{
				map[string]interface{}{"s_name": map[string]interface{}{"$regex": search}},
				map[string]interface{}{"s_entname": map[string]interface{}{"$regex": search}},
				map[string]interface{}{"s_rule": map[string]interface{}{"$regex": search}},
				map[string]interface{}{"s_departname": map[string]interface{}{"$regex": search}},
			}
		}
		list, _ := util.Mgo.Find(util.PROJECTCOLLNAME, query, nil, nil, false, start, limit)
		count := util.Mgo.Count(util.PROJECTCOLLNAME, query)
		f.ServeJson(map[string]interface{}{"draw": draw, "data": *list, "recordsFiltered": count, "recordsTotal": count})
	} else {
		query := map[string]interface{}{"s_type": "tag"}
		info, _ := util.Mgo.Find("v_field", query, nil, map[string]interface{}{"s_name": 1, "s_code": 1}, false, -1, -1)
		f.T["fields"] = *info
		_ = f.Render("project/project_list.html", &f.T)
	}
}

// ProjectSave 项目保存
func (f *Front) ProjectSave() {
	defer qu.Catch()
	s_name := f.GetString("s_name") //项目名称
	if s_name == "" {
		f.ServeJson(map[string]interface{}{"success": false, "msg": "缺少项目名称字段"})
		return
	}
	success := false        //导入数据是否成功
	msg := ""               //异常信息
	successNum := int64(0)  //导入成功条数
	importDataNum := 0      //查询数量
	var s_rulename []string //规则
	user := f.GetSession("user").(map[string]interface{})
	username := qu.ObjToString(user["s_name"])  //当前登录用户
	stype := f.GetString("s_type")              //新建项目类型：数据库导入、excel导入
	s_sourceinfo := f.GetString("s_sourceinfo") //数据表
	s_sourceinfo = "f_sourceinfo_" + s_sourceinfo
	if s_sourceinfo == "" {
		f.ServeJson(map[string]interface{}{"success": false, "msg": "缺少数据存储表名"})
		return
	}
	s_departname, s_entname := "", ""
	query := map[string]interface{}{
		"s_name": s_name,
	}
	set := map[string]interface{}{}
	//导入数据
	if stype == "excel" { //excel导入
		s_entname = f.GetString("s_entname") //公司名称
		if s_entname == "" {
			f.ServeJson(map[string]interface{}{"success": false, "msg": "缺少公司名称字段"})
			return
		}
		s_departname = f.GetString("s_departname") //部门名称
		rulename := f.GetString("s_rulename")      //规则名称
		s_rulename = strings.Split(rulename, ",")
		mf, _, err := f.GetFile("xlsx")
		if err == nil {
			importDataNum = ImportDataByExcel(s_sourceinfo, mf, &success, &msg, &successNum)
		}
		//保存项目信息
		set = map[string]interface{}{
			"s_name":       s_name,                        //项目名称
			"s_entname":    s_entname,                     //公司名称
			"s_departname": s_departname,                  //部门名称
			"s_rulename":   strings.Join(s_rulename, ","), //规则名称
			"i_importnum":  importDataNum,                 //导入数量
			"s_sourceinfo": s_sourceinfo,                  //源数据表
			"s_createname": username,                      //创建人
			"s_status":     "未开始",                         //项目状态
			"i_createtime": time.Now().Unix(),             //创建时间
			"s_importtype": "excel",                       //导入类型
		}
	} else if stype == "coll" { //数据库导入
		historyid := f.GetString("s_historyid")
		if historyid == "" {
			f.ServeJson(map[string]interface{}{"success": false, "msg": "数据导出ID字段"})
			return
		}
		s_departname, s_entname, s_rulename, importDataNum = ImportDataByColl(s_sourceinfo, historyid, &success, &msg, &successNum)
		qu.Debug(s_departname, s_entname, s_rulename, importDataNum)
		//保存项目信息
		set = map[string]interface{}{
			"s_name":       s_name,                        //项目名称
			"s_entname":    s_entname,                     //公司名称
			"s_departname": s_departname,                  //部门名称
			"s_rulename":   strings.Join(s_rulename, ","), //规则名称
			"i_importnum":  importDataNum,                 //导入数量
			"s_sourceinfo": s_sourceinfo,                  //源数据表
			"s_createname": username,                      //创建人
			"s_status":     "未开始",                         //项目状态
			"i_createtime": time.Now().Unix(),             //创建时间
			"s_importtype": "coll",                        //导入类型
			"s_historyid":  historyid,                     //源数据集标识
		}
	} else if stype == "edit" { //编辑保存
		success = true
		//s_entname = f.GetString("s_entname")       //公司名称
		s_departname = f.GetString("s_departname") //部门名称
		rulename := f.GetString("s_rulename")      //规则名称
		s_rulename = strings.Split(rulename, ",")
		s_personname := f.GetString("s_personname")
		fields := f.GetString("v_fields")
		v_fields := map[string]interface{}{}
		if err := json.Unmarshal([]byte(fields), &v_fields); err != nil {
			qu.Debug("V_Filelds Unmarshal Failed：", err)
			f.ServeJson(map[string]interface{}{"success": false})
			return
		}
		set = map[string]interface{}{
			//"s_name":       s_name,                        //项目名称
			//"s_entname":    s_entname,                     //公司名称
			"s_departname": s_departname,                  //部门名称
			"s_rulename":   strings.Join(s_rulename, ","), //规则名称
			"v_fields":     v_fields,                      //标注字段
			"i_updatetime": username,                      //更新人
			"i_createtime": time.Now().Unix(),             //更新时间
			"s_personname": s_personname,                  //售后人员
			//"i_starttime":,//开始时间
			//"i_completetime",//结束时间
		}
	}
	if success {
		success = util.Mgo.Update(util.PROJECTCOLLNAME, query, map[string]interface{}{"$set": set}, true, false)
		qu.Debug("Save Project：", success)
		if !success { //保存项目失败
			msg = "新建项目失败\n" + msg
		} else {
			msg = "新建项目成功\n" + msg
		}
	}
	qu.Debug("Create Project：", success, "importnum：", importDataNum, "successnum：", successNum, "failnum：", int64(importDataNum)-successNum)
	qu.Debug("Msg：", msg)
	//返回信息
	if stype == "edit" {
		f.ServeJson(map[string]interface{}{"success": success, "msg": msg})
	} else {
		f.ServeJson(map[string]interface{}{"success": success, "msg": msg, "importnum": importDataNum, "successnum": successNum, "failnum": int64(importDataNum) - successNum})
	}
}

// ProjectClear 项目清洗
func (f *Front) ProjectClear() {
	defer qu.Catch()
	if f.Method() == "POST" {
		projectid := f.GetString("s_projectid") //项目id
		project, _ := util.Mgo.FindById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{"s_status": 1, "s_sourceinfo": 1})
		if project != nil && len(*project) > 0 {
			if status := qu.ObjToString((*project)["s_status"]); status == "未开始" {
				//TODO:调用数据质量评估接口
				//点击清洗更新项目状态为进行中
				b := util.Mgo.UpdateById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{"$set": map[string]interface{}{"s_status": "进行中", "i_starttime": time.Now().Unix()}})
				qu.Debug("Update Porject："+projectid+"	Status Success：", b)
			}
		} else {
			qu.Debug("Search Porject Failed：", projectid)
			f.ServeJson("查询项目信息失败")
			return
		}
		sourceinfo := qu.ObjToString((*project)["s_sourceinfo"])                                                       //数据源表
		noTagAllDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": false})                    //达标数据总量
		noTagGiveDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": false, "b_isgive": true}) //达标数据已分发量
		noTagNoGiveDataNum := noTagAllDataNum - noTagGiveDataNum                                                       //达标待分发量
		tagAllDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": true})                       //未达标数据总量
		tagGiveDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": true, "b_isgive": true})    //未达标数据已分发量
		tagNoGiveDataNum := tagAllDataNum - tagGiveDataNum                                                             //未达标待分发量
		allGiveDataNum := noTagGiveDataNum + tagGiveDataNum                                                            //总分发量
		allNoGiveDataNum := noTagNoGiveDataNum + tagNoGiveDataNum                                                      //总待分发量
		allDataNum := allGiveDataNum + allNoGiveDataNum

		data := make(map[string]interface{})
		data["allDataNum"] = allDataNum
		data["allGiveDataNum"] = allGiveDataNum
		data["allNoGiveDataNum"] = allNoGiveDataNum
		data["noTagAllDataNum"] = noTagAllDataNum
		data["noTagGiveDataNum"] = noTagGiveDataNum
		data["noTagNoGiveDataNum"] = noTagNoGiveDataNum
		data["tagAllDataNum"] = tagAllDataNum
		data["tagGiveDataNum"] = tagGiveDataNum
		data["tagNoGiveDataNum"] = tagNoGiveDataNum
		data["s_projectid"] = projectid
		qu.Debug(data)
		f.ServeJson(data)
	} else {
		pid := f.GetString("pid")
		f.T["s_projectid"] = pid
		_ = f.Render("project/project_clear.html", &f.T)
	}

}

// ProjectTaskList 用户组任务分发列表
func (f *Front) ProjectTaskList() {
	defer qu.Catch()
	projectid := f.GetString("s_projectid") //项目id
	status := f.GetString("s_status")       //任务状态
	searchStr := f.GetString("search[value]")
	search := strings.TrimSpace(searchStr)
	start, _ := f.GetInteger("start")
	limit, _ := f.GetInteger("length")
	draw, _ := f.GetInteger("draw")
	query := map[string]interface{}{ //查找用户组任务
		"s_projectid": projectid,
		"s_stype":     "group",
	}
	if status != "-1" {
		query["s_status"] = status
	}
	if search != "" {
		query["$or"] = []interface{}{
			map[string]interface{}{"s_groupname": map[string]interface{}{"$regex": search}},
		}
	}
	list, _ := util.Mgo.Find(util.TASKCOLLNAME, query, nil, nil, false, start, limit)
	count := util.Mgo.Count(util.TASKCOLLNAME, query)
	for _, l := range *list {
		if status := qu.ObjToString(l["s_status"]); status == "进行中" { //更新任务进度
			groupId := qu.ObjToString(l["s_groupid"])
			giveNum := qu.IntAll(l["i_givenum"])
			sourceinfo := qu.ObjToString(l["s_sourceinfo"])
			tagNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"s_groupid": groupId, "b_istag": true})
			progress := fmt.Sprint(math.Ceil(float64(tagNum)/float64(giveNum))) + "%"
			l["s_progress"] = progress
			//同步数据库
			util.Mgo.UpdateById(util.TASKCOLLNAME, l["_id"], map[string]interface{}{"$set": map[string]interface{}{"s_progress": progress}})
		}
	}
	f.ServeJson(map[string]interface{}{"draw": draw, "data": *list, "recordsFiltered": count, "recordsTotal": count})
}

// ProjectTaskSave 用户组任务分发
func (f *Front) ProjectTaskSave() {
	defer qu.Catch()
	var groupArr []map[string]interface{}
	var taskArr []map[string]interface{}
	var groupIdArr []string
	groupIdTask := map[string]util.Task{}
	success := false
	msg := ""
	user := f.GetSession("user").(map[string]interface{})
	username := qu.ObjToString(user["s_name"])                                                  //当前登录用户
	projectid := f.GetString("s_projectid")                                                     //项目标识
	projectname := f.GetString("s_projectname")                                                 //项目名称
	sourceinfo := f.GetString("s_sourceinfo")                                                   //源数据表
	sourcetaskinfo := "f_sourcetaskinfo_" + strings.ReplaceAll(sourceinfo, "f_sourceinfo_", "") //任务日志表
	group := f.GetString("s_group")
	stype := f.GetString("s_type")
	if err := json.Unmarshal([]byte(group), &groupArr); err != nil {
		qu.Debug("V_Filelds Unmarshal Failed：", err)
		msg = "用户组信息解析失败"
	} else {
		if stype == "notag" { //如果分发的是达标数据且进行了初步质检，将没有质检记录的字段从v_taginfo标注记录中删除
			DeleleDataTagInfo(sourceinfo)
		}
		for _, groupInfo := range groupArr {
			groupId := qu.ObjToString(groupInfo["s_groupid"])
			groupIdArr = append(groupIdArr, groupId)
			givenum := qu.IntAll(groupInfo["i_givenum"])
			//groupIdMap[groupId] = givenum
			_id := primitive.NewObjectID()
			sid := mongodb.BsonIdToSId(_id)
			gt := util.Task{
				GroupId: groupId,
				GiveNum: givenum,
			}
			groupIdTask[sid] = gt
			groupTask := map[string]interface{}{
				"_id":              _id,                                       //生成任务id
				"s_projectid":      projectid,                                 //项目标识
				"s_projectname":    projectname,                               //项目名称
				"s_status":         "未开始",                                     //任务状态
				"s_personid":       qu.ObjToString(groupInfo["s_personid"]),   //任务负责人标识
				"s_personname":     qu.ObjToString(groupInfo["s_personname"]), //任务负责人
				"s_groupname":      qu.ObjToString(groupInfo["s_groupname"]),  //用户组名称
				"s_groupid":        groupId,                                   //用户组标识
				"i_givenum":        givenum,                                   //分发数据量
				"s_createname":     username,                                  //创建人
				"i_createtime":     time.Now().Unix(),                         //创建时间
				"s_progress":       "0%",                                      //完成进度
				"s_sourceinfo":     sourceinfo,                                //源数据表
				"s_sourcetaskinfo": sourcetaskinfo,                            //任务日志表
				"s_stype":          "group",                                   //任务类型
			}
			taskArr = append(taskArr, groupTask)
		}
	}
	if len(taskArr) > 0 {
		//分发数据后更新项目中用户组标识信息
		success = util.Mgo.UpdateById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{
			"$push": map[string]interface{}{
				"v_groupids": map[string]interface{}{
					"$each": groupIdArr,
				},
			},
		})
		if !success {
			msg = "更新项目:" + projectname + "用户组标识失败"
		} else { //用户组分发任务
			success = util.Mgo.SaveBulk(util.TASKCOLLNAME, taskArr...)
			if success {
				msg = "任务分发成功"
				UpdateSourceinfo(sourceinfo, sourcetaskinfo, stype, groupIdTask) //用户组分发任务成功后，给数据源打上用户组标识，同时生成任务临时表
			}
		}
	}
	qu.Debug("Msg：", msg)
	f.ServeJson(map[string]interface{}{"success": success, "msg": msg})
}

// ProjectGetEntnameList 模糊查询公司名称
func (f *Front) ProjectGetEntnameList() {
	defer qu.Catch()
	var entnameList []string
	entname := f.GetString("entname")
	query := map[string]interface{}{
		"username": map[string]interface{}{
			"$regex": entname,
		},
	}
	list, _ := util.MgoJy.Find(util.JyUser, query, nil, map[string]interface{}{"username": 1}, false, -1, -1)
	for _, l := range *list {
		entnameList = append(entnameList, qu.ObjToString(l["username"]))
	}
	f.ServeJson(map[string]interface{}{"entname": entnameList})
}

// ProjectTaskRepulse 用户组任务打回
// TODO 关联用户组下所有用户任务的打回
func (f *Front) ProjectTaskRepulse() {
	defer qu.Catch()
	success := false
	msg := ""
	user := f.GetSession("user").(map[string]interface{})
	username := qu.ObjToString(user["s_name"])
	status := f.GetString("s_status")
	taskId := f.GetString("id")
	groupId := f.GetString("s_groupid")
	sourceinfo := f.GetString("s_sourceinfo")
	sourcetaskinfo := f.GetString("s_sourcetaskinfo")
	if status == "已完成" {
		//更新数据状态
		//1、更新源数据表
		success1 := util.Mgo.Update(sourceinfo, map[string]interface{}{"s_groupid": groupId}, map[string]interface{}{
			"$set": map[string]interface{}{
				"b_istag":      false,
				"i_updatetime": time.Now().Unix(),
			},
			"$unset": map[string]interface{}{
				"s_userid": "",
			},
		}, false, true)
		//2、删除临时任务表中对应数据
		success2 := util.Mgo.Del(sourcetaskinfo, map[string]interface{}{"s_groupid": groupId})
		if success1 && success2 {
			//清除最迟完成时间，更新任务状态
			success = util.Mgo.UpdateById(util.TASKCOLLNAME, taskId, map[string]interface{}{
				"$set": map[string]interface{}{
					"s_status":       "未开始",
					"s_updateperson": username,
					"i_updatetime":   time.Now().Unix(),
					"s_progress":     "0%",
				},
				"$unset": map[string]interface{}{
					"i_completetime": "",
				},
			})
			if !success {
				msg += "更新任务：" + taskId + "失败"
			}
		} else {
			qu.Debug("Update "+sourceinfo+"：", success1, "	Delete "+sourcetaskinfo+"：", success2)
			if !success1 {
				msg += "更新" + sourceinfo + "数据失败;"
			}
			if !success2 {
				msg += "删除" + sourcetaskinfo + "数据失败;"
			}
		}
	}
	qu.Debug("Task Repulse:", success, "	Msg：", msg)
	f.ServeJson(map[string]interface{}{"success": success, "msg": msg})
}

// ProjectTaskRetrieve 用户组任务收回
// TODO 关联用户组下所有用户任务的收回
func (f *Front) ProjectTaskRetrieve() {
	defer qu.Catch()
	success := false
	msg := ""
	num := 0
	user := f.GetSession("user").(map[string]interface{})
	username := qu.ObjToString(user["s_name"])
	status := f.GetString("s_status")
	taskId := f.GetString("id")
	groupId := f.GetString("s_groupid")
	sourceinfo := f.GetString("s_sourceinfo")
	sourcetaskinfo := f.GetString("s_sourcetaskinfo")
	if status == "已完成" {
		count1 := util.Mgo.Count(sourceinfo, map[string]interface{}{"s_groupid": groupId, "b_istag": false})
		count2 := util.Mgo.Count(sourcetaskinfo, map[string]interface{}{"s_groupid": groupId, "s_complete": false})
		if count1 != count2 { //数据源表和临时表数量不一致
			qu.Debug("Count Is Not Same：", sourceinfo+"：", count1, sourceinfo+"：", count2)
		}
		num = count1
		if count1 == 0 { //收回数据量为0
			success = true
			//更新任务状态、完成时间
			success = util.Mgo.UpdateById(util.TASKCOLLNAME, taskId, map[string]interface{}{
				"$set": map[string]interface{}{
					"s_status":       "已完成",
					"s_updateperson": username,
					"i_updatetime":   time.Now().Unix(),
					"i_completetime": time.Now().Unix(),
					"s_progress":     "100%",
				},
			})
			if !success {
				msg += "更新任务：" + taskId + "失败"
			}
		} else {
			//1、更新源数据表
			success1 := util.Mgo.Update(sourceinfo, map[string]interface{}{"s_groupid": groupId, "b_istag": false}, map[string]interface{}{
				"$set": map[string]interface{}{
					"b_isgive":     false,
					"i_updatetime": time.Now().Unix(),
				},
				"$unset": map[string]interface{}{
					"s_groupid": "",
				},
			}, false, true)
			//2、删除临时任务表中对应未完成数据
			success2 := util.Mgo.Del(sourcetaskinfo, map[string]interface{}{"s_groupid": groupId, "s_complete": false})
			if success1 && success2 {
				//更新任务状态、完成时间
				success = util.Mgo.UpdateById(util.TASKCOLLNAME, taskId, map[string]interface{}{
					"$set": map[string]interface{}{
						"s_status":       "已完成",
						"s_updateperson": username,
						"i_updatetime":   time.Now().Unix(),
						"i_completetime": time.Now().Unix(),
					},
				})
				if !success {
					msg += "更新任务：" + taskId + "失败"
				}
			} else {
				qu.Debug("Update "+sourceinfo+"：", success1, "	Delete "+sourcetaskinfo+"：", success2)
				if !success1 {
					msg += "更新" + sourceinfo + "数据失败;"
				}
				if !success2 {
					msg += "删除" + sourcetaskinfo + "数据失败;"
				}
			}
		}
	}
	qu.Debug("Task Retrieve:", success, "	num：", num, "	Msg：", msg)
	f.ServeJson(map[string]interface{}{"success": success, "msg": msg, "num": num})
}

// ProjectTaskClose 用户组任务关闭
func (f *Front) ProjectTaskClose() {
	defer qu.Catch()
	success := false
	user := f.GetSession("user").(map[string]interface{})
	username := qu.ObjToString(user["s_name"])
	status := f.GetString("s_status")
	taskId := f.GetString("id")
	if status == "已完成" {
		//更新任务状态
		success = util.Mgo.UpdateById(util.TASKCOLLNAME, taskId, map[string]interface{}{
			"$set": map[string]interface{}{
				"s_status":       "已关闭",
				"s_updateperson": username,
				"i_updatetime":   time.Now().Unix(),
				"s_progress":     "100%",
			},
		})
	}
	f.ServeJson(map[string]interface{}{"success": success})
}

// DeleleDataTagInfo 删除标注记录
func DeleleDataTagInfo(sourceinfo string) {
	defer qu.Catch()
	sess := util.Mgo.GetMgoConn()
	defer util.Mgo.DestoryMongoConn(sess)
	ch := make(chan bool, 5)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	query := map[string]interface{}{ //达标数据可能会分发后收回、打回再分发
		"b_istagging": false, //达标数据
		"b_cleartag":  false, //未进行一次标注信息清理
	}
	fields := map[string]interface{}{
		"v_taginfo": 1,
		"v_check":   1,
	}
	updateArr := [][]map[string]interface{}{}
	it := sess.DB(util.Mgo.DbName).C(sourceinfo).Find(&query).Select(&fields).Iter()
	count, _ := sess.DB(util.Mgo.DbName).C(sourceinfo).Find(&query).Count()
	qu.Debug("Find Needs To Clearn Data Count：", count)
	n := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			update := []map[string]interface{}{}
			update = append(update, map[string]interface{}{"_id": tmp["_id"]})
			tagInfo, _ := tmp["v_taginfo"].(map[string]interface{})
			checkInfo, _ := tmp["v_check"].(map[string]interface{})
			if len(tagInfo) == 0 || len(checkInfo) == 0 {
				return
			}
			for f, _ := range tagInfo {
				if checkInfo[f] == nil {
					delete(tagInfo, f)
				}
			}
			update = append(update, map[string]interface{}{
				"$set": map[string]interface{}{
					"v_taginfo":  tagInfo,
					"b_cleartag": true,
				},
			})
			lock.Lock()
			updateArr = append(updateArr, update)
			if len(updateArr) > 500 {
				util.Mgo.UpdateBulk(sourceinfo, updateArr...)
				updateArr = [][]map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%100 == 0 {
			qu.Debug("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	lock.Lock()
	if len(updateArr) > 0 {
		util.Mgo.UpdateBulk(sourceinfo, updateArr...)
		updateArr = [][]map[string]interface{}{}
	}
	lock.Unlock()
}

// UpdateSourceinfo 用户组分发任务成功后，给数据源打上用户组标识，同时生成任务临时表
func UpdateSourceinfo(sourceinfo, sourcetaskinfo, stype string, groupIdInfo map[string]util.Task) {
	defer qu.Catch()
	for groupTaskId, tInfo := range groupIdInfo {
		groupid := tInfo.GroupId
		num := tInfo.GiveNum
		sess := util.Mgo.GetMgoConn()
		defer util.Mgo.DestoryMongoConn(sess)
		ch := make(chan bool, 5)
		wg := &sync.WaitGroup{}
		lock := &sync.Mutex{}
		query := map[string]interface{}{ //查找未分配对应stype的数据分发
			"b_isgive": false,
		}
		if stype == "notag" { //达标数据
			query["b_istagging"] = false
		} else if stype == "tag" { //未达标数据
			query["b_istagging"] = true
		}
		fields := map[string]interface{}{
			"title": 1,
		}
		saveArr := []map[string]interface{}{}
		updateArr := [][]map[string]interface{}{}
		qu.Debug("Query：", query)
		it := sess.DB(util.Mgo.DbName).C(sourceinfo).Find(&query).Select(&fields).Limit(int64(num)).Iter()
		n := 0
		for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
			ch <- true
			wg.Add(1)
			go func(tmp map[string]interface{}) {
				defer func() {
					<-ch
					wg.Done()
				}()
				id := mongodb.BsonIdToSId(tmp["_id"])
				title := qu.ObjToString(tmp["title"])
				update := []map[string]interface{}{}
				update = append(update, map[string]interface{}{"_id": tmp["_id"]})
				update = append(update, map[string]interface{}{
					"$set": map[string]interface{}{
						"s_groupid":    groupid,
						"b_isgive":     true,
						"i_updatetime": time.Now().Unix(),
					},
				})
				save := map[string]interface{}{
					"s_infoid":      id,
					"s_infotitle":   title,
					"s_groupid":     groupid,
					"i_createtime":  time.Now().Unix(),
					"s_complete":    false,
					"s_grouptaskid": groupTaskId,
				}
				lock.Lock()
				updateArr = append(updateArr, update)
				saveArr = append(saveArr, save)
				if len(updateArr) > 500 {
					util.Mgo.UpdateBulk(sourceinfo, updateArr...)
					updateArr = [][]map[string]interface{}{}
				}
				if len(saveArr) > 500 {
					util.Mgo.SaveBulk(sourcetaskinfo, saveArr...)
					saveArr = []map[string]interface{}{}
				}
				lock.Unlock()
			}(tmp)
			if n%100 == 0 {
				qu.Debug("current:", n)
			}
			tmp = map[string]interface{}{}
		}
		wg.Wait()
		lock.Lock()
		if len(updateArr) > 0 {
			util.Mgo.UpdateBulk(sourceinfo, updateArr...)
			updateArr = [][]map[string]interface{}{}
		}
		if len(saveArr) > 0 {
			util.Mgo.SaveBulk(sourcetaskinfo, saveArr...)
			saveArr = []map[string]interface{}{}
		}
		lock.Unlock()
	}
}

//ImportDataByExcel 通过excel获取数据源
func ImportDataByExcel(s_sourceinfo string, mf multipart.File, success *bool, msg *string, successNum *int64) (importDataNum int) {
	defer qu.Catch()
	binary, _ := ioutil.ReadAll(mf)
	xls, _ := xlsx.OpenBinary(binary)
	sheet := xls.Sheets[0]
	rows := sheet.Rows
	idcolnum := -1
	cellFieldName := map[int]string{}                //记录客户需求字段所在的列
	idInfoMap := map[string]map[string]interface{}{} //记录数据id及需要保存的字段信息
	for rn, row := range rows {
		if rn == 0 {
			for index, cell := range row.Cells {
				title := cell.Value
				if fieldName := util.CustomerFieldMap_HE[title]; fieldName != "" { //客户需求字段
					cellFieldName[index] = fieldName
				}
				if title == "唯一标识" || title == "信息标识" { //id所在列
					idcolnum = index
				}
			}
			if idcolnum == -1 {
				break
			}
			continue
		}
		if len(row.Cells) < len(rows[0].Cells) {
			break
		}
		tmp := map[string]interface{}{}
		for index, f := range cellFieldName {
			if val := row.Cells[index].Value; val != "" {
				if f == "capital" { //注册资金（万元）
					cf, _ := row.Cells[index].Float()
					tmp[f] = cf
				} else if f == "createtime" { //创建时间
					ci, _ := row.Cells[index].Int64()
					tmp[f] = ci
				} else {
					tmp[f] = val
				}
			}
		}
		id := row.Cells[idcolnum].String() //加密的id
		if id == "" {
			break
		}
		id = util.SE.DecodeString(id) //解密后id
		idInfoMap[id] = tmp
	}
	importDataNum = len(idInfoMap)
	qu.Debug("Load Excel Count：", importDataNum)
	if importDataNum > 0 {
		GetDataById(idInfoMap, "excel", s_sourceinfo, success, msg, successNum)
	} else {
		*success = false
		*msg = "查询数据失败"
	}
	idInfoMap = map[string]map[string]interface{}{}
	return
}

//ImportDataByColl 通过表获取数据源
func ImportDataByColl(s_sourceinfo, historyid string, success *bool, msg *string, successNum *int64) (departname, entname string, rulename []string, importDataNum int) {
	defer qu.Catch()
	rulenameMap := map[string]bool{}
	sess := util.MgoJy.GetMgoConn()
	defer util.MgoJy.DestoryMongoConn(sess)
	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	idInfoMap := map[string]map[string]interface{}{} //记录数据id及需要保存的字段信息
	query := map[string]interface{}{
		"historyId": historyid,
	}
	it := sess.DB(util.MgoJy.DbName).C(util.JyHistory).Find(&query).Iter()
	n := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			id := qu.ObjToString(tmp["id"])                //bidding id
			appid := qu.ObjToString(tmp["appid"])          //根据appid查user表获取公司名称
			departname = qu.ObjToString(tmp["departname"]) //部门名称。所有数据都应部门名称，若不一致，随机取
			needField := map[string]interface{}{}
			for f, _ := range util.CustomerFieldMap_EH {
				if tmp[f] != nil {
					needField[f] = tmp[f]
				}
			}
			if entname == "" { //获取一次公司名称即可
				user, _ := util.MgoJy.FindOne(util.JyUser, map[string]interface{}{"appid": appid})
				entname = qu.ObjToString((*user)["username"]) //公司名称
			}
			rname := qu.ObjToString(tmp["rulename"])
			lock.Lock()
			rulenameMap[rname] = true
			//rulename = append(rulename, qu.ObjToString(tmp["rulename"])) //规则名称
			idInfoMap[id] = needField
			lock.Unlock()
		}(tmp)
		if n%100 == 0 {
			qu.Debug("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	for r, _ := range rulenameMap {
		rulename = append(rulename, r)
	}
	importDataNum = len(idInfoMap) //查询数据总数
	if importDataNum > 0 {
		GetDataById(idInfoMap, "coll", s_sourceinfo, success, msg, successNum)
	} else {
		*msg = "查询数据失败"
	}
	idInfoMap = map[string]map[string]interface{}{}
	return
}

//GetDataById 通过id集从bidding、extract、project获取数据所有信息
func GetDataById(idsInfo map[string]map[string]interface{}, importType, s_sourceinfo string, success *bool, msg *string, successNum *int64) {
	*success = true
	var msgArr []string
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	ch := make(chan bool, 10)
	num := int64(0) //计数
	for id, info := range idsInfo {
		wg.Add(1)
		ch <- true
		go func(id string, tmp map[string]interface{}) {
			defer func() {
				wg.Done()
				<-ch
			}()
			/*
				1.查bidding
				2.查extract
				3.extract合并到bidding(删除item字段与客户需要的item不是一个含义)
				4.对比marked表,替换已标注过的字段值,补充标记
				5.合并客户所需字段信息,补充id字段
				//6.若为同步时,删除原有id对应的信息,新增该id对应的_id信息
				6.mgo查询项目信息
			*/
			tagInfoMap := map[string]interface{}{}  //记录数据已标注过的信息
			baseInfoMap := map[string]interface{}{} //记录其他信息
			//1.查bidding
			tmpBidColl := util.BidColl1 //bidding
			//查询bidding
			if id < util.BIDDINGSTARTID {
				tmpBidColl = util.BidColl2 //bidding_back
			}
			bidData, _ := util.MgoB.FindById(tmpBidColl, id, nil)
			if bidData != nil && len(*bidData) > 0 { //bidding表数据存在
				//2.查extract
				extData, _ := util.MgoE.FindById(util.ExtColl1, id, nil)
				if extData == nil || len(*extData) == 0 {
					extData, _ = util.MgoE.FindById(util.ExtColl2, id, nil)
				}
				//抽取表字段合并到bidding
				if extData != nil && len(*extData) > 0 {
					for k, v := range *extData {
						(*bidData)[k] = v
					}
				}
				//3.删除item
				//删除item
				delete((*bidData), "item")
				//4.对比marked表，对比marked表是否已标注该数据
				markData, _ := util.Mgo.FindById(util.AllToColl, id, nil)
				if markData != nil && len(*markData) > 0 {
					UpdateMarkColl(bidData, markData, &tagInfoMap, &baseInfoMap) //比对更新数据
				} else {
					baseInfoMap["i_ckdata"] = 0 //设置ck_data默认值0
					//多包、中标候选人、标的信息是否抽取
					//if packageMap, ok := (*bidData)["package"].(map[string]interface{}); ok && len(packageMap) > 0 {
					//	baseInfoMap["b_pkgisext"] = true
					//} else {
					//	baseInfoMap["b_pkgisext"] = false
					//}
					//if winorderArr, ok := (*bidData)["winnerorder"].([]interface{}); ok && len(winorderArr) > 0 {
					//	baseInfoMap["b_wodrisext"] = true
					//} else {
					//	baseInfoMap["b_wodrisext"] = false
					//}
					//if purchArr, ok := (*bidData)["purchasinglist"].([]interface{}); ok && len(purchArr) > 0 {
					//	baseInfoMap["b_pclisext"] = true
					//} else {
					//	baseInfoMap["b_pclisext"] = false
					//}
				}
				//合并导入表中客户所需的字段
				if len(tmp) > 0 {
					for k, v := range tmp {
						(*bidData)[k] = v
					}
				}
				//补充id
				//(*bidData)["id"] = id

				//if stype == "syncoll" { //同步数据时删除原始数据
				//	if util.MgoM.Delete(coll, `{"id":"`+id+`"}`) == 0 {
				//		lock.Lock()
				//		*msg += "同步未删除成功数据id:" + id + ";\n"
				//		*success = false
				//		lock.Unlock()
				//	}
				//}
				// 处理	package winner_all
				if p, o1 := (*bidData)["package"].(map[string]interface{}); o1 {
					for _, v := range p {
						v1 := v.(map[string]interface{})
						t := make(map[string]interface{})
						if v1["winner"] != nil {
							t["winner"] = v1["winner"]
						}
						if v1["bidamount"] != nil {
							t["bidamount"] = qu.Float64All(v1["bidamount"])
						}
						if len(t) > 0 {
							v1["winner_all"] = append([]map[string]interface{}{}, t)
						}
					}
				}
				// 补充filetext
				(*bidData)["filetext"] = util.GetFileText(*bidData)
				// 6.es查询项目合并信息
				//esQ := `{"query":{"bool":{"must":[{"term":{"ids":"` + id + `"}}]}}}`
				//info := util.Es.Get("projectset", "projectset", esQ)
				projectId := qu.ObjToString((*bidData)["projectId"])
				project, _ := util.MgoE.FindById(util.ProjectColl, projectId, map[string]interface{}{"ids": 1})
				if project != nil && len(*project) > 0 {
					ids := qu.ObjArrToStringArr((*project)["ids"].([]interface{}))
					if len(ids) > 0 {
						var infolist []map[string]interface{}
						for _, v := range ids {
							if v == id { // 当前公告
								continue
							}
							if v < util.BIDDINGSTARTID {
								tmpBidColl = util.BidColl2 //bidding_back
							}
							bid, b := util.MgoB.FindById(tmpBidColl, v, nil)
							if b && len(*bid) > 0 {
								tmp := make(map[string]interface{})
								tmp["id"] = v
								tmp["title"] = (*bid)["title"]
								tmp["href"] = (*bid)["href"]
								tmp["toptype"] = (*bid)["toptype"]
								tmp["subtype"] = (*bid)["subtype"]
								tmp["publishtime"] = (*bid)["publishtime"]
								tmp["detail"] = (*bid)["detail"]
								tmp["filetext"] = util.GetFileText(*bid)
								infolist = append(infolist, tmp)
							}
						}
						(*bidData)["info"] = infolist
					}
				} else {
					qu.Debug("Projectset Find Error", projectId)
				}
				baseInfoMap["id"] = id
				_id := (*bidData)["_id"]
				delete(*bidData, "_id")
				//保存数据
				baseInfoMap["_id"] = _id
				baseInfoMap["v_baseinfo"] = bidData
				if len(tagInfoMap) > 0 {
					baseInfoMap["v_taginfo"] = tagInfoMap
				}
				baseInfoMap["i_createtime"] = time.Now().Unix()
				baseInfoMap["b_isgive"] = false   //是否分配
				baseInfoMap["b_istag"] = false    //是否已标注
				baseInfoMap["b_cleartag"] = false //是否清理标注信息

				if util.Mgo.SaveByOriID(s_sourceinfo, baseInfoMap) {
					atomic.AddInt64(successNum, 1) //保存成功计数
				} else {
					lock.Lock()
					*success = false
					if importType == "excel" {
						msgArr = append(msgArr, "第"+fmt.Sprint(num+2)+"行未导入id："+id)
						//*msg += "第" + fmt.Sprint(num+2) + "行未保存成功数据_id：" + id + ";\n"
					} else {
						msgArr = append(msgArr, "未导入id:"+id)
						//*msg += "未保存成功数据_id:" + id + ";\n"
					}
					lock.Unlock()
				}
			} else {
				lock.Lock()
				*success = false
				if importType == "excel" {
					msgArr = append(msgArr, "第"+fmt.Sprint(num+2)+"行未查询id："+id)
					//*msg += "第" + fmt.Sprint(num+2) + "行未查询到数据：" + id + ";\n"
				} else {
					msgArr = append(msgArr, "未查询id:"+id)
					//*msg += "未查询到数据_id:" + id + ";\n"
				}
				lock.Unlock()
			}
		}(id, info)
	}
	wg.Wait()
	sort.Strings(msgArr)
	*msg = strings.Join(msgArr, ";\n")
}

// UpdateMarkColl 更新数据
func UpdateMarkColl(bidData, markData, tagInfoMap, baseInfoMap *map[string]interface{}) {
	defer qu.Catch()
	ckdata := qu.IntAll((*markData)["i_ckdata"])
	v_taginfo := (*markData)["v_taginfo"].(map[string]interface{})   //标注信息
	v_datainfo := (*markData)["v_datainfo"].(map[string]interface{}) //基本信息
	for fk, _ := range v_taginfo {
		if v_datainfo[fk] != nil {
			(*bidData)[fk] = v_datainfo[fk] //字段更新
		}
	}
	(*tagInfoMap) = v_taginfo //marked中已有的标注信息保存到新数据上
	if ckdata == 2 {          //某些字段已标注
		(*baseInfoMap)["i_ckdata"] = 0 //marked表中该条数据如果为字段验证，临时表ck_data:0;若为数据验证ck_data:1
	} else if ckdata == 1 {
		(*baseInfoMap)["i_ckdata"] = 1
	}
}