package esutil import ( "bytes" "encoding/json" "errors" "fmt" "io" "io/ioutil" "net/http" "strconv" "strings" "sync" "time" "andriodServer/extend/config" log "andriodServer/log" ) func GetEsDataReq(url string, parama string, picurl string, isSource bool) (error, map[string]interface{}) { //log.Log.Infoln("es 查询请求路径" + url) // 配置信息 获取 req, err := http.NewRequest("POST", url, strings.NewReader(parama)) if err != nil { return err, nil } req.Header.Add("Content-Type", "application/json") timeout := time.Duration(10 * time.Second) //超时时间50ms client := &http.Client{Timeout: timeout} resp, err := client.Do(req) if err != nil { return err, nil } defer resp.Body.Close() body, err := ioutil.ReadAll(resp.Body) if err != nil { return err, nil } jsonStr := string(body) var dat map[string]interface{} dec := json.NewDecoder(strings.NewReader(jsonStr)) if err := dec.Decode(&dat); err == io.EOF { return err, nil } else if err != nil { return err, nil } // 是否需要 解析 es 返回的 source if isSource { dat, ok := dat["hits"].(map[string]interface{}) if !ok { return errors.New("data is not type of map[string]interface{}"), nil } var data = make(map[string]interface{}, 2) data["total"] = dat["total"] sources := []interface{}{} for _, value := range dat["hits"].([]interface{}) { source, ok := value.(map[string]interface{})["_source"].(map[string]interface{}) if !ok { return errors.New("value is not type of map[string]interface{}"), nil } source["id"] = value.(map[string]interface{})["_id"] sdkType := source["sdkType"] if sdkType != nil { sdk, err := strconv.Atoi(sdkType.(string)) if err != nil { return err, nil } source["sdkType"] = sdkTypeToValue(sdk) } pmax, exist := source["picMaxUrl"].(string) //fmt.Println("picMaxUrl: ",pmax) if !exist { return errors.New("picMaxurl is not string"), nil } if !strings.HasPrefix(pmax, "http") { source["picMaxUrl"] = picurl + pmax } psm, exist := source["picSmUrl"].(string) if !exist { return errors.New("picSmUrl is not string"), nil } if !strings.HasPrefix(psm, "http") { source["picSmUrl"] = picurl + psm } if source["sdkType"] != "人脸" {//行为没有存储大图 source["picMaxUrl"] = source["picSmUrl"] } prace, exist := source["Race"] if exist { source["race"] = prace } pGender, exist := source["Gender"] if exist { source["gender"] = pGender } source["ageDescription"] = getAgeDesc(source["Age"]) source["videoNum"] = getVideoUrl(source) picDate := source["picDate"].(string) lastIdx := strings.LastIndex(picDate,":") picDateStr := picDate[:lastIdx] if err == nil { source["picDate"] = picDateStr } baseInfo := getSourceBaseInfo(source) source["baseInfo"] = baseInfo sources = append(sources, source) } data["datalist"] = sources return nil, data } else { return nil, dat } } func getAgeDesc(age interface{})(ageDesc string) { if age !=nil { ageInt := age.(float64) if ageInt >0 && ageInt<7 { ageDesc = "童年" } else if ageInt >=7 && ageInt<18 { ageDesc = "少年" } else if ageInt >=18 && ageInt<40 { ageDesc = "青年" } else if ageInt >=40 && ageInt<65 { ageDesc = "中年" } else if ageInt >=65 { ageDesc = "老年" } else { ageDesc = "" } } return ageDesc } type BaseInfo struct { TaskId string `json:"taskId"` TaskName string `json:"taskName"` LikePer string `json:"likePer"` TableId string `json:"tableId"` TableName string `json:"tableName"` PersonId string `json:"personId"` PersonPicUrl string `json:"personPicUrl"` PersonName string `json:"personName"` Gender string `json:"gender"` PhoneNum string `json:"phoneNum"` IDCard string `json:"IDCard"` MonitorLevel string `json:"monitorLevel"` Content string `json:"content"` } func getSourceBaseInfo(source map[string]interface{}) []BaseInfo { sdkType := source["sdkType"].(string) baseInfoArr := make([]BaseInfo,0) if sdkType == "人脸" { likePer,baseName,personId,idCard,personPicUrl,gender,content :="","","","","","","" if source["likePer"] !=nil { likePer = source["likePer"].(string) } if source["BaseName"] !=nil { baseName = source["BaseName"].(string) } if source["personId"] !=nil { personId = source["personId"].(string) } if source["idcard"] !=nil { idCard = source["idcard"].(string) } if source["personPicUrl"] !=nil { personPicUrl = source["personPicUrl"].(string) } if source["Gender"] !=nil { gender = source["Gender"].(string) } if source["content"] !=nil { content = source["content"].(string) } var baseInfo = BaseInfo{ TaskId:"",//2.0新字段 TaskName:"",//2.0新字段 LikePer:likePer, TableId:"",//2.0新字段 TableName:baseName, PersonId:personId, PersonName:idCard,//人员姓名,从管理平台获取 PersonPicUrl:personPicUrl, Gender:gender, PhoneNum:"",//手机号,从管理平台获取 IDCard:idCard, MonitorLevel:"",//2.0新字段 Content:content, } baseInfoArr = append(baseInfoArr, baseInfo) //bytes, err := json.Marshal(baseInfoArr) //if err !=nil { // return "" //} } return baseInfoArr } var videoCacheMap = make(map[string]string,0) var lock sync.RWMutex func setVideoCache(imgKey string, url string) { lock.Lock() defer lock.Unlock() videoCacheMap[imgKey] = url } func getVideoFromCache(imgKey string) string { lock.Lock() defer lock.Unlock() if v,ok := videoCacheMap[imgKey];ok { return v } return "" } func getVideoUrl(source map[string]interface{}) (videoUrl string){ imgKey := source["imgKey"].(string) //先从缓存里面取 cacheUrl := getVideoFromCache(imgKey) if cacheUrl !="" { return cacheUrl } picDate := source["picDate"].(string)//抓拍日期 cameraId := source["videoReqNum"].(string)//摄像机id indeviceId := source["indeviceid"].(string)//分析设备id reqUrl := "" if url,ok := config.ServerMap[indeviceId];!ok { return "" } else { reqUrl = url } log.Log.Infoln("reqUrl:",reqUrl) paramMap := make(map[string]interface{},0) paramMap["imgKey"] = imgKey paramMap["picDate"] = picDate paramMap["videoNum"] = cameraId respBytes, err := doPostRequest(reqUrl, "application/json", paramMap, nil, nil) if err !=nil{ return "" } var resp RespVideo err = json.Unmarshal(respBytes, &resp) if err !=nil { return "" } filePath := resp.FilePath videoUrl = "" if !strings.Contains(filePath, "/cut"){ videoUrl = "" fmt.Printf("videoReqUrl:%s ,imgKey:%s ,picDate:%s ,cameraId:%s ,filePath:%s \n ",reqUrl,imgKey,picDate,cameraId,filePath) } else { strArr := strings.Split(filePath, "/cut") ngxUrl := config.NgxMap[indeviceId] log.Log.Infoln("ngxUrl:",ngxUrl) if ngxUrl !="" && len(strArr) >0 { videoUrl = ngxUrl + strArr[1] } } if videoUrl == ""{ respMap := make(map[string]interface{},0) err := json.Unmarshal(respBytes, &respMap) if err !=nil { fmt.Println("resp UnmarshalToMap err: ",err) } fmt.Println("resp: ",respMap) } setVideoCache(imgKey, videoUrl) return videoUrl } type RespVideo struct{ FilePath string `json:"file_path"` } func doPostRequest(url string, contentType string, body map[string]interface{}, params map[string]string, headers map[string]string) ([]byte, error) { var resultBytes []byte var bodyJson []byte if body != nil { var err error bodyJson, err = json.Marshal(body) if err != nil { log.Log.Errorln("doPostRequestMarshal err:",err) return resultBytes, err } } request, err := http.NewRequest("POST", url, bytes.NewBuffer(bodyJson)) if err != nil { log.Log.Errorln("NewRequest ERR:",err) return resultBytes, err } request.Header.Set("Content-type", contentType) //add params q := request.URL.Query() if params != nil { for key, val := range params { q.Add(key, val) } request.URL.RawQuery = q.Encode() } // add headers if headers != nil { for key, val := range headers { request.Header.Add(key, val) } } timeOut:= time.Duration(8*time.Second)//set request timeout client := &http.Client{ Timeout:timeOut, } resp, err := client.Do(request) if err != nil { log.Log.Errorln("DoRequest ERR:",err) return resultBytes, err } defer resp.Body.Close() resultBytes, err = ioutil.ReadAll(resp.Body) if err != nil { log.Log.Errorln("ReadAll ERR:",err) return resultBytes, err } return resultBytes, nil } //sdk类型 func sdkTypeToValue(i int) string { value := []string{"人脸", "车辆", "人体", "入侵", "拥挤", "靠右行", "人员异常", "个体静止"} return value[i-1] } func PostAction(sec int, Eurl string, picurl string, ishub string, size int, lastT time.Time, curTime time.Time) []byte { //lastTimeStr := lastT.Format("2006-01-02 15:04:05") //curTimeStr := curTime.Format("2006-01-02 15:04:05") index := "videopersons,personaction" url := fmt.Sprintf("%s%s%s", Eurl, index, "/_search") startTime := time.Now() sizeStr :="" if size <=0 { sizeStr = "1000" } else { sizeStr = strconv.Itoa(size) } preSec := "10" if sec <=0 { preSec = "10" } var filterArr []string if ishub == "hub" { filterArr = append(filterArr,"{\"term\":{\"personIsHub\":\"1\"}}") } filterArr = append(filterArr, "{\"range\":{\"picDate\":{\"gte\":\"now+8h-"+preSec+"s\",\"lt\":\"now+8h\"}}}") filterStr := "" if len(filterArr) >0 { filterStr = strings.Join(filterArr, ",") } log.Log.Infoln("filterArr:", filterStr) prama := "{\"query\":{\"bool\":{\"filter\":["+filterStr+"]}},\"size\":\""+sizeStr+"\",\"sort\":[{\"picDate\":{\"order\":\"desc\"}}]," + "\"_source\":[\"baseInfo\",\"Gender\",\"BaseName\",\"Age\",\"personId\",\"personPicUrl\",\"indeviceName\",\"imgKey\",\"sdkType\",\"ageDescription\",\"indeviceid\",\"content\",\"Id\",\"picAddress\",\"picMaxUrl\",\"picDate\",\"Race\",\"videoNum\",\"picSmUrl\",\"taskName\",\"personIsHub\",\"idcard\",\"videoIp\",\"videoReqNum\"]" + "}" err, tokenRes := GetEsDataReq(url, prama, picurl, true) log.Log.Infoln("条数:",size,"耗时:",time.Since(startTime)) if err != nil { log.Log.Errorln(err) return nil } jsonstring, _ := json.Marshal(tokenRes) if len(jsonstring) <= 26 { return nil } return jsonstring }