pubsub/esutil.git

			@@ -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{}) {
			@@ -54,6 +54,42 @@

			aIOcean := AIOceanAnalysis(sources)
			return aIOcean, nil
			}

			// 根据抓拍人员id查询视频地址
			func AIOceanVideoUrlbyid(id string, indexName string, serverIp string, serverPort string) (string, error) {
			//var aIOceanInfo []protomsg.AIOcean
			//videopersonsPersonId := strings.Replace(strings.Trim(fmt.Sprint(id), "[]"), " ", "\",\"", -1)
			var dbinfoRequest = `
			{
			"query": {
			"bool": {
			"filter": [
			{
			"term": {
			"id": "` + id + `"
			}
			}
			]
			}
			},
			"_source": [
			"videoUrl"
			]
			}
			`
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search", []byte(dbinfoRequest))
			if err != nil {
			return "", err
			}

			sources, err := Sourcelist(buf)
			if err != nil {
			return "", err
			}
			videoUrl := sources[0]["videoUrl"].(string)
			//aIOcean := AIOceanAnalysis(sources)
			return videoUrl, nil
			}

			//根据抓拍库人员id查询特征值
			@@ -159,11 +195,30 @@

			}

			//根据摄像机列表和时间查询人员浏览轨迹
			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": [
			@@ -171,62 +226,193 @@
			"range": {
			"picDate": {
			"gte": "` + startTime + `",
			"lte": "` + endTime + `"
			"lte": "` + endTime + `"
			}
			}
			},
			{
			"term": {
			"targetInfo.targetType.raw": "Yolo"
			}
			},
			{
			"terms": {
			"cameraId": [
			"` + esCameraId + `"
			]
			"term":{
			"targetInfo.targetType.raw": "FaceDetect"
			}
			}
			]
			}
			},
			"size": 2147483647,
			"_source": {
			"includes": [
			"cameraId",
			"cameraName",
			"cameraAddr",
			"targetInfo.targetScore",
			"picDate",
			"updateTime",
			"picMaxUrl",
			"targetInfo.belongsTargetId",
			"targetInfo.targetLocation"
			]
			"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"
			]
			}
			}
			}
			}
			}
			}
			}`
			fmt.Println(personUrl)
			fmt.Println(personBody)
			buf, err := EsReq("POST", personUrl, []byte(personBody))
			buf, err := EsReq("POST", requestUrl, []byte(requestBody))
			if err != nil {
			return nil, err
			}

			sources, err := Sourcelist(buf)
			source, err := FaceSourceAggregations(buf, thresholdTime, thresholdStayTime)
			if err != nil {
			return nil, err
			}
			resData,err := PerSonAnalysis(sources)
			//println(sources)
			return resData, nil

			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 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)
			var buckersUrl = "http://" + serverIp + ":" + ServerPort + "/" + indexName + "/_search"
			var buckersBody = `{
			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": [
			@@ -242,12 +428,235 @@
			"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": [
			{
			"range": {
			"picDate": {
			"gte": "` + startTime + `",
			"lte": "` + endTime + `"
			}
			}
			},
			{
			"terms": {
			"cameraId": ["` + esCameraId + `"]
			"term": {
			"targetInfo.targetType.raw": "Yolo"
			}
			}
			]
			}
			},
			"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 + `
			]
			}
			},
			"size": 2147483647,
			"_source": {
			"includes": [
			"cameraId",
			"cameraName",
			"cameraAddr",
			"targetInfo.targetScore",
			"picDate",
			"updateTime",
			"picMaxUrl",
			"targetInfo.belongsTargetId",
			"targetInfo.targetLocation",
			"picWH"
			]
			}
			}`
			//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)
			source["result"] = resData
			source["total"] = len(resData)
			source["queryUseTime"] = queryUseTime
			//println(sources)
			return source, nil

			}

			//根据时间范围，摄像机列表，分组聚合人脸列表,返回分组数据
			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": [
			` + queryStr + `
			]
			}
			},
			@@ -271,7 +680,7 @@
			}
			}
			],
			"size": 100000000
			"size": 10000000
			},
			"aggs":{
			"top_attention_hits":{
			@@ -285,7 +694,7 @@
			}
			],
			"_source":{
			"includes":["baseInfo.targetId","cameraId","cameraName","cameraAddr","targetInfo.targetScore","targetInfo.picSmUrl","showLabels","baseInfo.tableId","baseInfo.tableName","baseInfo.bwType","baseInfo.targetName","baseInfo.compareScore","picDate","picMaxUrl"]
			"includes":["baseInfo.targetId","cameraId","cameraName","cameraAddr","targetInfo.targetScore","targetInfo.picSmUrl","showLabels","baseInfo.tableId","baseInfo.tableName","baseInfo.bwType","baseInfo.targetName","baseInfo.compareScore","picDate","picMaxUrl","picWH"]
			}
			}
			}
			@@ -295,33 +704,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更新（picurl）图片地址
			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"
			@@ -345,7 +860,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!")
			@@ -360,7 +875,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 {
			@@ -409,18 +924,25 @@
			return statu, errors.New("http response interface can not change map[string]interface{}")
			}
			middle, ok := out["updated"].(float64)
			if !ok {
			batches, ok1 := out["batches"].(float64)
			if !ok \|\| !ok1 {
			logPrint("first updated change error!")
			statu = 500
			return statu, errors.New("first updated change error!")
			}
			if middle == 1 {
			statu = 200
			return statu, nil
			}
			if middle == 0 {
			statu = 201
			return statu, errors.New("已经修改")
			if batches == 0 {
			logPrint("no such doc in database")
			statu = 400
			return statu, errors.New("目标数据不存在")
			} else {
			if middle == 1 {
			statu = 200
			return statu, nil
			}
			if middle == 0 {
			statu = 201
			return statu, errors.New("已经修改")
			}
			}
			return statu, nil
			}
			@@ -465,7 +987,12 @@
			isCollectStr := ""
			isCollect := compareArgs.Collection
			if isCollect != "" {
			isCollectStr = "{\"term\":{\"isCollect\":\"" + isCollect + "\"}},"
			//isCollectStr = "{\"term\":{\"isCollect\":\"" + isCollect + "\"}},"
			if isCollect == "1" {
			isCollectStr = "{\"term\":{\"isCollect\":true}},"
			} else if isCollect == "0" {
			isCollectStr = "{\"term\":{\"isCollect\":false}},"
			}
			}

			//判断布防等级
			@@ -606,7 +1133,7 @@
			if category != "all" {
			filterArr = append(filterArr, ` {
			"term":{
			"targetInfo.targetType":"`+category+`"
			"targetInfo.targetType.raw":"`+category+`"
			}
			}`)

			@@ -626,7 +1153,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
			@@ -716,7 +1243,6 @@
			}
			}
			}`
			//logPrint(DSLJson)
			buf, err := EsReq("POST", url, []byte(DSLJson))
			if err != nil {
			return total, err
			@@ -754,9 +1280,9 @@
			}
			},
			"aggs":{
			"sdkName_status":{
			"taskName_status":{
			"terms":{
			"field":"sdkName.raw"
			"field":"taskName.raw"
			}
			}
			}
			@@ -775,11 +1301,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{})
			@@ -798,10 +1324,19 @@
			}

			//聚合任务列表，taskId+taskName
			func AggregateTaskList(serverIp string, serverPort string, indexName string, analyServerId string) (sources []map[string]interface{}, err error) {
			func AggregateTaskList(serverIp string, serverPort string, indexName string, analyServerId string, cameraIds []string) (sources []map[string]interface{}, err error) {
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search"
			serverFilterStr := ""
			cameIdFilterStr := ""
			if cameraIds != nil && len(cameraIds) > 0 {
			cameIdsStr := strings.Replace(strings.Trim(fmt.Sprint(cameraIds), "[]"), " ", "\",\"", -1)
			cameIdFilterStr = `,{
			"term": {
			"cameraId": "` + cameIdsStr + `"
			}
			}`
			}
			if analyServerId != "" {
			serverFilterStr = `,
			"query": {
			@@ -810,8 +1345,9 @@
			{
			"term": {
			"analyServerId": "` + analyServerId + `"
			}
			}
			}
			` + cameIdFilterStr + `
			]
			}
			}`
			@@ -1097,7 +1633,7 @@
			}
			wg.Wait()

			fmt.Println("lenth_all: ", len(dbinfos))
			//fmt.Println("lenth_all: ", len(dbinfos))

			return dbinfos, nil
			}
			@@ -1143,7 +1679,7 @@
			}

			//按日期范围，服务器Id删除数据
			func DeleteAnalyServerData(serverIp string, serverPort string, indexName string, startTime string, endTime string, analyServerId string) (result bool, err error) {
			func DeleteAnalyServerData(serverIp string, serverPort string, indexName string, startTime string, endTime string, analyServerId string) (total int, err error, ) {
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_delete_by_query"
			deleteJson := `{
			"query":{
			@@ -1165,20 +1701,17 @@
			}
			}
			} `
			fmt.Println(url)
			fmt.Println(deleteJson)
			buf, err := EsReq("POST", url, []byte(deleteJson))
			if err != nil {
			return false, errors.New("请求失败")
			return -1, errors.New("请求失败")
			}
			deleteRes, err := SourceDeleted(buf)
			if err != nil {
			return false, errors.New("解码失败")
			return -1, errors.New("解码失败")
			}
			if deleteRes == -1 {
			result = false
			} else {
			result = true
			}
			return result, nil
			return deleteRes, nil
			}

			//给所有节点追加删除任务信息
			@@ -1257,3 +1790,66 @@
			}
			return result, nil
			}

			type ShardInfo struct {
			ShardIndex string `json:"shardIndex"` //分片所属索引名称
			ShardNum int `json:"shardNum"` //分片号
			ShardRole string `json:"shardRole"` //分片角色(主分片：primary 副本分片：replica)
			ShardState string `json:"shardState"` //分片状态(启用：STARTED 未启用：UNASSIGNED)
			ShardDocs int `json:"shardDocs"` //分片已保存文档数
			ShardStore string `json:"shardStore"` //分片当前存储数据大小
			ShardIp string `json:"shardIp"` //分片所在节点ip
			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
			}