From e189b1942c8130d473a1a4128c3d9ef5edfaa564 Mon Sep 17 00:00:00 2001 From: sunty <1172534965@qq.com> Date: 星期五, 23 十月 2020 10:58:30 +0800 Subject: [PATCH] fix --- EsApi.go | 652 ++++++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 files changed, 596 insertions(+), 56 deletions(-) diff --git a/EsApi.go b/EsApi.go index b812bf5..96348c7 100644 --- a/EsApi.go +++ b/EsApi.go @@ -1,15 +1,15 @@ package esutil import ( + "basic.com/pubsub/protomsg.git" "encoding/json" "errors" "fmt" + "sort" "strconv" "strings" "sync" "time" - - "basic.com/pubsub/protomsg.git" ) var logPrint = func(i ...interface{}) { @@ -159,11 +159,305 @@ } -//鏍规嵁鎽勫儚鏈哄垪琛ㄥ拰鏃堕棿鏌ヨ浜哄憳娴忚杞ㄨ抗 -func GetPersonDataByCameraIdAndTime(cameraId []string, startTime string, endTime string, serverIp string, ServerPort string, indexName string) ([]map[string]interface{}, error) { - esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1) - personUrl := "http://" + serverIp + ":" + ServerPort + "/" + indexName + "/_search" - personBody := `{ +/**************************************customer analysis util start**************************************/ +/*******************sort []map util*******************/ +type MapsSort struct { + Key string + MapList []map[string]interface{} +} + +func (m *MapsSort) Len() int { + return len(m.MapList) +} + +func (m *MapsSort) Less(i, j int) bool { + return m.MapList[i][m.Key].(string) > m.MapList[j][m.Key].(string) +} + +func (m *MapsSort) Swap(i, j int) { + m.MapList[i], m.MapList[j] = m.MapList[j], m.MapList[i] +} + +/*******************sort []map util*******************/ +//鏍规嵁鏃堕棿鑼冨洿鑱氬悎鎵�鏈夊尯鍩熶汉淇℃伅锛岃繑鍥炲浐瀹氭潯鏁� +func GetFaceDataByTimeAndTotal(startTime string, endTime string, total int, thresholdTime int, thresholdStayTime int, serverIp string, serverPort string, indexName string) (resData []map[string]interface{}, err error) { + var requestUrl = "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" + var requestBody = `{ + "query": { + "bool": { + "filter": [ + { + "range": { + "picDate": { + "gte": "` + startTime + `", + "lte": "` + endTime + `" + } + } + }, + { + "term":{ + "targetInfo.targetType.raw": "FaceDetect" + } + } + ] + } + }, + "size": 0, + "aggs": { + "buckets_aggs": { + "composite": { + "sources": [ + { + "faceId": { + "terms": { + "field": "baseInfo.targetId" + } + } + }, + { + "areaId": { + "terms": { + "field": "targetInfo.areaId" + } + } + } + ], + "size": 10000000 + }, + "aggs": { + "top_attention_hits": { + "top_hits": { + "size": 1000000, + "sort": [ + { + "picDate": { + "order": "asc" + } + } + ], + "_source": { + "includes": [ + "baseInfo.targetId", + "targetInfo.picSmUrl", + "targetInfo.areaId", + "picDate" + ] + } + } + } + } + } + } +}` + buf, err := EsReq("POST", requestUrl, []byte(requestBody)) + if err != nil { + return nil, err + } + source, err := FaceSourceAggregations(buf, thresholdTime, thresholdStayTime) + if err != nil { + return nil, err + } + if len(source) == 0 { + return source, nil + } + faceSource := make([]map[string]interface{}, 0) + for index, info := range source { + if int(info["stayTime"].(float64)) > thresholdStayTime { + faceSource = append(faceSource, source[index]) + } + } + mapsSort := MapsSort{} + mapsSort.Key = "endTime" + mapsSort.MapList = faceSource + sort.Sort(&mapsSort) + if len(faceSource) > total { + return mapsSort.MapList[:total], nil + } + return mapsSort.MapList, nil +} + +func GetFaceDataByTimeAndId(startTime string, endTime string, id string, thresholdTime int, thresholdStayTime int, serverIp string, serverPort string, indexName string) (resData []map[string]interface{}, err error) { + var requestUrl = "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" + var requestBody = `{ + "query": { + "bool": { + "filter": [ + { + "range": { + "picDate": { + "gte": "` + startTime + `", + "lte": "` + endTime + `" + } + } + }, + { + "term":{ + "targetInfo.targetType.raw": "FaceDetect" + } + }, + { + "term":{ + "baseInfo.targetId": "` + id + `" + } + } + ] + } + }, + "size": 0, + "aggs": { + "buckets_aggs": { + "composite": { + "sources": [ + { + "faceId": { + "terms": { + "field": "baseInfo.targetId" + } + } + }, + { + "areaId": { + "terms": { + "field": "targetInfo.areaId" + } + } + } + ], + "size": 10000000 + }, + "aggs": { + "top_attention_hits": { + "top_hits": { + "size": 1000000, + "sort": [ + { + "picDate": { + "order": "asc" + } + } + ], + "_source": { + "includes": [ + "baseInfo.targetId", + "targetInfo.picSmUrl", + "targetInfo.areaId", + "picDate" + ] + } + } + } + } + } + } +}` + buf, err := EsReq("POST", requestUrl, []byte(requestBody)) + if err != nil { + return nil, err + } + source, err := FaceSourceAggregations(buf, thresholdTime, thresholdStayTime) + if err != nil { + return nil, err + } + if len(source) == 0 { + return source, nil + } + faceSource := make([]map[string]interface{}, 0) + for index, info := range source { + if int(info["stayTime"].(float64)) > thresholdStayTime { + faceSource = append(faceSource, source[index]) + } + } + mapsSort := MapsSort{} + mapsSort.Key = "startTime" + mapsSort.MapList = faceSource + sort.Sort(&mapsSort) + return mapsSort.MapList, nil +} + +func GetFaceIdDeduplication(startTime string, endTime string, serverIp string, serverPort string, indexName string) (ids []map[string]interface{}, err error) { + var requestUrl = "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" + var requestBody = `{ + "query": { + "bool": { + "filter": [ + { + "range": { + "picDate": { + "gte": "` + startTime + `", + "lte": "` + endTime + `" + } + } + }, + { + "term": { + "targetInfo.targetType.raw": "FaceDetect" + } + } + ] + } + }, + "size": 0, + "aggs": { + "buckets_aggs": { + "composite": { + "sources": [ + { + "faceId": { + "terms": { + "field": "baseInfo.targetId" + } + } + } + ], + "size": 10000000 + }, + "aggs": { + "top_attention_hits": { + "top_hits": { + "size": 1, + "sort": [ + { + "picDate": { + "order": "desc" + } + } + ], + "_source": { + "includes": [ + "picDate" + ] + } + } + } + } + } + } + } +}` + //fmt.Println(requestUrl) + //fmt.Println(requestBody) + buf, err := EsReq("POST", requestUrl, []byte(requestBody)) + if err != nil { + return nil, err + } + ids, err1 := SourceDeduplication(buf) + if err1 != nil { + return nil, err1 + } + if len(ids) > 1 { + mapsSort := MapsSort{} + mapsSort.Key = "lastTime" + mapsSort.MapList = ids + sort.Sort(&mapsSort) + return mapsSort.MapList, nil + } + return ids, nil +} + +//缁熻鍚勪釜鍖哄煙浜烘暟 +func StatisticsEveryAreaPersonsNumber(startTime string, endTime string, serverIp string, serverPort string, indexName string) ([]map[string]interface{}, error) { + var requestUrl = "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" + var requestBody = `{ "query": { "bool": { "filter": [ @@ -179,14 +473,73 @@ "term": { "targetInfo.targetType.raw": "Yolo" } - }, - { - "terms": { - "cameraId": [ - "` + esCameraId + `" - ] - } } + ] + } + }, + "size": 0, + "aggs": { + "buckets_aggs": { + "composite": { + "sources": [ + { + "areaId": { + "terms": { + "field": "targetInfo.areaId" + } + } + } + ], + "size": 10000000 + } + } + } +}` + buf, err := EsReq("POST", requestUrl, []byte(requestBody)) + if err != nil { + return nil, err + } + result, err := SourceStatistics(buf) + if err != nil { + return nil, err + } + return result, nil +} + +/**************************************customer analysis util end**************************************/ +//鏍规嵁鎽勫儚鏈哄垪琛ㄥ拰鏃堕棿鏌ヨ浜哄憳娴忚杞ㄨ抗 +func GetPersonDataByCameraIdAndTime(cameraId []string, startTime string, endTime string, serverIp string, serverPort string, indexName string) (map[string]interface{}, error) { + + var filterArr []string + if cameraId != nil && len(cameraId) > 0 { + esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1) + filterArr = append(filterArr, `{ + "terms": { + "cameraId": ["`+esCameraId+`"] + } + }`) + } + filterArr = append(filterArr, `{ + "range": { + "picDate": { + "gte": "`+startTime+`", + "lte": "`+endTime+`" + } + } + }`) + filterArr = append(filterArr, ` { + "term": { + "targetInfo.targetType.raw": "Yolo" + } + }`) + queryStr := strings.Join(filterArr, ",") + + personUrl := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" + personBody := `{ + "query": { + "bool": { + "filter": [ + ` + queryStr + ` ] } }, @@ -208,47 +561,66 @@ }` //fmt.Println(personUrl) //fmt.Println(personBody) + source := make(map[string]interface{}) + queryStartTime := time.Now() buf, err := EsReq("POST", personUrl, []byte(personBody)) if err != nil { return nil, err } - + queryUseTime := time.Now().Sub(queryStartTime).Seconds() * 1000 sources, err := Sourcelist(buf) if err != nil { return nil, err } - resData,err := PerSonAnalysis(sources) + resData, err := PerSonAnalysis(sources) + source["result"] = resData + source["total"] = len(resData) + source["queryUseTime"] = queryUseTime //println(sources) - return resData, nil + return source, nil } -//鏍规嵁鏃堕棿鑼冨洿锛屾憚鍍忔満鍒楄〃锛屽垎缁勮仛鍚堜汉鑴稿垪琛� -func GetFaceDataBucketsByCameraIdAndTime(cameraId []string, startTime string, endTime string, thresholdTime float64, serverIp string, ServerPort string, indexName string) (buckersDate map[string]interface{}, err error) { - esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1) +//鏍规嵁鏃堕棿鑼冨洿锛屾憚鍍忔満鍒楄〃锛屽垎缁勮仛鍚堜汉鑴稿垪琛�,杩斿洖鍒嗙粍鏁版嵁 +func GetFaceDataBucketsByCameraIdAndTimeReturnByGrouped(cameraId []string, personId []string, startTime string, endTime string, thresholdTime float64, serverIp string, ServerPort string, indexName string) (buckersDate map[string]interface{}, err error) { + var filterArr []string + if cameraId != nil && len(cameraId) > 0 { + esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1) + filterArr = append(filterArr, `{ + "terms": { + "cameraId": ["`+esCameraId+`"] + } + }`) + } + if personId != nil && len(personId) > 0 { + esPersonId := strings.Replace(strings.Trim(fmt.Sprint(personId), "[]"), " ", "\",\"", -1) + filterArr = append(filterArr, `{ + "terms": { + "baseInfo.targetId": ["`+esPersonId+`"] + } + }`) + } + filterArr = append(filterArr, `{ + "range": { + "picDate": { + "gte": "`+startTime+`", + "lte": "`+endTime+`" + } + } + }`) + filterArr = append(filterArr, ` { + "term": { + "targetInfo.targetType.raw": "FaceDetect" + } + }`) + queryStr := strings.Join(filterArr, ",") + var buckersUrl = "http://" + serverIp + ":" + ServerPort + "/" + indexName + "/_search" var buckersBody = `{ "query": { "bool": { "filter": [ - { - "range": { - "picDate": { - "gte": "` + startTime + `", - "lte": "` + endTime + `" - } - } - }, - { - "term": { - "targetInfo.targetType.raw": "FaceDetect" - } - }, - { - "terms": { - "cameraId": ["` + esCameraId + `"] - } - } + ` + queryStr + ` ] } }, @@ -272,7 +644,7 @@ } } ], - "size": 100000000 + "size": 10000000 }, "aggs":{ "top_attention_hits":{ @@ -296,33 +668,139 @@ }` //fmt.Println(buckersUrl) //fmt.Println(buckersBody) + sources := make(map[string]interface{}) + queryStartTime := time.Now() buf, err := EsReq("POST", buckersUrl, []byte(buckersBody)) if err != nil { return nil, err } - - sources, err := SourceAggregations(buf, thresholdTime) + queryUseTime := time.Now().Sub(queryStartTime).Seconds() * 1000 + //fmt.Println(queryUseTime) + tmpSources, err := SourceAggregationsReturnByGrouped(buf, thresholdTime) if err != nil { return nil, err } + sources["result"] = tmpSources + sources["total"] = len(tmpSources) + sources["queryUseTime"] = queryUseTime //println(sources) + return sources, nil +} + +//鏍规嵁鏃堕棿鑼冨洿锛屾憚鍍忔満鍒楄〃锛屽垎缁勮仛鍚堜汉鑴稿垪琛� +func GetFaceDataBucketsByCameraIdAndTime(cameraId []string, personId []string, startTime string, endTime string, thresholdTime float64, serverIp string, ServerPort string, indexName string) (buckersDate map[string]interface{}, err error) { + var filterArr []string + if cameraId != nil && len(cameraId) > 0 { + esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1) + filterArr = append(filterArr, `{ + "terms": { + "cameraId": ["`+esCameraId+`"] + } + }`) + } + if personId != nil && len(personId) > 0 { + esPersonId := strings.Replace(strings.Trim(fmt.Sprint(personId), "[]"), " ", "\",\"", -1) + filterArr = append(filterArr, `{ + "terms": { + "baseInfo.targetId": ["`+esPersonId+`"] + } + }`) + } + filterArr = append(filterArr, `{ + "range": { + "picDate": { + "gte": "`+startTime+`", + "lte": "`+endTime+`" + } + } + }`) + filterArr = append(filterArr, ` { + "term": { + "targetInfo.targetType.raw": "FaceDetect" + } + }`) + queryStr := strings.Join(filterArr, ",") + + var buckersUrl = "http://" + serverIp + ":" + ServerPort + "/" + indexName + "/_search" + var buckersBody = `{ + "query": { + "bool": { + "filter": [ + ` + queryStr + ` + ] + } + }, + "size": 0, + "aggs": { + "buckets_aggs": { + "composite": { + "sources": [ + { + "baseInfo.targetId": { + "terms": { + "field": "baseInfo.targetId" + } + } + }, + { + "cameraId": { + "terms": { + "field": "cameraId" + } + } + } + ], + "size": 10000000 + }, + "aggs":{ + "top_attention_hits":{ + "top_hits":{ + "size": 1000000, + "sort": [ + { + "picDate": { + "order": "asc" + } + } + ], + "_source":{ + "includes":["baseInfo.targetId","cameraId","cameraName","cameraAddr","targetInfo.targetScore","targetInfo.picSmUrl","showLabels","baseInfo.tableId","baseInfo.tableName","baseInfo.bwType","baseInfo.targetName","baseInfo.compareScore","picDate","picMaxUrl","picWH"] + } + } + } + } + } + } +}` + //fmt.Println(buckersUrl) + //fmt.Println(buckersBody) + queryStartTime := time.Now() + buf, err := EsReq("POST", buckersUrl, []byte(buckersBody)) + if err != nil { + return nil, err + } + queryUseTime := time.Now().Sub(queryStartTime).Seconds() * 1000 + + sources, err := SourceAggregations(buf, thresholdTime, queryUseTime) + if err != nil { + return nil, err + } return sources, nil } //鏍规嵁鎶撴媿浜哄憳id鏇存柊锛坧icurl锛夊浘鐗囧湴鍧� func UpdatePicUrlById(id string, picUrl string, indexName string, serverIp string, serverPort string) (err error) { - updateTime := time.Now().Format("2006-01-02 15:04:05") + updateTime := time.Now().Format("2006-01-02 15:04:05") tRes, err := AIOceaninfosbyid([]string{id}, indexName, serverIp, serverPort) if err != nil || len(tRes) == 0 { return err } picMaxUrls := tRes[0].PicMaxUrl - sourceStr := ` - "lang":"painless", - "inline": "ctx._source.picMaxUrl.add('` + picUrl + `');ctx._source.updateTime='`+updateTime+`'" + sourceStr := ` + "source": "ctx._source.picMaxUrl.add('` + picUrl + `');ctx._source.updateTime='` + updateTime + `'" ` if len(picMaxUrls) >= 2 { - sourceStr = `"source": "ctx._source.picMaxUrl[1]='` + picUrl + `';ctx._source.updateTime='`+updateTime+`'"` + sourceStr = `"source": "ctx._source.picMaxUrl[1]='` + picUrl + `';ctx._source.updateTime='` + updateTime + `'"` } var info interface{} url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query?refresh=true" @@ -346,7 +824,7 @@ } ` //logPrint("url: ", url, videoUrlInfo) - fmt.Println(url, picUrlInfo) + //fmt.Println(url, picUrlInfo) buf, err := EsReq("POST", url, []byte(picUrlInfo)) if err != nil { logPrint("http request videoUrlInfo info is err!") @@ -361,7 +839,7 @@ } middle, ok := out["updated"].(float64) if !ok { - logPrint("first updated change error!") + logPrint("first updated change error!", out) return errors.New("first updated change error!") } if middle == 1 { @@ -607,7 +1085,7 @@ if category != "all" { filterArr = append(filterArr, ` { "term":{ - "targetInfo.targetType":"`+category+`" + "targetInfo.targetType.raw":"`+category+`" } }`) @@ -627,7 +1105,7 @@ "sort":[{"picDate":{"order":"desc"}}], "_source": {"includes":[],"excludes":["*.feature"]} }` - logPrint(DSLJson) + //logPrint(DSLJson) buf, err := EsReq("POST", url, []byte(DSLJson)) if err != nil { return aIOceanInfo, err @@ -717,7 +1195,6 @@ } } }` - //logPrint(DSLJson) buf, err := EsReq("POST", url, []byte(DSLJson)) if err != nil { return total, err @@ -755,9 +1232,9 @@ } }, "aggs":{ - "sdkName_status":{ + "taskName_status":{ "terms":{ - "field":"sdkName.raw" + "field":"taskName.raw" } } } @@ -776,11 +1253,11 @@ if !ok { return nil, errors.New("first hits change error!") } - sdkName_status, ok := middle["sdkName_status"].(map[string]interface{}) + sdkName_status, ok := middle["taskName_status"].(map[string]interface{}) if !ok { return nil, errors.New("first hits change error!") } - + //fmt.Println(sdkName_status) for _, in := range sdkName_status["buckets"].([]interface{}) { var source = make(map[string]interface{}, 0) tmpbuf, ok := in.(map[string]interface{}) @@ -1098,7 +1575,7 @@ } wg.Wait() - fmt.Println("lenth_all: ", len(dbinfos)) + //fmt.Println("lenth_all: ", len(dbinfos)) return dbinfos, nil } @@ -1258,3 +1735,66 @@ } return result, nil } + +type ShardInfo struct { + ShardIndex string `json:"shardIndex"` //鍒嗙墖鎵�灞炵储寮曞悕绉� + ShardNum int `json:"shardNum"` //鍒嗙墖鍙� + ShardRole string `json:"shardRole"` //鍒嗙墖瑙掕壊(涓诲垎鐗囷細primary 鍓湰鍒嗙墖锛歳eplica) + ShardState string `json:"shardState"` //鍒嗙墖鐘舵��(鍚敤锛歋TARTED 鏈惎鐢細UNASSIGNED) + ShardDocs int `json:"shardDocs"` //鍒嗙墖宸蹭繚瀛樻枃妗f暟 + ShardStore string `json:"shardStore"` //鍒嗙墖褰撳墠瀛樺偍鏁版嵁澶у皬 + ShardIp string `json:"shardIp"` //鍒嗙墖鎵�鍦ㄨ妭鐐筰p + ShardNode string `json:"shardNode"` //鍒嗙墖鎵�鍦ㄨ妭鐐瑰悕绉� +} + +//鑾峰彇绱㈠紩鍒嗙墖淇℃伅 +func GetShardsByIndex(serverIp string, serverPort string, indexName string) ([]ShardInfo, error) { + url := "http://" + serverIp + ":" + serverPort + "/_cat/shards?v" + buf, err := EsReq("GET", url, []byte("")) + if err != nil { + return nil, err + } + var inf = []ShardInfo{} + res := strings.Split(string(buf), "\n")[1:] + for _, r := range res { + if r != "" { + + inx := strings.Fields(r) + index := inx[0] + shard, _ := strconv.Atoi(inx[1]) + prired := inx[2] + if prired == "r" { + prired = "replica" + } + if prired == "p" { + prired = "primary" + } + state := inx[3] + docs := 0 + store := "" + ip := "" + node := "" + if state == "STARTED" { + docs, _ = strconv.Atoi(inx[4]) + store = inx[5] + ip = inx[6] + node = inx[7] + } + if index == indexName { + inf = append(inf, ShardInfo{ + ShardIndex: index, + ShardNum: shard, + ShardRole: prired, + ShardState: state, + ShardDocs: docs, + ShardStore: store, + ShardIp: ip, + ShardNode: node, + }) + + } + } + + } + return inf, nil +} -- Gitblit v1.8.0