From f4e8f206a6760bdc31734dfcb1c65916b5b76311 Mon Sep 17 00:00:00 2001 From: liuxiaolong <736321739@qq.com> Date: 星期四, 22 八月 2019 12:05:58 +0800 Subject: [PATCH] add cluster --- controllers/fileController.go | 529 +++++++++++++++++++++++++++++++++++++++++----------------- 1 files changed, 371 insertions(+), 158 deletions(-) diff --git a/controllers/fileController.go b/controllers/fileController.go index 8bb9a36..a4c67ae 100644 --- a/controllers/fileController.go +++ b/controllers/fileController.go @@ -1,16 +1,14 @@ package controllers import ( - "basic.com/pubsub/protomsg.git" + "basic.com/dbapi.git" "basic.com/fileServer/WeedFSClient.git" - "basic.com/valib/deliver.git" + "basic.com/pubsub/protomsg.git" "bytes" "encoding/base64" - "encoding/json" "errors" "fmt" "github.com/gin-gonic/gin" - "github.com/gogo/protobuf/proto" "github.com/satori/go.uuid" "image" "image/jpeg" @@ -19,8 +17,10 @@ "mime/multipart" "net/http" "path" + "sort" "strconv" "strings" + "sync" "time" "webserver/extend/code" "webserver/extend/config" @@ -152,6 +152,7 @@ type CompareResult struct { Id string `json:"id"` + CompareScore float32 `json:"compareScore"` CameraId string `json:"cameraId"` CameraAddr string `json:"cameraAddr"` PicDate string `json:"picDate"` @@ -162,9 +163,11 @@ Sex string `json:"sex"` AgeDescription string `json:"ageDescription"` Race string `json:"race"` + TaskId string `json:"taskId"` TaskName string `json:"taskName"` BaseInfo []DbPersonVo `json:"baseInfo"` VideoUrl string `json:"videoUrl"` + SdkName string `json:"sdkName"` } type DbPersonVo struct { BwType string `json:"bwType"` @@ -178,128 +181,347 @@ Sex string `json:"sex"` TableId string `json:"tableId"` TableName string `json:"tableName"` + Enable int32 `json:"enable"` } -var captureTable = "capturetable" // 鎶撴媿搴� -type SearchCondition struct { - PicUrl string `json:"picUrl"` - Databases []string `json:"databases"` - Threshold string `json:"threshold"` +type ScoreIndex struct { + CompareScore float32 + Index int } + +//濉厖鍚戝墠绔繑鍥炵殑鏁版嵁 +func FillDataToCompareResult(compResult *protomsg.SdkCompareResult) []CompareResult { + + var resultList = make([]CompareResult, len(compResult.CompareResult)) + dbPersonM := make(map[string]ScoreIndex, 0) + captureM := make(map[string]ScoreIndex, 0) + personIds := make([]string,0) + captureIds := make([]string,0) + for idx,v :=range compResult.CompareResult{ + if v.Tableid == service.CaptureTable { + captureM[v.Id] = ScoreIndex{ + Index: idx, + CompareScore: v.CompareScore, + } + captureIds = append(captureIds,v.Id) + } else { + dbPersonM[v.Id] = ScoreIndex{ + Index: idx, + CompareScore: v.CompareScore, + } + personIds = append(personIds,v.Id) + } + } + logger.Debug("comp len(personIds):", len(personIds)) + logger.Debug("comp len(captureIds):", len(captureIds)) + + var dbpersons []protomsg.Dbperson + if len(personIds) >0 { + var dbpApi dbapi.DbPersonApi + dbpersons, _ = dbpApi.Dbpersoninfosbyid(personIds) + } + + if len(dbpersons) >0 { + var dtApi dbapi.DbTableApi + for _,p :=range dbpersons { + var dbP = DbPersonVo { + PersonId: p.Id, + IdCard: p.IdCard, + CompareScore: util.ParseScore(dbPersonM[p.Id].CompareScore), + MonitorLevel: p.MonitorLevel, + PersonName: p.PersonName, + PersonPicUrl: p.PersonPicUrl, + PhoneNum: p.PhoneNum, + Sex: p.Sex, + TableId: p.TableId, + Enable: p.Enable, + } + dbTableInfos, _ := dtApi.DbtablesById([]string{ p.TableId }) + if dbTableInfos !=nil{ + dbP.BwType = dbTableInfos[0].BwType + dbP.TableName = dbTableInfos[0].TableName + } + + resultList[dbPersonM[p.Id].Index] = CompareResult{ + BaseInfo:[]DbPersonVo{ dbP }, + } + } + } + if len(captureIds) >0 { + videopersons, _ := esApi.Videopersonsinfosbyid(captureIds, config.EsInfo.EsIndex.VideoPersons.IndexName, config.EsInfo.Masterip, config.EsInfo.Httpport) + logger.Debug("comp videoPersons.len:",len(videopersons)) + for _,vp :=range videopersons { + isAlarmInt, _ := strconv.Atoi(vp.IsAlarm) + bi := make([]DbPersonVo,0) + for _,p :=range vp.BaseInfo { + bi = append(bi, DbPersonVo{ + PersonId: p.PersonId, + IdCard: p.IdCard, + CompareScore: util.ParseScore(p.CompareScore), + MonitorLevel: parseMonitorLevel(p.MonitorLevel), + PersonName: p.PersonName, + PersonPicUrl: p.PersonPicUrl, + PhoneNum: p.PhoneNum, + Sex: p.Sex, + TableId: p.TableId, + BwType: strconv.Itoa(int(p.BwType)), + TableName: p.TableName, + }) + } + vpE := CompareResult{ + Id: vp.Id, + CompareScore: util.ParseScore(captureM[vp.Id].CompareScore), + CameraId: vp.CameraId, + CameraAddr: vp.CameraAddr, + PicDate: vp.PicDate, + PicMaxUrl: vp.PicMaxUrl, + PicSmUrl: vp.PicSmUrl, + IsAlarm: isAlarmInt, + Sex: vp.Sex, + AgeDescription: vp.AgeDescription, + Race: vp.Race, + TaskName: vp.TaskName, + TaskId: vp.TaskId, + VideoUrl: vp.VideoUrl, + BaseInfo: bi, + SdkName: "浜鸿劯", + } + resultList[captureM[vp.Id].Index] = vpE + } + } + return resultList +} + +func parseMonitorLevel(level string) string { + if level == "1" { + return "涓�绾�" + } + if level == "2" { + return "浜岀骇" + } + if level == "3" { + return "涓夌骇" + } + return level +} + // @Summary 浠ュ浘鎼滃浘 // @Description 浠ュ浘鎼滃浘 // @Accept json // @Produce json // @Tags 浠ュ浘鎼滃浘 -// @Param condition body controllers.SearchCondition true "鎼滅储鍙傛暟" +// @Param condition body models.EsSearch true "鎼滅储鍙傛暟" // @Success 200 {string} json "{"code":200, msg:"", data:"", success:true}" // @Failure 500 {string} json "{"code":500, msg:"", data:"", success:false}" // @Router /data/api-v/dbperson/searchByPhoto [POST] func (controller FileController) SearchByPhoto(c *gin.Context) { - var condition SearchCondition - err := c.BindJSON(&condition) - if err !=nil || condition.PicUrl == "" { + var searchBody models.EsSearch + err := c.BindJSON(&searchBody) + if err !=nil{ util.ResponseFormat(c, code.RequestParamError, "鍙傛暟鏈夎") return } - compThreshold, err := strconv.ParseFloat(condition.Threshold,32) - if condition.PicUrl == "" || err != nil { - util.ResponseFormat(c, code.RequestParamError, "鍙傛暟鏈夎") - return - } - if face,ok := faceExtractedMap[condition.PicUrl];!ok{ - util.ResponseFormat(c, code.RequestParamError, "鍙傛暟鏈夎") - return - } else { - arg := protomsg.CompareArgs{ - TableIds:[]string{ captureTable }, - FaceFeature:face.FaceBytes, - CompareThreshold:float32(compThreshold), - } - if condition.Databases !=nil { - arg.TableIds = append(arg.TableIds, condition.Databases...) - } - b, err := proto.Marshal(&arg) - if err !=nil{ - util.ResponseFormat(c, code.ComError, "璇锋眰marshal澶辫触") + var faceB []byte + if searchBody.CaptureId != "" {//鍋氭煡鎵炬浜�,鎼滄墍鏈夋姄鎷嶅拰搴曞簱 + searchBase64Fea, err := esApi.GetVideoPersonFaceFeatureById(searchBody.CaptureId, config.EsInfo.EsIndex.VideoPersons.IndexName, config.EsInfo.Masterip, config.EsInfo.Httpport) + if err !=nil { + util.ResponseFormat(c, code.ComError, "鎶撴媿鏁版嵁涓嶅瓨鍦紝璇锋鏌�") return } - compServerList := config.CompServerInfo.Url - logger.Debug("compServerList:", compServerList) - resultList :=make([]CompareResult,0) - for _,str :=range compServerList{ - reqUrl := "tcp://"+str - reqClient := deliver.NewClient(deliver.ReqRep, reqUrl) - err = reqClient.Send(b) - if err !=nil{ - logger.Debug("reqClient.Send err:",err) - continue + if searchBase64Fea !=""{ + decodeF, err := base64.StdEncoding.DecodeString(searchBase64Fea) + if err !=nil { + util.ResponseFormat(c, code.ComError, "鏈潯鎶撴�曠壒寰佷笉鏄痓ase64锛岃妫�鏌�") + return } - resultB, err := reqClient.Recv() - if err !=nil{ - logger.Debug("reqClient.Recv err:",err) - continue - } - rList :=make([]protomsg.Esinfo,0) - err = json.Unmarshal(resultB, &rList) - if err !=nil{ - logger.Debug("recv result Unmarshal err:", err) - continue - } - logger.Debug("rList:", rList) - - dbPersonM := make(map[string]*protomsg.Esinfo,0) - captureM := make(map[string]*protomsg.Esinfo,0) - personIds :=make([]string,0) - captureIds := make([]string,0) - for _,v :=range rList{ - if v.Tableid == captureTable { - captureM[v.Id] = &v - personIds = append(personIds,v.Id) - } else { - dbPersonM[v.Id] = &v - captureIds = append(captureIds,v.Id) - } - } - logger.Debug("personIds:", personIds) - logger.Debug("captureIds:",captureIds) - esServerIp := config.EsInfo.Masterip - esServerPort := config.EsInfo.Httpport - index := config.EsInfo.EsIndex.Dbtablepersons.IndexName - dbpersons, e := esApi.Dbpersoninfosbyid(personIds, index, esServerIp, esServerPort) - if e !=nil { - logger.Debug("Dbpersoninfosbyid.err:",e) - } - logger.Debug("dbpersons:", dbpersons) - if e ==nil && dbpersons !=nil { - for _,p :=range dbpersons { - var dbP = DbPersonVo { - PersonId: p.Id, - IdCard: p.IdCard, - CompareScore: dbPersonM[p.Id].CompareScore, - MonitorLevel: p.MonitorLevel, - PersonName: p.PersonName, - PersonPicUrl: p.PersonPicUrl, - PhoneNum: p.PhoneNum, - Sex: p.Sex, - TableId: p.TableId, - } - dbTableInfos, _ := esApi.Dbtablefosbyid([]string{p.TableId}, config.EsInfo.EsIndex.DbTables.IndexName, esServerIp, esServerPort) - if dbTableInfos !=nil{ - dbP.BwType = dbTableInfos[0].BwType - dbP.TableName = dbTableInfos[0].TableName - } - var cr = CompareResult{ - BaseInfo:[]DbPersonVo{ dbP }, - } - resultList = append(resultList,cr) - } - } - logger.Debug("resultList:",resultList) + faceB = decodeF } - util.ResponseFormat(c,code.Success,resultList) + } else {//鍋氫互鍥炬悳鍥� + if searchBody.PicUrl == "" || len(searchBody.DataBases) == 0 { + util.ResponseFormat(c, code.RequestParamError, "鍙傛暟鏈夎") + return + } + if face,ok := faceExtractedMap[searchBody.PicUrl];!ok{ + util.ResponseFormat(c, code.RequestParamError, "璇烽噸鏂颁笂浼犲浘鐗�") + return + } else { + faceB = face.FaceBytes + } } + + var sysSetApi dbapi.SysSetApi + analyServerId := "" + flag, sysconf := sysSetApi.GetServerInfo() + if flag { + analyServerId = sysconf.ServerId + } else { + util.ResponseFormat(c, code.ComError, "analyServerId涓虹┖锛岄厤缃湁璇�") + return + } + + if faceB == nil { + util.ResponseFormat(c, code.RequestParamError, "璇烽噸鏂颁笂浼犲浘鐗�") + return + } + arg := protomsg.CompareArgs{ + FaceFeature: faceB, + CompareThreshold: searchBody.Threshold, + } + + var hasCompEsPerson = false + if searchBody.DataBases !=nil { + for idx,tableId :=range searchBody.DataBases { + if tableId == "esData" { + searchBody.DataBases = append(searchBody.DataBases[:idx], searchBody.DataBases[idx+1:]...) + hasCompEsPerson = true + break + } + } + arg.TableIds = searchBody.DataBases + } else { + arg.TableIds = []string{} + } + arg.Source = true // 鏍囪瘑鏉ユ簮鏄痺eb + arg.AlarmLevel = searchBody.AlarmLevel + arg.Tasks = searchBody.Tasks + arg.TreeNodes = searchBody.TreeNodes + arg.Tabs = searchBody.Tabs + arg.SearchTime = searchBody.SearchTime + arg.InputValue = searchBody.InputValue + arg.Collection = searchBody.Collection + arg.AnalyServerId = analyServerId + + logger.Debug("arg.TableIds:", arg.TableIds, ",alarmLevel:",arg.AlarmLevel,",treeNodes:",arg.TreeNodes,",searchTime:",arg.SearchTime, + ",inputValue:",arg.InputValue,",tasks:",arg.Tasks,",compThreshold:",arg.CompareThreshold) + timeStart := time.Now() + + compareService := service.NewFaceCompareService(arg) + var totalData service.CompareList + if len(arg.TableIds) >0 {//鏈夋瘮瀵瑰簳搴� + dbPersonTargets := compareService.CompareDbPersons() + if dbPersonTargets !=nil { + totalData = append(totalData,*dbPersonTargets...) + } + } + if hasCompEsPerson {//鏈夋瘮瀵笶s鎶撴媿 + esPersons := compareService.CompareVideoPersons() + if esPersons !=nil { + totalData = append(totalData, *esPersons...) + } + } + + logger.Debug("comp 姣斿缁撴潫锛岀敤鏃讹細",time.Since(timeStart)) + service.SetCompResultByNum(&service.CompareOnce{ + CompareNum: compareService.CompareNum, + CompareData: &totalData, + }) + + m := make(map[string]interface{},3) + if totalData != nil && totalData.Len() > 0{ + sort.Sort(totalData) + total := totalData.Len() + + m["compareNum"] = compareService.CompareNum + m["total"] = total + var sCompResult protomsg.SdkCompareResult + if total <= searchBody.Size { + sCompResult.CompareResult = totalData + } else { + sCompResult.CompareResult = totalData[0:searchBody.Size] + } + resultList := FillDataToCompareResult(&sCompResult) + m["totalList"] = resultList + logger.Debug("comp 姣斿鍔犳帓搴忚繑鍥炵敤鏃�:", time.Since(timeStart)) + + } else { + m["total"] = 0 + m["compareNum"] = compareService.CompareNum + m["totalList"] = []CompareResult{} + } + util.ResponseFormat(c,code.Success,m) } + +func GetCompareDataTwice(co *service.CompareOnce,arg *models.EsSearch) map[string]interface{} { + //camIds := arg.TreeNodes + //tabIds := arg.Tabs + //taskIds := arg.Tasks + //alarmLevels := arg.AlarmLevel + //timeArr := arg.SearchTime + //input := arg.InputValue + //resultList := make([]CompareResult,0) + //for _,each :=range *co.CompareData { + // if each.Id !=""{ + // n :=0 + // if len(camIds) == 0 || isInArr(each.CameraId,camIds) { + // n++ + // } + // if len(tabIds) == 0 { + // n++ + // } + // if len(taskIds) == 0 || isInArr(each.TaskId, taskIds) { + // n++ + // } + // //鍒ゆ柇鎶ヨ绛夌骇 + // if len(alarmLevels) == 0 { + // + // } + // n++ + // //鍒ゆ柇鏃堕棿 + // if len(timeArr) == 0 { + // + // } + // n++ + // if input == ""{ + // + // } + // n++ + // if n == 6 { + // resultList = append(resultList,cr) + // } + // } else { + // resultList = append(resultList,cr) + // } + //} + + //logger.Debug("GetCompareDataTwice.data:",resultList) + //service.SortByScore(resultList) + from := (arg.Page-1)*arg.Size + to := from + arg.Size + var sCompResult protomsg.SdkCompareResult + total := len(*co.CompareData) + if total <= to { + sCompResult.CompareResult = (*co.CompareData)[from:total] + } else { + sCompResult.CompareResult = (*co.CompareData)[from:to] + } + resultList := FillDataToCompareResult(&sCompResult) + + m := make(map[string]interface{},0) + m["total"] = total + m["compareNum"] = arg.CompareNum + m["totalList"] = resultList + //if from < len(resultList) { + // if to <= len(resultList) { + // m["totalList"] = resultList[from:to] + // } else { + // lte := len(resultList) + // m["totalList"] = resultList[from:lte] + // } + //} else { + // m["totalList"] = []CompareResult{} + //} + return m +} + +func isInArr(id string,arr []string) bool { + for _,v :=range arr { + if v == id { + return true + } + } + return false +} + // @Description 浜哄憳鐓х墖涓婁紶骞惰幏鍙栫壒寰佸�� // @Router /data/api-v/dbperson/fileUploadTest [POST] @@ -343,7 +565,6 @@ func uploadFileReturnAddr(file multipart.File, filename string, tableId string) (string, map[string]interface{}, error) { defer file.Close() - field := "" // weedfs 涓婁紶 fileBytes, err := ioutil.ReadAll(file) if err !=nil { @@ -358,6 +579,9 @@ } var rcFace *protomsg.Rect if err ==nil && len(faceArr) >0 { + if len(faceArr) >1 { + return "",nil, errors.New("TooManyFaces") + } for _,r := range faceArr { //鎷垮埌浜鸿劯鐨勫潗鏍� rcFace = r.Pos.RcFace @@ -368,55 +592,33 @@ } var weedfsUri = "http://"+config.WeedFs.Ip+":"+strconv.Itoa(config.WeedFs.UploadPort)+"/submit" //鏍规嵁浜鸿劯鍧愭爣鎵e嚭浜鸿劯灏忓浘 + t1 := time.Now() cutFaceImgData := util.SubImg(*pI, int(rcFace.Left), int(rcFace.Top), int(rcFace.Right), int(rcFace.Bottom)) - fileInfo, e := esutil.PostFormData(weedfsUri, filename, "file", cutFaceImgData) + logger.Debug("SubImg鐢ㄦ椂锛�", time.Since(t1)) + t1 = time.Now() + weedFilePath, e := WeedFSClient.UploadFile(weedfsUri, filename, cutFaceImgData) + logger.Debug("涓婁紶鍒皐eedfs鐢ㄦ椂:", time.Since(t1)) + t1 = time.Now() if e != nil { fmt.Println(e.Error()) return "", nil, e - } else { - field = fileInfo[picUrlField].(string) // 鏂囦欢璺緞 } - if strings.Contains(field,"/"){ - idx := strings.Index(field, "/") - field = field[idx+1:] - } - fmt.Println("field:",field) + ext := path.Ext(filename) fileNameOnly := strings.TrimSuffix(filename, ext) dbperson := new(models.Dbtablepersons) - dbperson.PersonPicUrl = field // 鍥剧墖璺粡 + dbperson.PersonPicUrl = weedFilePath // 鍥剧墖璺粡 dbperson.TableId = tableId // dbperson.PersonName = fileNameOnly // 鍥剧墖鍚� - // 婕旂ずbase64缂栫爜 - + dbperson.Enable = 1 //榛樿鏈夋晥 dbperson.FaceFeature = faceBase64 // 鐗瑰緛鍊糱ase64 鐮� result := addDbPerson(dbperson) + logger.Debug("addDbPerson鐢ㄦ椂锛�", time.Since(t1)) return fileNameOnly, result, nil } - -//func getFaceFeaFromSdk(fileBytes []byte) ([]*protomsg.ResultFaceExtCom,error,*protomsg.Image){ -// s := service.NewFaceSdkService(fileBytes) -// i, err := s.ReadFromUploadImg() -// if err !=nil{ -// fmt.Println("readFromUploadImg err:",err) -// return nil,err,i -// } -// bc, err := service.ImgCompress(i) -// if err !=nil { -// fmt.Println("ImgCompress err:",err) -// return nil,err,i -// } -// s.PushImgMsg(bc) -// s.GetFaceFea() -// if s.Result == nil{ -// return nil,errors.New("no fea"),i -// } else { -// return s.Result,nil,i -// } -//} /**涓婁紶鏂规硶**/ @@ -444,33 +646,44 @@ addResult := make(map[string]interface{}, 0) successList := make([]string, 0) failList := make([]string,0) + tAllStart := time.Now() + var wg sync.WaitGroup + var lock sync.Mutex for _, head := range fileHeaders { - //鏂囦欢鐨勫悕绉� - filename := head.Filename - file, err := head.Open() - fmt.Println(file, err, filename) - if err != nil { - log.Fatal(err) - } - - field, _, err1 := uploadFileReturnAddr(file, filename, tableId) - //extNames = append(extNames, field) - - if err1 != nil || field == "" { - failList = append(failList, filename) - } else { - successList = append(successList, filename) - } + wg.Add(1) + go func(head *multipart.FileHeader,tableId string) { + defer wg.Done() + tIStart := time.Now() + filename := head.Filename + file, err := head.Open() + fmt.Println(file, err, filename) + if err != nil { + lock.Lock() + failList = append(failList, filename) + lock.Unlock() + return + } + field, _, err1 := uploadFileReturnAddr(file, filename, tableId) + lock.Lock() + if err1 != nil || field == "" { + failList = append(failList, filename) + } else { + successList = append(successList, filename) + } + lock.Unlock() + logger.Debug(filename,"鍒囧浘鐢ㄦ椂锛�",time.Since(tIStart)) + }(head, tableId) + wg.Wait() } + logger.Debug("鍒�",len(fileHeaders),"寮犱汉鑴哥敤鏃讹細", time.Since(tAllStart)) addResult["successList"] = successList addResult["failList"] = failList addResult["fields"] = extNames - //addResult["detail"] = details - //fields := fdfsclient.UploadFileByBuffer100(bytess, extNames) // fastdfs 涓婁紶 + if len(successList)>0 { - util.ResponseFormat(c, code.Success, addResult) + util.ResponseFormat(c, code.DbPersonUploadSuccess, addResult) } else { - util.ResponseFormat(c, code.ServiceInsideError, err.Error()) + util.ResponseFormat(c, code.DbPersonUploadFail, addResult) } } -- Gitblit v1.8.0