package front import ( "encoding/json" "fmt" "github.com/shopspring/decimal" "github.com/tealeg/xlsx" "go.mongodb.org/mongo-driver/bson/primitive" "io/ioutil" "mime/multipart" "mongodb" qu "qfw/util" "sort" "strconv" "strings" "sync" "sync/atomic" "time" "util" ) // ProjectIsExists 获取所有项目名称 func (f *Front) ProjectIsExists() { defer qu.Catch() name := f.GetString("s_name") exists := false project, _ := util.Mgo.FindOne(util.PROJECTCOLLNAME, map[string]interface{}{"s_name": name}) if project != nil && len(*project) > 0 { exists = true } f.ServeJson(map[string]interface{}{"exists": exists}) } // 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("s_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, map[string]interface{}{"_id": -1}, 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) qu.Debug(query, len(*info)) 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_login"]) //当前登录用户 stype := f.GetString("s_type") //新建项目类型:数据库导入、excel导入 s_sourceinfoTmp := f.GetString("s_sourceinfo") //数据表 s_sourceinfo := "f_sourceinfo_" + s_sourceinfoTmp if stype != "edit" && 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") qu.Debug(s_entname, s_departname, s_rulename) 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", //导入类型 "b_isassessment": false, //是否进行了质量评估 } } 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, //源数据集标识 "b_isassessment": false, //是否进行了质量评估 } } 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, "msg": err}) 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 = "保存项目成功" } } //qu.Debug("Msg:", msg) //返回信息 if stype == "edit" { f.ServeJson(map[string]interface{}{"success": success, "msg": msg}) } else { qu.Debug("Create Project:", success, "importnum:", importDataNum, "successnum:", successNum, "failnum:", int64(importDataNum)-successNum) f.ServeJson(map[string]interface{}{"success": success, "msg": msg, "importnum": importDataNum, "successnum": successNum, "failnum": int64(importDataNum) - successNum}) } } // ProjectComplete 项目提交完成 func (f *Front) ProjectComplete() { defer qu.Catch() user := f.GetSession("user").(map[string]interface{}) username := qu.ObjToString(user["s_login"]) //当前登录用户 success := false msg := "" projectId := f.GetString("s_projectid") sourceInfo := f.GetString("s_sourceinfo") //status := f.GetString("s_status") info, _ := util.Mgo.FindById(util.PROJECTCOLLNAME, projectId, `{"s_status": 1}`) if len(*info) <= 0 { f.ServeJson(map[string]interface{}{"success": false, "msg": "查询项目失败"}) return } status := (*info)["s_status"] if status == "进行中" { //查询该项目下未完成的用户组和用户任务 query := map[string]interface{}{ "s_projectid": projectId, "s_status": map[string]interface{}{ "$in": []string{"未开始", "进行中"}, }, } taskCount := util.Mgo.Count(util.TASKCOLLNAME, query) dataCount := util.Mgo.Count(sourceInfo, map[string]interface{}{"b_istag": false}) //未标注数据个数 qu.Debug("No Tag Count:", dataCount) if dataCount == 0 && taskCount == 0 { //全部完成 success = util.Mgo.UpdateById(util.PROJECTCOLLNAME, projectId, map[string]interface{}{ "$set": map[string]interface{}{ "s_status": "已完成", "i_completetime": time.Now().Unix(), "s_updateperson": username, "i_updatetime": time.Now().Unix(), }, }) if !success { msg = "更新项目失败" } } if taskCount != 0 { msg += "任务未全部完成," } if dataCount != 0 { msg += "数据未全部标注" } } else { msg = "项目未开始" } f.ServeJson(map[string]interface{}{"success": success, "msg": msg}) } // ProjectQualityAssessment 数据质量评估 func (f *Front) ProjectQualityAssessment() { defer qu.Catch() msg := "" success := false //质量评估 projectid := f.GetString("pid") //项目id project, _ := util.Mgo.FindById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{"b_isassessment": 1, "s_sourceinfo": 1, "v_fields": 1}) if project != nil && len(*project) > 0 { if isAssessment, ok := (*project)["b_isassessment"].(bool); ok && !isAssessment { if fields, ok := (*project)["v_fields"].(map[string]interface{}); ok && len(fields) > 0 { var fieldsArr []string for f, _ := range fields { fieldsArr = append(fieldsArr, f) } sourceinfo := qu.ObjToString((*project)["s_sourceinfo"]) qu.Debug("质量评估字段:", fieldsArr) success = QuaFieldScore(fieldsArr, sourceinfo) //调用数据质量评估接口 if success { //点击清洗更新项目状态为进行中 b := util.Mgo.UpdateById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{"$set": map[string]interface{}{"b_isassessment": true, "s_status": "进行中", "i_starttime": time.Now().Unix()}}) qu.Debug("Update Porject:"+projectid+" Status Success:", b) } else { msg = "质量评估失败" } } else { msg = "项目标注字段查询失败" } } else if ok && isAssessment { success = true } else { msg = "查询项目失败" } } else { msg = "查询项目失败" } f.ServeJson(map[string]interface{}{"success": success, "msg": msg}) } // ProjectGroupTaskList 用户组任务分发列表 func (f *Front) ProjectGroupTaskList() { defer qu.Catch() projectid := f.GetString("pid") //项目id if f.Method() == "POST" { 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}}, } } qu.Debug("Query:", query) list, _ := util.Mgo.Find(util.TASKCOLLNAME, query, map[string]interface{}{"_id": -1}, 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"]) groupTaskId := mongodb.BsonIdToSId(l["_id"]) giveNum := qu.IntAll(l["i_givenum"]) sourceinfo := qu.ObjToString(l["s_sourceinfo"]) //tagNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_isgivegroup": true, "s_grouptaskid": groupTaskId, "b_istag": true}) tagNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"s_grouptaskid": groupTaskId, "b_istag": true}) progressFloat := float64(tagNum) / float64(giveNum) value, _ := strconv.ParseFloat(fmt.Sprintf("%.4f", progressFloat), 64) decimalValue := decimal.NewFromFloat(value) decimalValue = decimalValue.Mul(decimal.NewFromInt(100)) value, _ = decimalValue.Float64() progress := fmt.Sprint(value) + "%" 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}) } else { project, _ := util.Mgo.FindById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{"s_sourceinfo": 1}) if project != nil && len(*project) > 0 { sourceinfo := qu.ObjToString((*project)["s_sourceinfo"]) //数据源表 okAllDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": false}) //达标数据总量 okIsGiveDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": false, "b_isgivegroup": true}) //达标数据已分发量 okNotGiveDataNum := okAllDataNum - okIsGiveDataNum //达标待分发量 okIsTagDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": false, "b_istag": true}) //达标已标注量 IsNoOkAllDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": true}) //未达标数据总量 IsNoOkIsGiveDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": true, "b_isgivegroup": true}) //未达标数据已分发量 IsNotOkNotGiveDataNum := IsNoOkAllDataNum - IsNoOkIsGiveDataNum //未达标待分发量 IsNotOkIsTagDataNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_istagging": true, "b_istag": true}) //未达标已标注量 allGiveDataNum := okIsGiveDataNum + IsNoOkIsGiveDataNum //总分发量 allNoGiveDataNum := okNotGiveDataNum + IsNotOkNotGiveDataNum //总待分发量 allIsTagDataNum := okIsTagDataNum + IsNotOkIsTagDataNum //已标注总量 allDataNum := allGiveDataNum + allNoGiveDataNum // 查询全部实际可分发数据量(未分发、未标注) okRealGiveNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_isgivegroup": false, "b_istag": false}) //qu.Debug("数据总量:", allDataNum, "已分发总量:", allGiveDataNum, "待分发总量:", allNoGiveDataNum, "已标注总量:", allIsTagDataNum) //qu.Debug("达标量:", okAllDataNum, "达标已分发量:", okIsGiveDataNum, "达标待分发量:", okNotGiveDataNum, "达标已标注量:", okIsTagDataNum) //qu.Debug(" 未达标量:", IsNoOkAllDataNum, " 未达标已分发量:", IsNoOkIsGiveDataNum, " 未达标待分发量:", IsNotOkNotGiveDataNum, " 未达标已标注量:", IsNotOkIsTagDataNum) f.T["s_projectid"] = projectid f.T["s_sourceinfo"] = sourceinfo f.T["allDataNum"] = allDataNum f.T["okAllDataNum"] = okAllDataNum f.T["okIsGiveDataNum"] = okIsGiveDataNum f.T["okNotGiveDataNum"] = okNotGiveDataNum f.T["IsNoOkAllDataNum"] = IsNoOkAllDataNum f.T["IsNoOkIsGiveDataNum"] = IsNoOkIsGiveDataNum f.T["IsNotOkNotGiveDataNum"] = IsNotOkNotGiveDataNum f.T["allGiveDataNum"] = allGiveDataNum f.T["allNoGiveDataNum"] = allNoGiveDataNum f.T["allIsTagDataNum"] = allIsTagDataNum f.T["okIsTagDataNum"] = okIsTagDataNum f.T["IsNotOkIsTagDataNum"] = IsNotOkIsTagDataNum f.T["okRealGiveNum"] = okRealGiveNum _ = f.Render("project/project_clear.html", &f.T) } else { qu.Debug("Project Find Error") f.ServeJson("查询项目失败") } } } // ProjectGroupTaskSave 用户组任务分发 func (f *Front) ProjectGroupTaskSave() { defer qu.Catch() var groupArr []map[string]interface{} var taskArr []map[string]interface{} var groupIdArr []string var groupTaskIdArr []string groupIdTask := map[string]util.Task{} success := false msg := "" user := f.GetSession("user").(map[string]interface{}) username := qu.ObjToString(user["s_login"]) //当前登录用户 projectid := f.GetString("s_projectid") //项目标识 project, _ := util.Mgo.FindById(util.PROJECTCOLLNAME, projectid, nil) projectname := qu.ObjToString((*project)["s_name"]) //项目名称 sourceinfo := qu.ObjToString((*project)["s_sourceinfo"]) //源数据表 group := f.GetString("s_group") stype := f.GetString("s_type") qu.Debug("项目id:", projectid, " 项目名称:", projectname) if err := json.Unmarshal([]byte(group), &groupArr); err != nil { qu.Debug("GroupInfo Unmarshal Failed:", err) msg = "用户组信息解析失败" } else { qu.Debug("用户组信息:", groupArr, stype) //if stype != "tag" { //如果分发的是达标数据或者全部数据且进行了初步质检,将没有质检记录的字段从v_taginfo标注记录中删除 // DeleleDataTagInfo(sourceinfo) //} // 查询实际可分发数据量(未分发、未标注) realNum := util.Mgo.Count(sourceinfo, map[string]interface{}{"b_isgivegroup": false, "b_istag": false}) for _, groupInfo := range groupArr { groupId := qu.ObjToString(groupInfo["s_groupid"]) groupIdArr = append(groupIdArr, groupId) givenum := qu.IntAll(groupInfo["i_givenum"]) if givenum <= 0 { continue } if realNum <= 0 { break } if realNum < givenum { givenum = realNum } groupTaskId := primitive.NewObjectID() groupTaskIdStr := mongodb.BsonIdToSId(groupTaskId) groupTaskIdArr = append(groupTaskIdArr, groupTaskIdStr) gt := util.Task{ UserId: groupId, GiveNum: givenum, } groupIdTask[groupTaskIdStr] = gt groupTask := map[string]interface{}{ "_id": groupTaskId, //生成任务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_stype": "group", //任务类型 "s_entname": qu.ObjToString((*project)["s_entname"]), //公司名称 "s_departname": qu.ObjToString((*project)["s_departname"]), //部门名称 "s_rulename": qu.ObjToString((*project)["s_rulename"]), //规则名称 } realNum = realNum - givenum taskArr = append(taskArr, groupTask) } } if len(taskArr) > 0 { //分发数据后更新项目中用户组标识信息和用户组任务id success = util.Mgo.UpdateById(util.PROJECTCOLLNAME, projectid, map[string]interface{}{ "$push": map[string]interface{}{ "v_groupids": map[string]interface{}{ "$each": groupIdArr, }, "v_grouptaskids": map[string]interface{}{ "$each": groupTaskIdArr, }, }, }) if !success { msg = "更新项目:" + projectname + "用户组标识失败" } else { //用户组分发任务 success = util.Mgo.SaveBulk(util.TASKCOLLNAME, taskArr...) if success { msg = "任务分发成功" UpdateSourceInfoByGroup(sourceinfo, stype, groupIdTask) //用户组分发任务成功后,给数据源打上用户组标识 } else { msg = "任务分发失败" } } } qu.Debug("Success:", success, "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) if len(*list) > 0 { for _, l := range *list { entnameList = append(entnameList, qu.ObjToString(l["username"])) } f.ServeJson(map[string]interface{}{"entname": entnameList}) } else { f.ServeJson(map[string]interface{}{"entname": []string{}}) } } // ProjectGroupTaskRepulse 用户组任务打回 func (f *Front) ProjectGroupTaskRepulse() { defer qu.Catch() success := false msg := "" user := f.GetSession("user").(map[string]interface{}) username := qu.ObjToString(user["s_login"]) //status := f.GetString("s_status") groupTaskId := f.GetString("taskid") sourceinfo := f.GetString("s_sourceinfo") currenttime := time.Now().Unix() //更新数据源 success = util.Mgo.Update(sourceinfo, map[string]interface{}{"s_grouptaskid": groupTaskId}, map[string]interface{}{ "$set": map[string]interface{}{ "b_istag": false, "b_check": false, // 质检标记 "i_ckdata": 0, "b_isgiveuser": false, "i_updatetime": currenttime, }, "$unset": map[string]interface{}{ "s_userid": "", "s_usertaskid": "", "s_login": "", "v_taginfo": "", "v_checkinfo": "", }, }, false, true) if success { util.Mgo.Update(util.TASKCOLLNAME, map[string]interface{}{ "s_stype": "user", "s_parentid": groupTaskId, }, map[string]interface{}{ "$set": map[string]interface{}{ "s_status": "已关闭", "s_progress": "%0", "i_updatetime": currenttime, "s_updateperson": username, }, }, false, true) //更新用户组任务 清除最迟完成时间,更新任务状态 success = util.Mgo.UpdateById(util.TASKCOLLNAME, groupTaskId, map[string]interface{}{ "$set": map[string]interface{}{ "s_status": "未开始", "s_updateperson": username, "i_updatetime": currenttime, "s_progress": "0%", }, "$unset": map[string]interface{}{ "i_completetime": "", }, }) if !success { msg = "更新用户组任务失败" } } else { msg = "更新数据源信息失败" } qu.Debug("Task Repulse:", success, " Msg:", msg) f.ServeJson(map[string]interface{}{"success": success, "msg": msg}) } // ProjectGroupTaskRetrieve 用户组任务收回 func (f *Front) ProjectGroupTaskRetrieve() { defer qu.Catch() user := f.GetSession("user").(map[string]interface{}) username := qu.ObjToString(user["s_login"]) groupTaskId := f.GetString("taskid") sourceInfo := f.GetString("s_sourceinfo") //status := f.GetString("s_status") //未开始、进行中 //giveNum, _ := f.GetInteger("i_givenum") //收回时要更新的分发数据量 msg := "" success := false count := 0 groupTask, _ := util.Mgo.FindById(util.TASKCOLLNAME, groupTaskId, map[string]interface{}{"v_sonids": 1, "s_status": 1}) if len(*groupTask) <= 0 { f.ServeJson(map[string]interface{}{"success": false, "msg": "查询任务失败"}) return } status := qu.ObjToString((*groupTask)["s_status"]) if status == "未开始" { //未开始的用户组任务,暂未给用户分发任务 success = true } else { //进行中的用户组任务需更新其下用户信息 if groupTask != nil && len(*groupTask) > 0 { if sonIds, ok := (*groupTask)["v_sonids"].([]interface{}); ok && len(sonIds) > 0 { //更新所有用户任务信息 userTaskIdStatus := map[string]string{} //封装要回收的用户任务信息 for _, sId := range sonIds { userTaskId := qu.ObjToString(sId) userTask, _ := util.Mgo.FindById(util.TASKCOLLNAME, userTaskId, map[string]interface{}{"s_status": 1}) if userTask != nil && len(*userTask) > 0 { if statusTmp := qu.ObjToString((*userTask)["s_status"]); statusTmp != "已完成" && statusTmp != "已关闭" { //已完成和已关闭的任务不收回 userTaskIdStatus[userTaskId] = statusTmp } } else { qu.Debug("Find User Task:", userTaskId, "Error") } } qu.Debug("userTaskIdStatus:", len(userTaskIdStatus)) if len(userTaskIdStatus) > 0 { //收回用户组下所有用户信息 //用户组收回时,若已有用户任务在未开始时收回或关闭,调用RetrieveTaskByUser返回的总收回量count就遗漏了用户收回或关闭任务的量 msg, _, success = RetrieveCloseTaskByUser(sourceInfo, username, userTaskIdStatus) //用户信息收回 } else { //用户组下所有用户任务都已完成 success = true } } else { //没有分配给用户任务 success = true } } else { msg = "用户组任务查找失败" } } if success { //所有用户信息收回成功后,更新用户组任务相关信息 count = util.Mgo.Count(sourceInfo, map[string]interface{}{ //统计该用户组任务下未标注的数据量 "s_grouptaskid": groupTaskId, "b_istag": false, }) UpdateGroupTaskAndSourceInfo(groupTaskId, sourceInfo, username, status, count, &msg, &success) } qu.Debug(success, count, msg) f.ServeJson(map[string]interface{}{"success": success, "msg": msg, "count": count}) } // ProjectGroupTaskClose 用户组任务关闭 func (f *Front) ProjectGroupTaskClose() { defer qu.Catch() user := f.GetSession("user").(map[string]interface{}) username := qu.ObjToString(user["s_login"]) groupTaskId := f.GetString("taskid") sourceInfo := f.GetString("s_sourceinfo") //status := f.GetString("s_status") //未开始、进行中 msg := "" success := false count := 0 groupTask, _ := util.Mgo.FindById(util.TASKCOLLNAME, groupTaskId, map[string]interface{}{"s_status": 1}) if len(*groupTask) <= 0 { f.ServeJson(map[string]interface{}{"success": false, "msg": "查询任务失败"}) return } status := qu.ObjToString((*groupTask)["s_status"]) if status == "未开始" { //未开始的用户组任务,暂未给用户分发任务;已完成只更新用户组任务 success = true } else { //进行中的用户组任务需更新其下用户信息 //groupTask, _ := util.Mgo.FindById(util.TASKCOLLNAME, groupTaskId, map[string]interface{}{"v_sonids": 1}) //if groupTask != nil && len(*groupTask) > 0 { // if sonIds, ok := (*groupTask)["v_sonids"].([]interface{}); ok && len(sonIds) > 0 { //更新所有用户任务信息 // userTaskIdStatus := map[string]string{} //封装要关闭的用户任务信息 // for _, sId := range sonIds { // userTaskId := qu.ObjToString(sId) // userTask, _ := util.Mgo.FindById(util.TASKCOLLNAME, userTaskId, map[string]interface{}{"s_status": 1}) // if userTask != nil && len(*userTask) > 0 { // if statusTmp := qu.ObjToString((*userTask)["s_status"]); statusTmp == "已完成" && statusTmp != "已关闭" { //已关闭的任务不更新 // userTaskIdStatus[userTaskId] = statusTmp // } // } else { // qu.Debug("Find User Task:", userTaskId, "Error") // } // } // if len(userTaskIdStatus) > 0 { //关闭用户组下所有用户信息 // msg, _, success = RetrieveCloseTaskByUser(sourceInfo, username, userTaskIdStatus) //用户信息收回 // } else { //用户组下所有用户任务都已关闭 // success = true // } // } else { //没有分配给用户任务 // success = true // } //} else { // msg = "用户组任务查找失败" //} } if success { //所有用户信息关闭成功后,更新用户组任务相关信息 count = util.Mgo.Count(sourceInfo, map[string]interface{}{ //统计该用户组任务下未标注的数据量 "s_grouptaskid": groupTaskId, "b_istag": false, }) UpdateGroupTaskAndSourceInfo(groupTaskId, sourceInfo, username, status, count, &msg, &success) } qu.Debug(success, count, msg) f.ServeJson(map[string]interface{}{"success": success, "msg": msg, "count": count}) } // UpdateGroupTaskAndSourceInfo 更新用户组任务相关信息 func UpdateGroupTaskAndSourceInfo(groupTaskId, sourceInfo, username, status string, count int, msg *string, success *bool) { defer qu.Catch() qu.Debug("GroupTaskStatus:", status, " Count:", count) if count != 0 { //更新数据源 query := map[string]interface{}{ "s_grouptaskid": groupTaskId, "b_istag": false, } set := map[string]interface{}{ "b_isgivegroup": false, "i_updatetime": time.Now().Unix(), } unset := map[string]interface{}{ "s_groupid": "", "s_grouptaskid": "", } *success = util.Mgo.Update(sourceInfo, query, map[string]interface{}{"$set": set, "$unset": unset}, false, true) } //更新用户组任务 if *success { taskSet := map[string]interface{}{ "s_status": "已完成", "s_updateperson": username, "i_updatetime": time.Now().Unix(), "i_completetime": time.Now().Unix(), "s_progress": "100%", } if status == "未开始" { taskSet["i_starttime"] = time.Now().Unix() taskSet["s_status"] = "已关闭" } inc := map[string]interface{}{ "i_givenum": -count, } *success = util.Mgo.UpdateById(util.TASKCOLLNAME, groupTaskId, map[string]interface{}{"$set": taskSet, "$inc": inc}) if !*success { *msg = "更新用户组任务失败" } } else { *msg = "更新数据源信息失败" } } // 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{}) set := map[string]interface{}{ "b_cleartag": true, } unset := map[string]interface{}{} if len(tagInfo) != 0 && len(checkInfo) != 0 { //有质检信息,删除v_taginfo未质检的字段 for f, _ := range tagInfo { if checkInfo[f] == nil { delete(tagInfo, f) } } set["v_taginfo"] = tagInfo } else if len(tagInfo) != 0 && len(checkInfo) == 0 { //没有质检删除v_taginfo字段 unset["v_taginfo"] = "" } if len(unset) > 0 { update = append(update, map[string]interface{}{ "$set": set, "$unset": unset, }) } else { update = append(update, map[string]interface{}{ "$set": set, }) } 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() } // UpdateSourceInfoByGroup 用户组分发任务成功后,给数据源打上用户组标识 func UpdateSourceInfoByGroup(sourceinfo, stype string, groupIdInfo map[string]util.Task) { defer qu.Catch() for groupTaskId, tInfo := range groupIdInfo { groupId := tInfo.UserId 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_isgivegroup": false, "b_istag": false, } if stype == "notag" { //达标数据 query["b_istagging"] = false } else if stype == "tag" { //未达标数据 query["b_istagging"] = true } fields := map[string]interface{}{ "v_baseinfo": 1, } 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() }() 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, "s_grouptaskid": groupTaskId, "b_isgivegroup": true, "i_updatetime": time.Now().Unix(), }, // 分发时 删除程序检测的标记 "$unset": map[string]interface{}{ "s_excp": "", "s_excp_info": "", }, }) lock.Lock() updateArr = append(updateArr, update) //saveArr = append(saveArr, save) 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() } } //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{} //记录客户需求字段所在的列 idInfoArr := []util.Data{} //记录数据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 idInfoArr = append(idInfoArr, util.Data{ ID: id, Info: tmp, }) //idInfoMap[id] = tmp } importDataNum = len(idInfoArr) qu.Debug("Load Excel Count:", importDataNum) if importDataNum > 0 { GetDataById(idInfoArr, "excel", s_sourceinfo, success, msg, successNum) if *success { if !util.Mgo.CreateIndex(s_sourceinfo, util.SourceInfoIndexArr) { //创建数据源表同时生成字段索引 qu.Debug("创建数据源表:", s_sourceinfo, "失败") return } } } else { *success = false *msg = "查询数据失败" } idInfoArr = []util.Data{} 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{} idInfoArr := []util.Data{} //记录数据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() for _, r := range strings.Split(rname, ",") { rulenameMap[r] = true } //rulename = append(rulename, qu.ObjToString(tmp["rulename"])) //规则名称 //idInfoMap[id] = needField idInfoArr = append(idInfoArr, util.Data{ ID: id, Info: 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(idInfoArr) //查询数据总数 if importDataNum > 0 { GetDataById(idInfoArr, "coll", s_sourceinfo, success, msg, successNum) if *success { if !util.Mgo.CreateIndex(s_sourceinfo, util.SourceInfoIndexArr) { //创建数据源表同时生成字段索引 qu.Debug("创建数据源表:", s_sourceinfo, "失败") return } } } else { *msg = "查询数据失败" } idInfoArr = []util.Data{} return } //GetDataById 通过id集从bidding、extract、project获取数据所有信息 func GetDataById(idInfoArr []util.Data, 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 i, data := range idInfoArr { wg.Add(1) ch <- true go func(index int, tmpData util.Data) { 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{}{} //记录其他信息 id := tmpData.ID tmp := tmpData.Info //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, map[string]interface{}{"attach_text": 0}) if extData == nil || len(*extData) == 0 { extData, _ = util.MgoE.FindById(util.ExtColl2, id, map[string]interface{}{"attach_text": 0}) } //抽取表字段合并到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((*info)[0]["_id"]) if projectId != "" && len(projectId) == 24 { //舍弃项目id为空和合并错误的 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) } } else { qu.Debug("ProjectId Is Null", id) } 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_isgivegroup"] = false //是否分配给用户组 baseInfoMap["b_istag"] = false //是否已标注 //baseInfoMap["b_cleartag"] = false //是否清理标注信息 baseInfoMap["b_isgiveuser"] = false //是否分配给用户 baseInfoMap["b_check"] = 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(index+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(index+2)+"行,未查询到id:"+id) //*msg += "第" + fmt.Sprint(num+2) + "行未查询到数据:" + id + ";\n" } else { msgArr = append(msgArr, "未查询id:"+id) //*msg += "未查询到数据_id:" + id + ";\n" } lock.Unlock() } }(i, data) } wg.Wait() sort.Strings(msgArr) *msg = strings.Join(msgArr, ";
") } // 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_baseinfo := (*markData)["v_baseinfo"].(map[string]interface{}) //基本信息 for fk, _ := range v_taginfo { if v_baseinfo[fk] != nil { (*bidData)[fk] = v_baseinfo[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 } } func (f *Front) ProjectCheckSuc() { defer qu.Catch() if f.Method() == "POST" { sourceinfo := f.GetString("s_sourceinfo") query := map[string]interface{}{ "b_istagging": false, } b := util.Mgo.Update(sourceinfo, query, map[string]interface{}{"$set": map[string]interface{}{"b_istag": true}}, false, true) f.ServeJson(map[string]interface{}{"success": b, "msg": "更新数据失败"}) } } func (f *Front) ProjectPassSuc() { defer qu.Catch() if f.Method() == "POST" { sourceinfo := f.GetString("s_sourceinfo") query := map[string]interface{}{ "b_istag": false, } b := util.Mgo.Update(sourceinfo, query, map[string]interface{}{"$set": map[string]interface{}{"b_istag": true, "i_ckdata": 2}}, false, true) f.ServeJson(map[string]interface{}{"success": b, "msg": "更新数据失败"}) } } func (f *Front) ProjectTagNum() { defer qu.Catch() if f.Method() == "POST" { sourceinfo := f.GetString("s_sourceinfo") count := util.Mgo.Count(sourceinfo, map[string]interface{}{"i_ckdata": 2}) if count > 0 { f.ServeJson(map[string]interface{}{"success": true}) } else { f.ServeJson(map[string]interface{}{"success": false, "msg": "暂时没有可质检的数据"}) } } }