pubsub/esutil.git

			@@ -1,111 +1,781 @@
			package esutil

			import (
			"encoding/json"
			"errors"
			"fmt"
			"strconv"
			"strings"
			"basic.com/pubsub/protomsg.git"
			"encoding/json"
			"errors"
			"fmt"
			"sort"
			"strconv"
			"strings"
			"sync"
			"time"
			)

			"basic.com/pubsub/protomsg.git"
			)

			// 查询底库人员信息
			func Personinfos( queryIndex int, queryNums int, indexName string, serverIp string, serverPort string, analyServerId string) ([]*protomsg.Esinfo, error){
			var dbinfos []*protomsg.Esinfo
			point := strconv.Itoa(queryIndex)
			number := strconv.Itoa(queryNums)
			JsonDSL := ""
			if indexName == "videopersons" {
			JsonDSL = ` {
			"from": ` + point + `,
			"query": {
			"bool": {
			"filter": [
			{
			"term": {
			"analyServerId": "` + analyServerId + `"
			}
			}
			]
			}
			},
			"size":`+ number +`,
			"_source": [
			"id",
			"faceFeature"
			]
			}`
			}else {
			JsonDSL = ` {
			"from": ` + point + `,
			"query": {
			"match_all": {}
			},
			"size":`+ number +`,
			"_source": [
			"id",
			"tableId",
			"faceFeature"
			]
			}`
			}
			//fmt.Println("url: "+"http://"+serverIp+":"+serverPort+"/"+indexName+"/_search","body: ",JsonDSL)
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(JsonDSL))
			if err != nil {
			return dbinfos ,errors.New("http request dbtablename info is err!")
			}

			// 返回 _source 数组
			sources, err := Sourcelist(buf)
			if err != nil {
			return dbinfos,err
			}

			// 返回所有查询的数据
			dbpersoninfos := Parsesources(sources)
			return dbpersoninfos, nil
			var logPrint = func(i ...interface{}) {
			fmt.Println(i)
			}

			// 根据底库id查询底库信息
			func Dbtablefosbyid (id []string, indexName string, serverIp string, serverPort string) ([]protomsg.Dbtable, error) {
			var dbinfo []protomsg.Dbtable
			dbtableId := strings.Replace(strings.Trim(fmt.Sprint(id), "[]"), " ", "\",\"", -1)
			var dbinfoRequest = `
			func InitLog(fn func(i ...interface{})) {
			if fn != nil {
			logPrint = fn
			}
			}

			//*********************重庆Start********************************//

			//判断时间是否再范围之内
			func isTimeInRange(timeStr, startStr, endStr string) bool {
			layout := "15:04:05"

			timeStamp, err := time.Parse(layout, timeStr)
			if err != nil {
			fmt.Println("Error parsing timestamp:", err)
			return false
			}

			startTime, err := time.Parse(layout, startStr)
			if err != nil {
			fmt.Println("Error parsing start time:", err)
			return false
			}

			endTime, err := time.Parse(layout, endStr)
			if err != nil {
			fmt.Println("Error parsing end time:", err)
			return false
			}

			if startTime.After(endTime) {
			// 跨越日期的情况
			//fmt.Println("跨日期",timeStamp, timeStamp.After(startTime), timeStamp.Before(endTime))
			return timeStamp.After(startTime) \|\| timeStamp.Before(endTime)
			} else {
			// 不跨越日期的情况
			//fmt.Println("不跨日期",timeStamp, timeStamp.After(startTime), timeStamp.Before(endTime))
			return timeStamp.After(startTime) && timeStamp.Before(endTime)
			}
			}

			//判断两个时间先后
			func compareTimes(time1Str, time2Str string) int {
			layout := "15:04:05"

			time1, err := time.Parse(layout, time1Str)
			if err != nil {
			fmt.Println("Error parsing time 1:", err)
			return 0
			}

			time2, err := time.Parse(layout, time2Str)
			if err != nil {
			fmt.Println("Error parsing time 2:", err)
			return 0
			}

			if time1.Before(time2) {
			return -1 // time1 在 time2 之前
			} else if time1.After(time2) {
			return 1 // time1 在 time2 之后
			} else {
			return 0 // time1 和 time2 相等
			}
			}

			//判断日期相差几天
			func daysBetweenDates(date1Str, date2Str string) int {
			layout := "2006-01-02"

			date1, err := time.Parse(layout, date1Str)
			if err != nil {
			fmt.Println("Error parsing date 1:", err)
			return 0
			}

			date2, err := time.Parse(layout, date2Str)
			if err != nil {
			fmt.Println("Error parsing date 2:", err)
			return 0
			}

			duration := date2.Sub(date1)
			days := int(duration.Hours() / 24)

			return days
			}

			//计算时间阈值
			func checkTimeDifference(timestampStr1 string, timestampStr2 string, intervalInMinutes int) bool {
			layout := "2006-01-02 15:04:05"
			timestampStr1 = timestampStr1[:19]
			timestampStr2 = timestampStr2[:19]
			// 将字符串解析为时间
			time1, err := time.Parse(layout, timestampStr1)
			if err != nil {
			fmt.Println("时间解析失败:", err)
			return false
			}
			time2, err := time.Parse(layout, timestampStr2)
			if err != nil {
			fmt.Println("时间解析失败:", err)
			return false
			}

			// 计算时间差
			diff := time2.Sub(time1)

			// 检查时间差是否小于等于指定的间隔
			if diff.Minutes() <= float64(intervalInMinutes) {
			return true
			} else {
			return false
			}
			}

			////格式化时间hh:mm:ss:zzz to hh:mm:ss
			//func formatTime(inputTime string) (string, error) {
			// parsedTime, err := time.Parse("15:04:05:000", inputTime)
			// if err != nil {
			// return "", err
			// }
			//
			// formattedTime := parsedTime.Format("15:04:05")
			// return formattedTime, nil
			//}

			func decodeActivityId(frequency int, intervalInMinutes int, source []map[string]interface{}) ([]map[string]interface{}, error) {
			docInfo := make([]map[string]interface{}, 0)
			for _, info := range source {
			tmpInfo := make(map[string]interface{})
			documentNumber := info["key"].(string)
			tmpInfo["documentNumber"] = documentNumber
			//fmt.Println("documentNumber: ", documentNumber)
			topHits := info["top_hits"].(map[string]interface{})
			hits := topHits["hits"].(map[string]interface{})
			hitsResult := hits["hits"].([]interface{})
			dataId := make([]string, 0)
			picUrl := ""
			if hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["baseInfo"] != nil {
			baseInfo := hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["baseInfo"]
			if v, ok := baseInfo.([]interface{}); ok {
			picUrl = v[0].(map[string]interface{})["targetPicUrl"].(string)
			} else if v, ok := baseInfo.(map[string]interface{}); ok {
			picUrl = v["targetPicUrl"].(string)
			}
			} else {
			if hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["targetInfo"] != nil {
			picUrl = hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["targetInfo"].([]interface{})[0].(map[string]interface{})["picSmUrl"].(string)
			}
			}
			tmpInfo["picUrl"] = picUrl
			lastCaptureTime := ""
			for _, sourceInfo := range hitsResult {
			rSourceInfo := sourceInfo.(map[string]interface{})
			source := rSourceInfo["_source"].(map[string]interface{})
			captureTime := source["picDate"].(string)
			id := source["id"].(string)
			if lastCaptureTime != "" {
			if checkTimeDifference(lastCaptureTime[:19], captureTime[:19], intervalInMinutes) {
			//fmt.Println("时间小于连续阈值", lastCaptureTime, captureTime)
			continue
			}
			}
			lastCaptureTime = captureTime
			dataId = append(dataId, id)
			}

			if len(dataId) > frequency {
			tmpInfo["id"] = dataId
			docInfo = append(docInfo, tmpInfo)
			}
			}
			return docInfo, nil
			}

			func DayNightActivityQuery(comIds []string, cameraIds []string, docNumber string, startDateStr string, endDateStr string, activeHour string, frequency int,
			intervalInMinutes int, indexName string, serverIp string, serverPort string) (map[string]interface{}, error) {
			esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			hours := strings.Split(activeHour, "-")
			startTimeStr := hours[0]
			endTimeStr := hours[1]
			layoutTime := "15:04:05"
			startTime, err := time.Parse(layoutTime, startTimeStr)
			if err != nil {
			return nil, err
			}
			endTime, err := time.Parse(layoutTime, endTimeStr)
			if err != nil {
			return nil, err
			}
			layoutDate := "2006-01-02"
			startDate, err := time.Parse(layoutDate, startDateStr)
			if err != nil {
			fmt.Println("Error parsing start time:", err)
			return nil, err
			}

			endDate, err := time.Parse(layoutDate, endDateStr)
			if err != nil {
			fmt.Println("Error parsing end time:", err)
			return nil, err
			}
			rangePicDate := make([]string, 0)
			if startTime.Before(endTime) {
			for !startDate.After(endDate) {
			sDateTime := startDate.Format(layoutDate) + " " + startTimeStr
			eDateTime := startDate.Format(layoutDate) + " " + endTimeStr
			rangePicDate = append(rangePicDate, `
			{
			"range": {
			"picDate": {
			"gte": "`+sDateTime+`",
			"lt": "`+eDateTime+`"
			}
			}
			}`)
			startDate = startDate.AddDate(0, 0, 1)
			}
			} else if startTime.After(endTime) {
			if startDate.Year() == endDate.Year() && startDate.Month() == endDate.Month() && startDate.Day() == endDate.Day() {
			fmt.Println("跨日期不支持单日查询")
			return nil, errors.New("跨日期不支持单日查询！")
			}
			for !startDate.After(endDate) {
			sDateTime := startDate.Format(layoutDate) + " " + startTimeStr
			startDate = startDate.AddDate(0, 0, 1)
			//fendDate := startDate.AddDate(0, 0, 1)
			if startDate.Year() == endDate.Year() && startDate.Month() == endDate.Month() && startDate.Day() == endDate.Day() {
			eDateTime := startDate.Format(layoutDate) + " " + endTimeStr
			rangePicDate = append(rangePicDate, `
			{
			"range": {
			"picDate": {
			"gte": "`+sDateTime+`",
			"lt": "`+eDateTime+`"
			}
			}
			}`)
			//fmt.Println("startDate: ", startDate, "/t endDate: ", endDate)
			break
			}
			eDateTime := startDate.Format(layoutDate) + " " + endTimeStr
			rangePicDate = append(rangePicDate, `
			{
			"range": {
			"picDate": {
			"gte": "`+sDateTime+`",
			"lt": "`+eDateTime+`"
			}
			}
			}`)
			}
			}
			//fmt.Println(rangePicDate)
			//return nil, nil
			filterDocIdAttr := ""
			if docNumber != "" {
			filterDocIdAttr = "{\"term\": {\"documentNumber\": \"" + docNumber + "\"}},"
			}
			comIdsStr := ""
			if comIds != nil && len(comIds) > 0 {
			esComIds := strings.Replace(strings.Trim(fmt.Sprint(comIds), "[]"), " ", "\",\"", -1)
			comIdsStr = "{\"terms\":{\"communityId\":[\"" + esComIds + "\"]}},"
			}
			cameraIdsStr := ""
			if comIds != nil && len(cameraIds) > 0 {
			esCameraIds := strings.Replace(strings.Trim(fmt.Sprint(cameraIds), "[]"), " ", "\",\"", -1)
			cameraIdsStr = "{\"terms\":{\"cameraId\":[\"" + esCameraIds + "\"]}},"
			}

			var result = make(map[string]interface{})
			docInfos := make([]map[string]interface{}, 0)
			for _, rpd := range rangePicDate {
			//if comIdsStr != "" \|\| filterDocIdAttr != "" \|\| cameraIdsStr != "" {
			// rpd += ","
			//}
			queryDSL := `
			{
			"size": 0,
			"query": {
			"bool": {
			"filter": [
			` + comIdsStr + `
			` + cameraIdsStr + `
			` + filterDocIdAttr + `
			` + rpd + `
			],
			"must_not": [
			{
			"term": {
			"documentNumber": ""
			}
			}
			]
			}
			},
			"aggs": {
			"group_by_documentnumber": {
			"terms": {
			"field": "documentNumber",
			"size": 100000
			},
			"aggs": {
			"top_hits": {
			"top_hits": {
			"_source": [
			"id",
			"picDate",
			"baseInfo.targetPicUrl",
			"targetInfo.picSmUrl"
			],
			"size": 100000,
			"sort": [
			{
			"picDate": {
			"order": "asc"
			}
			}
			]
			}
			}
			}
			}
			}
			}`
			fmt.Println(esURL)
			fmt.Println(queryDSL)
			buf, err := EsReq("POST", esURL, []byte(queryDSL))
			if err != nil {
			return nil, err
			}
			source, err := SourceAggregationList(buf)
			if err != nil {
			return nil, err
			}
			//fmt.Println(source)
			//docResult 本次查询结果
			docResult, err := decodeActivityId(frequency, intervalInMinutes, source)
			if err != nil {
			return nil, err
			}
			if len(docResult) == 0 {
			continue
			}
			if len(docInfos) == 0 {
			docInfos = append(docInfos, docResult...)
			continue
			}
			for _, dr := range docResult {
			var found bool
			for _, di := range docInfos {
			if dr["documentNumber"].(string) == di["documentNumber"].(string) {
			// 找到了相同的documentNumber，将id合并到di["id"]中
			ids, ok := di["id"].([]string)
			if !ok {
			// 如果id不是字符串切片类型，则进行初始化
			ids = []string{}
			}
			ids = append(ids, dr["id"].([]string)...)
			di["id"] = ids
			//docInfos[diIndex] = di // 更新原始docInfos中的值
			found = true
			break
			}
			}
			if !found {
			// 没有找到相同的documentNumber，将dr添加到docInfos中
			docInfos = append(docInfos, dr)
			}
			}
			}
			if len(docInfos) == 0 {
			return nil, nil
			}
			DataInfos, err := GetInfosByIds(docInfos[0]["id"].([]string), indexName, serverIp, serverPort)
			result["documentNumbers"] = docInfos
			result["datalist"] = DataInfos
			return result, nil
			}

			type acmInfo struct {
			documentNumber string
			camerasInfos []camerasInfo
			}

			type camerasInfo struct {
			cameraId string
			captureInfos []captureInfo
			}

			type captureInfo struct {
			id string
			picDate string
			}

			func addSecondsToTimestamp(timestamp string, seconds int) (string, error) {
			parsedTime, err := time.Parse("2006-01-02 15:04:05", timestamp)
			if err != nil {
			return "", err
			}
			newTime := parsedTime.Add(time.Second * time.Duration(seconds))
			newTimestamp := newTime.Format("2006-01-02 15:04:05")
			return newTimestamp, nil
			}

			func decodeInfo(intervalInMinutes int, source []map[string]interface{}) ([]acmInfo, error) {
			acmInfos := make([]acmInfo, 0)
			for _, info := range source {
			var aInfo acmInfo
			documentNumber := info["key"].(string)
			aInfo.documentNumber = documentNumber
			groupByCameraId := info["group_by_cameraId"].(map[string]interface{})
			cameraBuckets := groupByCameraId["buckets"].([]interface{})
			for _, cameraInfo := range cameraBuckets {
			var camsInfo camerasInfo
			cInfo := cameraInfo.(map[string]interface{})
			cameraId := cInfo["key"].(string)
			camsInfo.cameraId = cameraId
			dataBuckets := cInfo["top_hits"].(map[string]interface{})["hits"].(map[string]interface{})["hits"].([]interface{})
			markTime := ""
			for _, dataInfo := range dataBuckets {
			var capInfo captureInfo
			dInfo := dataInfo.(map[string]interface{})
			dSource := dInfo["_source"].(map[string]interface{})
			id := dSource["id"].(string)
			picDate := dSource["picDate"].(string)
			//addFlag := false
			if markTime == "" {
			//addFlag = true
			markTime = picDate
			} else {
			if checkTimeDifference(markTime, picDate, intervalInMinutes) {
			//fmt.Println(markTime, picDate)
			markTime = picDate
			continue
			}
			markTime = picDate
			}
			capInfo.id = id
			capInfo.picDate = picDate
			camsInfo.captureInfos = append(camsInfo.captureInfos, capInfo)
			}
			aInfo.camerasInfos = append(aInfo.camerasInfos, camsInfo)
			}
			acmInfos = append(acmInfos, aInfo)
			}
			return acmInfos, nil
			}

			type addResultIds struct {
			documentNumber string
			unionIds []unionId
			}

			type unionId struct {
			baseId string
			targetId string
			}

			func addResultInfo(source []map[string]interface{}, targetAddResultIds *[]addResultIds, bId string) {
			found := false
			for _, info := range source {
			documentNumber := info["key"].(string)
			dataBuckets := info["top_hits"].(map[string]interface{})["hits"].(map[string]interface{})["hits"].([]interface{})
			id := dataBuckets[0].(map[string]interface{})["_source"].(map[string]interface{})["id"].(string)
			//fmt.Println("documentNumber: ", documentNumber, "\tid: ", id)
			for i, docInfo := range *targetAddResultIds {
			if documentNumber == docInfo.documentNumber {
			//fmt.Println("新更新")
			(targetAddResultIds)[i].unionIds = append((targetAddResultIds)[i].unionIds, unionId{baseId: bId, targetId: id})
			found = true
			break
			}
			}
			if !found {
			//fmt.Println("新添加")
			var targetAddResultId addResultIds
			targetAddResultId.documentNumber = documentNumber
			targetAddResultId.unionIds = append(targetAddResultId.unionIds, unionId{baseId: bId, targetId: id})
			targetAddResultIds = append(targetAddResultIds, targetAddResultId)
			}

			}
			}

			func removeDuplicates(nums []string) []string {
			result := make([]string, 0)
			seen := make(map[string]bool)

			for _, num := range nums {
			if !seen[num] {
			result = append(result, num)
			seen[num] = true
			}
			}

			return result
			}

			func findAnalyzeCoordinatedMovementsInfos(infos []acmInfo, docNumber string, beforeTime int, afterTime int, frequency int,
			indexName string, serverIp string, serverPort string) (map[string]interface{}, error) {
			//baseAddResultIds := make([]addResultIds, 0)
			targetAddResultIds := make([]addResultIds, 0)
			esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			for _, info := range infos {
			for _, cInfo := range info.camerasInfos {
			for _, pInfo := range cInfo.captureInfos {
			gteDate, err := addSecondsToTimestamp(pInfo.picDate[:19], beforeTime)
			if err != nil {
			fmt.Println(err)
			}
			lteDate, err := addSecondsToTimestamp(pInfo.picDate[:19], afterTime)
			if err != nil {
			fmt.Println(err)
			}
			queryDSL := `
			{
			"size": 0,
			"query": {
			"bool": {
			"filter": [
			{
			"range": {
			"picDate": {
			"gte": "` + gteDate + `",
			"lte": "` + lteDate + `"
			}
			}
			},
			{
			"term": {
			"cameraId": "` + cInfo.cameraId + `"
			}
			}
			],
			"must_not": [
			{
			"term": {
			"documentNumber": "` + docNumber + `"
			}
			}
			]
			}
			},
			"aggs": {
			"group_by_documentnumber": {
			"terms": {
			"field": "documentNumber",
			"size": 100000
			},
			"aggs": {
			"top_hits": {
			"top_hits": {
			"_source": [
			"id",
			"cameraId",
			"picDate"
			],
			"size": 10000,
			"sort": [
			{
			"picDate": {
			"order": "asc"
			}
			}
			]
			}
			}
			}
			}
			}
			}`
			//fmt.Println(esURL, queryDSL)
			buf, err := EsReq("POST", esURL, []byte(queryDSL))
			if err != nil {
			return nil, err
			}
			source, err := SourceAggregationList(buf)
			if err != nil {
			return nil, err
			}
			//fmt.Println("pInfo.id: ", pInfo.id)
			addResultInfo(source, &targetAddResultIds, pInfo.id)
			//fmt.Println("targetAddResultIds: ", targetAddResultIds)
			if err != nil {
			return nil, err
			}
			//fmt.Println(source)
			}
			}
			}
			//fmt.Println("targetAddResultIds: ", targetAddResultIds)
			baseIds := make([]string, 0)
			targetIds := make([]string, 0)
			for _, tAIdInfo := range targetAddResultIds {
			if len(tAIdInfo.unionIds) >= frequency {
			for _, unionId := range tAIdInfo.unionIds {
			baseIds = append(baseIds, unionId.baseId)
			targetIds = append(targetIds, unionId.targetId)
			}
			}
			}

			rdbaseIds := removeDuplicates(baseIds)
			rdtargetIds := removeDuplicates(targetIds)
			baseInfos, err := GetInfosByIds(rdbaseIds, indexName, serverIp, serverPort)
			if err != nil {
			return nil, err
			}
			targetInfos, err := GetInfosByIds(rdtargetIds, indexName, serverIp, serverPort)
			if err != nil {
			return nil, err
			}
			docNumberMap := make(map[string][]interface{})
			for _, tinfo := range targetInfos {
			docNumber := tinfo["documentNumber"].(string)
			docNumberMap[docNumber] = append(docNumberMap[docNumber], tinfo)
			}
			targetRecordInfos := make([]map[string]interface{}, 0)
			for docNumber, infos := range docNumberMap {
			ifs := make(map[string]interface{})
			ifs["documentNumber"] = docNumber
			ifs["recordInfos"] = infos
			targetRecordInfos = append(targetRecordInfos, ifs)
			}
			result := make(map[string]interface{})
			result["baseRecordInfo"] = baseInfos
			result["targetRecordInfo"] = targetRecordInfos
			return result, nil
			}

			func AnalyzeCoordinatedMovements(comIds []string, docNumber string, startDate string, endDate string, beforeTime int, afterTime int, frequency int,
			intervalInMinutes int, indexName string, serverIp string, serverPort string) (map[string]interface{}, error) {
			esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			//判断社区IDs
			comIdsStr := ""
			if comIds == nil \|\| len(comIds) > 0 {
			esComIds := strings.Replace(strings.Trim(fmt.Sprint(comIds), "[]"), " ", "\",\"", -1)
			comIdsStr = "{\"terms\":{\"communityId\":[\"" + esComIds + "\"]}},"
			}
			queryDSL := `
			{
			"size": 0,
			"query": {
			"bool": {
			"filter": [
			{
			"range": {
			"picDate": {
			"gte": "` + startDate + `",
			"lte": "` + endDate + `"
			}
			}
			},
			` + comIdsStr + `
			{
			"term": {
			"documentNumber": "` + docNumber + `"
			}
			}
			]
			}
			},
			"aggs": {
			"group_by_documentnumber": {
			"terms": {
			"field": "documentNumber",
			"size": 100000
			},
			"aggs": {
			"group_by_cameraId": {
			"terms": {
			"field": "cameraId",
			"size": 10000
			},
			"aggs": {
			"top_hits": {
			"top_hits": {
			"_source": [
			"id",
			"cameraId",
			"picDate"
			],
			"size": 10000,
			"sort": [
			{
			"picDate": {
			"order": "asc"
			}
			}
			]
			}
			}
			}
			}
			}
			}
			}
			}`
			//fmt.Println(esURL)
			//fmt.Println(queryDSL)
			buf, err := EsReq("POST", esURL, []byte(queryDSL))
			if err != nil {
			return nil, err
			}
			source, err := SourceAggregationList(buf)
			if err != nil {
			return nil, err
			}
			docResult, err := decodeInfo(intervalInMinutes, source)
			if err != nil {
			return nil, err
			}
			//fmt.Println(docResult)
			result, err := findAnalyzeCoordinatedMovementsInfos(docResult, docNumber, beforeTime, afterTime, frequency, indexName, serverIp, serverPort)
			if err != nil {
			return nil, err
			}
			//fmt.Println(result)
			return result, nil
			}

			func GetInfosByIds(ids []string, indexName string, serverIp string, serverPort string) ([]map[string]interface{}, error) {
			captureIds := strings.Replace(strings.Trim(fmt.Sprint(ids), "[]"), " ", "\",\"", -1)
			esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			queryDSL := `
			{
			"query": {
			"bool": {
			"filter": [{
			"terms": {
			"id": [
			"`+ dbtableId +`"
			"` + captureIds + `"
			]
			}
			}]
			}
			},
			"size":1000000
			"size":1000000,
			"sort":[{"picDate":{"order":"desc"}}],
			"_source": {"includes":[],"excludes":["*.feature"]}
			}
			`
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfoRequest))
			if err != nil {
			return dbinfo, err
			}
			`
			buf, err := EsReq("POST", esURL, []byte(queryDSL))
			if err != nil {
			return nil, err
			}
			sources, err := Sourcelist(buf)
			if err != nil {
			return nil, err
			}

			sources, err := Sourcelist(buf)
			if err != nil {
			return dbinfo , err
			}

			dbtable := Dbtablebyid(sources)
			return dbtable, nil
			return sources, nil
			}

			// *********************重庆End**********************************//
			// 根据抓拍人员id查询抓拍人员信息
			func Videopersonsinfosbyid (id []string, indexName string, serverIp string, serverPort string) ([]protomsg.Videopersons, error) {
			var videopersonsInfo []protomsg.Videopersons
			func AIOceaninfosbyid(id []string, indexName string, serverIp string, serverPort string) ([]protomsg.AIOcean, error) {
			var aIOceanInfo []protomsg.AIOcean
			videopersonsPersonId := strings.Replace(strings.Trim(fmt.Sprint(id), "[]"), " ", "\",\"", -1)
			var dbinfoRequest = `
			{
			@@ -114,7 +784,7 @@
			"filter": [{
			"terms": {
			"id": [
			"`+ videopersonsPersonId +`"
			"` + videopersonsPersonId + `"
			]
			}
			}]
			@@ -123,178 +793,922 @@
			"size":1000000
			}
			`
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfoRequest))
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search", []byte(dbinfoRequest))
			if err != nil {
			return videopersonsInfo, err
			return aIOceanInfo, err
			}

			sources, err := Sourcelist(buf)
			if err != nil {
			return videopersonsInfo , err
			return aIOceanInfo, err
			}

			videoperson := Videopersonsbyid(sources)
			println(videoperson)
			return videoperson,nil
			aIOcean := AIOceanAnalysis(sources)
			return aIOcean, nil
			}

			// 根据底库人员id查询底库人员信息
			func Dbpersoninfosbyid (id []string, indexName string, serverIp string, serverPort string) ([]protomsg.Dbperson, error) {
			var dbinfo []protomsg.Dbperson
			dbtablePersonId := strings.Replace(strings.Trim(fmt.Sprint(id), "[]"), " ", "\",\"", -1)
			var dbinfoRequest = `
			{
			"query": {
			"bool": {
			"filter": [{
			"terms": {
			"id": [
			"`+ dbtablePersonId +`"
			]
			}
			}]
			}
			},
			"size":1000000
			}
			// 根据抓拍人员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 dbinfo, err
			}
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search", []byte(dbinfoRequest))
			if err != nil {
			return "", err
			}

			sources, err := Sourcelist(buf)
			if err != nil {
			return dbinfo , err
			}

			dbperson := Dbpersonbyid(sources)
			println(dbperson)
			return dbperson,nil
			sources, err := Sourcelist(buf)
			if err != nil {
			return "", err
			}
			videoUrl := sources[0]["videoUrl"].(string)
			//aIOcean := AIOceanAnalysis(sources)
			return videoUrl, nil
			}
			//根据抓拍库人员id查询特征值
			func GetVideoPersonFaceFeatureById (id string, indexName string, serverIp string, serverPort string) (string, error) {
			var jsonDSL = `

			// 根据抓拍库人员id查询特征值
			func GetVideoPersonFaceFeatureById(id string, indexName string, serverIp string, serverPort string) (string, error) {
			var jsonDSL = `
			{
			"query": {
			"bool": {
			"filter": [{
			"term": {
			"id":"`+ id +`"
			"id":"` + id + `"
			}
			}]
			}
			},
			"_source":["faceFeature"]
			"_source":["targetInfo.feature"]
			}
			`
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(jsonDSL))
			if err != nil {
			return "", err
			}
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search", []byte(jsonDSL))
			if err != nil {
			return "", err
			}

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

			feature := sources[0]["targetInfo"].([]interface{})[0].(map[string]interface{})["feature"].(string)
			return feature, nil
			}

			// 根据tableid 查询tablename
			func Dbtablename(tableid string, indexName string, serverIp string, serverPort string) (tablename string, err error) {
			var dbinfotable =` {
			"query": {
			"bool": {
			"filter": [
			// 根据目标id查询已追加条数
			func GetLinkTagInfoSize(id string, indexName string, serverIp string, serverPort string) (size int, err error) {
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			queryDSL := `{
			"query": {
			"term":{
			"id":"` + id + `"
			}
			}
			}`
			buf, err := EsReq("POST", url, []byte(queryDSL))
			if err != nil {
			return -1, err
			}
			source, err := Sourcelist(buf)
			if err != nil {
			return -1, err
			}
			if source[0]["linkTagInfo"] != nil {
			size = len(source[0]["linkTagInfo"].([]interface{}))
			} else {
			return -1, errors.New("该数组不存在")
			}
			return size, nil
			}

			// 根据目标id追加跟踪信息
			func AppendTargetInfo(id string, targetInfo string, indexName string, serverIp string, serverPort string, updateTime string) (string, error) {
			if targetInfo == "" {
			return "", errors.New("append data is nil")
			}
			var info interface{}
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query?refresh=true"
			jsonDSL := `{
			"query": {
			"term":{
			"id":"` + id + `"
			}
			},
			"script": {
			"lang": "painless",
			"inline": "ctx._source.linkTagInfo.add(params.newparam);ctx._source.updateTime='` + updateTime + `'",
			"params": {
			"newparam": ` + targetInfo + `
			}
			}
			}`
			logPrint(jsonDSL)
			buf, err := EsReq("POST", url, []byte(jsonDSL))
			if err != nil {
			return "", err
			}
			json.Unmarshal(buf, &info)
			out, ok := info.(map[string]interface{})
			logPrint(out)
			if !ok {
			return "", errors.New("http response interface can not change map[string]interface{}")
			}
			middle, ok := out["updated"].(float64)
			if !ok {
			return "", errors.New("first updated change error!")
			}
			mes := ""
			if middle == 1 {
			mes = "追加成功"
			}
			if middle == 0 {
			mes = "已经追加"
			}
			return mes, nil

			}

			/************************************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": [
			{
			"term": {
			"id":"`+tableid+`"
			"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"
			]
			}
			}
			}
			}
			},
			"_source": [
			"tableName"
			],
			"size":1000000
			}
			`
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfotable))
			if err != nil {
			return "" ,errors.New("http request dbtablename info is err!")
			}
			sources, err := Sourcelist(buf)
			if err != nil {
			return "",err
			}
			for _, source := range sources {
			if name, ok := source["tableName"].(string); ok {
			tablename = name
			break
			}
			}
			return tablename, nil
			}`
			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
			}

			//根据抓拍人员id更新（videourl）摄像机地址
			func UpdateVideourlById(id string, videoUrl string, indexName string, serverIp string, serverPort string)(statu int){
			var info interface{}
			var videoUrlInfo = `
			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": [
			{
			"range": {
			"picDate": {
			"gte": "` + startTime + `",
			"lte": "` + endTime + `"
			}
			}
			},
			{
			"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 + `
			]
			}
			},
			"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)
			sources := make(map[string]interface{})
			queryStartTime := time.Now()
			buf, err := EsReq("POST", buckersUrl, []byte(buckersBody))
			if err != nil {
			return nil, err
			}
			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")
			tRes, err := AIOceaninfosbyid([]string{id}, indexName, serverIp, serverPort)
			if err != nil \|\| len(tRes) == 0 {
			return err
			}
			picMaxUrls := tRes[0].PicMaxUrl
			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 + `'"`
			}
			var info interface{}
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query?refresh=true"

			var picUrlInfo = `
			{
			"script": {
			"source": "ctx._source.videoUrl='` + videoUrl + `'"
			` + sourceStr + `
			},
			"query": {
			"bool": {
			"filter": [
			{
			"term": {
			"id": "` + id + `"
			}
			}
			]
			}
			}
			}
			`
			//logPrint("url: ", url, videoUrlInfo)
			//fmt.Println(url, picUrlInfo)
			buf, err := EsReq("POST", url, []byte(picUrlInfo))
			if err != nil {
			logPrint("http request videoUrlInfo info is err!")
			return err
			}
			json.Unmarshal(buf, &info)
			//logPrint(info)
			out, ok := info.(map[string]interface{})
			if !ok {
			logPrint("http response interface can not change map[string]interface{}")
			return errors.New("http response interface can not change map[string]interface{}")
			}
			middle, ok := out["updated"].(float64)
			if !ok {
			logPrint("first updated change error!", out)
			return errors.New("first updated change error!")
			}
			if middle == 1 {
			return nil
			}
			if middle == 0 {
			return errors.New("已经修改")
			}
			return nil
			}

			// 根据抓拍人员id更新（videourl）摄像机地址
			func UpdateVideourlById(id string, videoUrl string, indexName string, serverIp string, serverPort string, command int) (statu int, err error) {

			var info interface{}
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query?refresh=true"
			sourceStr := "ctx._source.videoUrl='" + videoUrl + "'"
			if command >= 0 {
			sourceStr = "ctx._source.linkTagInfo[" + strconv.Itoa(command) + "].videoUrl='" + videoUrl + "'"
			}
			var videoUrlInfo = `
			{
			"script": {
			"source": "` + sourceStr + `"
			},
			"query": {
			"term": {
			"id": "` +id+ `"
			"id": "` + id + `"
			}
			},
			"size":1000000
			}
			}
			`
			buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_update_by_query",[]byte(videoUrlInfo))
			if err != nil {
			fmt.Println("http request videoUrlInfo info is err!")
			statu = 500
			return
			}
			json.Unmarshal(buf, &info)
			out, ok := info.(map[string]interface{})
			if !ok {
			fmt.Println("http response interface can not change map[string]interface{}")
			statu = 500
			return
			}

			middle, ok := out["updated"].(float64)
			if !ok {
			fmt.Println("first updated change error!")
			statu = 500
			return
			}
			if middle == 1{
			statu = 200
			return
			}
			if middle == 0{
			statu = 201
			return
			}
			return statu
			//logPrint("url: ", url, videoUrlInfo)
			buf, err := EsReq("POST", url, []byte(videoUrlInfo))
			if err != nil {
			logPrint("http request videoUrlInfo info is err!")
			statu = 500
			return statu, err
			}
			json.Unmarshal(buf, &info)
			//logPrint(info)
			out, ok := info.(map[string]interface{})
			if !ok {
			logPrint("http response interface can not change map[string]interface{}")
			statu = 500
			return statu, errors.New("http response interface can not change map[string]interface{}")
			}
			middle, ok := out["updated"].(float64)
			batches, ok1 := out["batches"].(float64)
			if !ok \|\| !ok1 {
			logPrint("first updated change error!")
			statu = 500
			return statu, errors.New("first updated change error!")
			}
			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
			}

			//获取当前节点抓拍库所有人员ID
			func GetAllLocalVideopersonsId(compareArgs protomsg.CompareArgs, indexName string, serverIp string, serverPort string) (capturetable []string) {
			ts := time.Now()
			// 获取当前节点抓拍库所有人员ID缓存
			func GetAllLocalVideopersonsId(compareArgs protomsg.CompareArgs, indexName string, serverIp string, serverPort string, alarmLevelTypes string) (capturetable []string) {
			queryStr := ""
			queryBody := compareArgs.InputValue
			//检索框
			if queryBody != "" {
			queryStr = "\"must\":[{\"multi_match\":{\"query\":\"" + queryBody + "\",\"fields\":[\"alarmRules.alarmLevel^1.5\",\"ageDescription^1.5\",\"taskName^1.5\",\"baseInfo.tableName^1.5\",\"sex^2.0\",\"race^2.0\",\"content^1.0\",\"baseInfo.idCard^1.8\",\"cameraAddr^1.0\"]," +
			queryStr = "\"must\":[{\"multi_match\":{\"query\":\"" + queryBody + "\",\"fields\":[\"cameraAddr^1.5\",\"taskName^1.5\",\"sdkName^1.5\",\"showLabels^3.0\",\"baseInfo.tableName^1.5\",\"baseInfo.targetName^1.5\",\"baseInfo.labels^1.5\",\"alarmRules.alarmLevel^1.5\",\"linkTag^1.5\"]," +
			"\"type\":\"cross_fields\",\"operator\":\"OR\",\"slop\":0,\"prefix_length\":0,\"max_expansions\":50,\"zero_terms_query\":\"NONE\",\"auto_generate_synonyms_phrase_query\":true,\"fuzzy_transpositions\":true,\"boost\":1}}],"
			}
			if compareArgs.SearchTime == nil \|\| len(compareArgs.SearchTime) != 2 {
			return nil
			}
			gteDate := compareArgs.SearchTime[0]
			lteDate := compareArgs.SearchTime[1]
			@@ -324,18 +1738,29 @@
			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}},"
			}
			}

			//判断布防等级
			alarmLevelStr := ""
			if alarmLevelTypes != "" {
			alarmLevelStr = "{\"terms\":{\"alarmRules.alarmLevel.raw\":[\"" + alarmLevelTypes + "\"]}},"
			}

			//使用es底层机制处理分页

			analyServerFilterStr := ""
			analyServerId := compareArgs.AnalyServerId
			if analyServerId == "" {
			fmt.Println("no analyServerId")
			return
			if analyServerId != "" {
			analyServerFilterStr = "{\"term\":{\"analyServerId\":\"" + analyServerId + "\"}},"
			}
			analyServerFilterStr := "{\"term\":{\"analyServerId\":\"" + analyServerId + "\"}},"

			ts := time.Now()
			//首次请求头
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search?search_type=dfs_query_then_fetch;scroll=1m"
			@@ -343,14 +1768,16 @@
			var lock sync.RWMutex
			var wg sync.WaitGroup

			for i := 0; i < 32; i++ {
			for i := 0; i < 48; i++ {
			//请求体
			prama := "{" +
			"\"slice\":{\"id\":" + strconv.Itoa(i) + ",\"max\":48}," +
			"\"size\":\"1000\"," +
			"\"query\":{\"bool\":{" + queryStr +
			"\"filter\":[" +
			"{\"term\":{\"targetInfo.targetType.raw\":\"FaceDetect\"}}," +
			cameraIdStr +
			alarmLevelStr +
			taskIdStr +
			isCollectStr +
			esTableIdStr +
			@@ -359,27 +1786,29 @@
			"\"_source\":[\"id\"]" +
			"}"
			wg.Add(1)
			go func() {
			go func(reqParam string) {
			defer wg.Done()

			//fmt.Println(url)
			//fmt.Println(prama)
			buf, err := EsReq("POST", url, []byte(prama))
			logPrint(url)
			logPrint(prama)
			buf, err := EsReq("POST", url, []byte(reqParam))

			if err != nil {
			fmt.Println("http request videoUrlInfo info is err!")
			fmt.Println(len(capturetable))
			logPrint("http request videoUrlInfo info is err!")
			logPrint(len(capturetable))
			return
			}

			sources, err := Sourcelistforscroll(buf)

			if err != nil {
			fmt.Println(len(capturetable))
			logPrint(len(capturetable))
			return
			}
			for _, source := range sources["sourcelist"].([]map[string]interface{}) {
			lock.Lock()
			capturetable = append(capturetable, source["id"].(string))
			lock.Unlock()
			}

			scroll_id := sources["scroll_id"].(string)
			@@ -397,12 +1826,12 @@
			"scroll": "1m",
			"scroll_id" : "` + scroll_id + `"
			}`
			//fmt.Println(scroll_url)
			//fmt.Println(jsonDSL)
			logPrint(scroll_url)
			logPrint(jsonDSL)
			buf, err := EsReq("POST", scroll_url, []byte(jsonDSL))

			if err != nil {
			fmt.Println("lenth1: ", len(capturetable))
			logPrint("lenth1: ", len(capturetable))
			return
			}
			nextSources, err := Sourcelistforscroll(buf)
			@@ -412,16 +1841,16 @@
			}

			nextM := nextSources["sourcelist"].([]map[string]interface{})
			//fmt.Println("id",nextSources)
			//logPrint("id",nextSources)
			if nextM == nil \|\| len(nextM) == 0 {
			//fmt.Println("lenth: ", len(capturetable))
			//logPrint("lenth: ", len(capturetable))
			return
			}
			//fmt.Println("id")
			//logPrint("id")
			for _, source := range nextM {
			tmpList = append(tmpList, source["id"].(string))
			}
			//fmt.Println("tmpList: ", len(tmpList))
			//logPrint("tmpList: ", len(tmpList))
			lock.Lock()
			capturetable = append(capturetable, tmpList...)
			lock.Unlock()
			@@ -429,129 +1858,73 @@
			next_scroll_id = nextSources["scroll_id"].(string)
			}


			fmt.Println(len(capturetable))

			}()
			}(prama)
			}
			wg.Wait()

			//fmt.Println("lenth_all: ", len(capturetable))
			//fmt.Println("耗时：", time.Since(ts))
			logPrint("lenth_all: ", len(capturetable))
			logPrint("耗时：", time.Since(ts))
			return capturetable
			}

			//获取底库人员ID
			func GetDbpersonsId(compareArgs protomsg.CompareArgs,indexName string, serverIp string, serverPort string) (source map[string][]string) {
			queryStr := ""
			queryBody := compareArgs.InputValue
			//检索框
			if queryBody != "" {
			queryStr = "\"must\":[{\"multi_match\":{\"query\":\"" + queryBody + "\",\"fields\":[\"personName^1.5\",\"age^1.5\",\"idCard^1.5\",\"phoneNum^1.5\",\"sex^2.0\",\"reserved^2.0\"]," +
			"\"type\":\"cross_fields\",\"operator\":\"OR\",\"slop\":0,\"prefix_length\":0,\"max_expansions\":50,\"zero_terms_query\":\"NONE\",\"auto_generate_synonyms_phrase_query\":true,\"fuzzy_transpositions\":true,\"boost\":1}}],"
			}
			//判断库表ID
			tableId := compareArgs.Tabs
			esTableId := ""
			esTableIdStr := ""
			if tableId != nil && len(tableId) > 0 {
			esTableId = strings.Replace(strings.Trim(fmt.Sprint(tableId), "[]"), " ", "\",\"", -1)
			esTableIdStr = "{\"terms\":{\"tableId\":[\"" + esTableId + "\"]}}"
			}

			prama := "{" +
			"\"size\":\"100000000\"," +
			"\"query\":{\"bool\":{" + queryStr +
			"\"filter\":[" +
			esTableIdStr +
			"]}}," +
			"\"_source\":[\"id\",\"tableId\"]" +
			"}"

			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search?search_type=dfs_query_then_fetch"
			fmt.Println(url)
			fmt.Println(prama)
			buf, err := EsReq("POST", url,[]byte(prama))
			if err != nil {
			fmt.Println("http request videoUrlInfo info is err!")
			return
			}
			sources, err := Sourcelist(buf)
			if err != nil {
			return
			}
			tabsource := make(map[string][]string)
			for _, source := range sources{
			tableId := source["tableId"].(string)
			id := source["id"].(string)
			tabsource[tableId] = append(tabsource[tableId], id)
			}
			return tabsource
			}

			//初始化实时抓拍
			func InitRealTimeCapture(serverIp string, serverPort string, indexName string, isAlarm bool ) ([]protomsg.Videopersons, error){
			var videopersonsInfo []protomsg.Videopersons
			// 初始化实时抓拍
			func InitRealTimeCapture(serverIp string, serverPort string, indexName string, isAlarm string, category string, quantity int) ([]protomsg.AIOcean, error) {
			var aIOceanInfo []protomsg.AIOcean
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search"
			queryStr := ""
			if isAlarm == true {
			queryStr = `"query":{
			"match_all":{}
			},`
			} else {
			queryStr = `"query":{
			var filterArr []string
			if isAlarm != "all" {
			filterArr = append(filterArr, ` {
			"term":{
			"isAlarm":"`+isAlarm+`"
			}
			}`)
			}

			if category != "all" {
			filterArr = append(filterArr, ` {
			"term":{
			"targetInfo.targetType.raw":"`+category+`"
			}
			}`)

			}

			queryStr := `"query":{
			"bool":{
			"filter":[
			{
			"term":{
			"isAlarm":1
			}
			}
			` + strings.Join(filterArr, ",") + `
			]
			}
			},`
			}
			DSLJson := `{
			"size":20,
			`+queryStr+`
			"sort":[{"picDate":{"order":"desc"}}],
			"_source": ["baseInfo", "alarmRules", "sex", "analyServerName", "sdkName", "ageDescription", "content", "id", "cameraAddr", "picMaxUrl", "picDate", "race", "videoUrl", "picSmUrl", "taskName", "personIsHub", "isAlarm", "analyServerIp", "cameraId"]
			}`

			DSLJson := `{
			"size":` + strconv.Itoa(quantity) + `,
			` + queryStr + `
			"sort":[{"picDate":{"order":"desc"}}],
			"_source": {"includes":[],"excludes":["*.feature"]}
			}`
			//logPrint(DSLJson)
			buf, err := EsReq("POST", url, []byte(DSLJson))
			if err != nil {
			return videopersonsInfo, err
			return aIOceanInfo, err
			}

			sources, err := Sourcelist(buf)
			if err != nil {
			return videopersonsInfo, err
			return aIOceanInfo, err
			}

			videoperson := Videopersonsbyid(sources)
			//fmt.Println(len(videoperson))
			return videoperson, nil
			aIOcean := AIOceanAnalysis(sources)
			//logPrint(len(videoperson))
			return aIOcean, nil
			}

			//实时抓拍
			func RealTimeCapture(serverIp string, serverPort string, indexName string, isAlarm bool ) ([]protomsg.Videopersons, error){
			var videopersonsInfo []protomsg.Videopersons
			// 实时抓拍
			func RealTimeCapture(serverIp string, serverPort string, indexName string, isAlarm bool) ([]protomsg.AIOcean, error) {
			var aIOceanInfo []protomsg.AIOcean
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search"
			queryStr := ""
			if isAlarm == true {
			fmt.Println("continue")
			} else {
			queryStr = `
			{
			"term":{
			"isAlarm":1
			}
			}
			`
			}
			DSLJson := `{
			"size":20,
			"query":{
			@@ -565,47 +1938,63 @@
			}
			}
			},
			`+queryStr+`
			{
			"term":{
			"isAlarm":` + strconv.FormatBool(isAlarm) + `
			}
			}
			]
			}
			},
			"_source": ["baseInfo", "alarmRules", "sex", "analyServerName", "sdkName", "ageDescription", "content", "id", "cameraAddr", "picMaxUrl", "picDate", "race", "videoUrl", "picSmUrl", "taskName", "personIsHub", "isAlarm", "analyServerIp", "cameraId", "isAckAlarm"]
			"_source": {"includes":[],"excludes":["*.feature"]}
			}`

			buf, err := EsReq("POST", url, []byte(DSLJson))
			if err != nil {
			return videopersonsInfo, err
			return aIOceanInfo, err
			}

			sources, err := Sourcelist(buf)
			if err != nil {
			return videopersonsInfo, err
			return aIOceanInfo, err
			}

			videoperson := Videopersonsbyid(sources)
			fmt.Println(len(videoperson))
			return videoperson, nil
			aIOcean := AIOceanAnalysis(sources)
			logPrint(len(aIOcean))
			return aIOcean, nil
			}

			//综合统计
			func StatisticsComprehensive(serverIp string, serverPort string, indexName string) (total int, err error){
			// 综合统计
			func StatisticsComprehensive(serverIp string, serverPort string, indexName string, isAlarm string) (total int, err error) {
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search"
			isAlarmStr := ""
			if isAlarm != "all" {
			isAlarmStr = ` {
			"term":{
			"isAlarm":"` + isAlarm + `"
			}
			},`

			}
			DSLJson := `{
			"size":0,
			"query":{
			"bool":{
			"filter":[{
			"filter":[
			` + isAlarmStr + `
			{
			"range":{
			"picDate":{
			"gte":"now+8H/d"
			"gte":"now+8h/d"
			}
			}
			}]
			}
			]
			}
			}
			}`
			buf, err := EsReq("POST",url,[]byte(DSLJson))
			buf, err := EsReq("POST", url, []byte(DSLJson))
			if err != nil {
			return total, err
			}
			@@ -620,13 +2009,14 @@
			return total, errors.New("first hits change error!")
			}
			total = int(middle["total"].(float64))
			//fmt.Println(total)
			return total,nil
			//logPrint(total)
			return total, nil
			}
			//实时报警任务比率
			func RealTimeAlarmTaskRate(serverIp string, serverPort string,indexName string) (sources map[string]int,err error){

			// 实时报警任务比率
			func RealTimeAlarmTaskRate(serverIp string, serverPort string, indexName string) (sources []map[string]interface{}, err error) {
			url := "http://" + serverIp + ":" + serverPort +
			"/"+indexName+"/_search"
			"/" + indexName + "/_search"
			DSLJson := `{
			"size":0,
			"query":{
			@@ -641,15 +2031,15 @@
			}
			},
			"aggs":{
			"sdkName_status":{
			"taskName_status":{
			"terms":{
			"field":"taskName.raw"
			}
			}
			}
			}`
			buf, err := EsReq("POST",url,[]byte(DSLJson))
			if err != nil {
			buf, err := EsReq("POST", url, []byte(DSLJson))
			if err != nil {
			return nil, err
			}
			var info interface{}
			@@ -662,23 +2052,592 @@
			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!")
			}
			var source = make(map[string]int,0)
			for _, in := range sdkName_status["buckets"].([]interface{}){
			//fmt.Println(sdkName_status)
			for _, in := range sdkName_status["buckets"].([]interface{}) {
			var source = make(map[string]interface{}, 0)
			tmpbuf, ok := in.(map[string]interface{})
			if !ok {
			fmt.Println("change to source error!")
			logPrint("change to source error!")
			continue
			}
			sdkName := tmpbuf["key"].(string)
			count := int(tmpbuf["doc_count"].(float64))
			source[sdkName] = count
			//fmt.Println("in",in)
			//sources[in["key"].(string)] = int(in["doc_count"].(float64))
			source["name"] = sdkName
			source["value"] = count
			sources = append(sources, source)
			}
			//fmt.Println("sources",source)
			return source,nil
			//logPrint("tmpSource",sources)
			return sources, nil
			}

			// 聚合任务列表，taskId+taskName
			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": {
			"bool": {
			"filter": [
			{
			"term": {
			"analyServerId": "` + analyServerId + `"
			}
			}
			` + cameIdFilterStr + `
			]
			}
			}`
			}
			DSLJson := `{
			"size": 0,
			"aggs": {
			"task_status": {
			"composite": {
			"sources": [
			{
			"taskId": {
			"terms": {
			"field": "taskId"
			}
			}
			},
			{
			"taskName": {
			"terms": {
			"field": "taskName.raw"
			}
			}
			}
			],
			"size":"1000000"
			}
			}
			}
			` + serverFilterStr + `
			}`
			buf, err := EsReq("POST", url, []byte(DSLJson))
			if err != nil {
			return nil, err
			}
			var info interface{}
			json.Unmarshal(buf, &info)
			out, ok := info.(map[string]interface{})
			if !ok {
			return nil, errors.New("http response interface can not change map[string]interface{}")
			}
			middle, ok := out["aggregations"].(map[string]interface{})
			if !ok {
			return nil, errors.New("first hits change error!")
			}
			task_status, ok := middle["task_status"].(map[string]interface{})
			if !ok {
			return nil, errors.New("first hits change error!")
			}

			for _, in := range task_status["buckets"].([]interface{}) {
			var source = make(map[string]interface{}, 0)
			tmpbuf, ok := in.(map[string]interface{})
			if !ok {
			logPrint("change to source error!")
			continue
			}
			task := tmpbuf["key"].(map[string]interface{})
			count := int(tmpbuf["doc_count"].(float64))
			taskName := task["taskName"].(string)
			taskId := task["taskId"].(string)
			source["taskName"] = taskName
			source["taskId"] = taskId
			source["count"] = count
			sources = append(sources, source)
			}
			//logPrint("tmpSource",sources)
			return sources, nil

			}

			// 添加即将删除信号
			func AddDeleteSignal() {

			}

			/**************************************以下为sdkCompare比对缓存使用方法*******************************************/
			//获取查询总数缓存
			func GetTotal(serverIp string, serverPort string, indexName string, shards string, targetType string) (total int) {
			JsonDSL := `{
			"size": 0,
			"query": {
			"bool": {
			"filter": [{
			"term": {
			"targetInfo.targetType.raw": "` + targetType + `"
			}
			}]
			}
			}
			}`

			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search?preference=_shards:" + shards + "\|_only_local"
			buf, err := EsReq("POST", url, []byte(JsonDSL))
			if err != nil {
			return
			}
			var info interface{}
			json.Unmarshal(buf, &info)
			out, ok := info.(map[string]interface{})
			if !ok {
			return
			}
			middle, ok := out["hits"].(map[string]interface{})
			if !ok {
			return
			}
			total = int(middle["total"].(float64))
			return total

			}

			// 查询时间段数据缓存
			func GetPeriodInfos(serverIp string, serverPort string, startTime string, endTime string, indexName string, shards string, targetType string) ([]*protomsg.MultiFeaCache, error) {
			var capdbinfo []*protomsg.MultiFeaCache
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search?preference=_shards:" + shards + "\|_only_local"
			var source []string
			switch targetType {
			case "face", "FaceDetect":
			source = []string{"id", "targetInfo.feature", "analyServerId", "cameraId"}
			case "track":
			source = []string{"id", "targetInfo.feature", "analyServerId", "cameraId", "targetInfo.attachTarget.feature", "targetInfo.targetLocation", "linkTagInfo.targetInfo.feature", "linkTagInfo.targetInfo.attachTarget.feature", "linkTagInfo.cameraId", "linkTagInfo.targetInfo.targetLocation"}
			}
			JsonDSL := `
			{
			"query": {
			"bool": {
			"filter": [
			{
			"term": {
			"targetInfo.targetType.raw": "` + targetType + `"
			}
			},
			{
			"range": {
			"picDate": {
			"gte": "` + startTime + `",
			"lt": "` + endTime + `"
			}
			}
			}
			]
			}
			},
			"size": 1000000,
			"_source": ["` + strings.Replace(strings.Trim(fmt.Sprint(source), "[]"), " ", "\",\"", -1) + `"]
			}
			`
			//logger.Debug(url)
			//logger.Debug(JsonDSL)
			//logPrint(JsonDSL)
			buf, err := EsReq("POST", url, []byte(JsonDSL))
			if err != nil {
			return capdbinfo, errors.New("http request dbtablename info is err!")
			}

			// 返回 _source 数组
			sources, err := Sourcelist(buf)
			if err != nil {
			return capdbinfo, err
			}
			//logPrint(sources)
			// 返回所有查询的数据
			capdbinfos := Parsesources(sources)
			return capdbinfos, nil
			}

			// 查询底库人员信息缓存
			func GetOceanFeatures(serverIp string, serverPort string, queryNums int, indexName string, shards string, targetType string) ([]*protomsg.MultiFeaCache, error) {
			//queryIndexNum int
			//var dbinfos []*protomsg.MultiFeaCache
			dbinfos := make([]*protomsg.MultiFeaCache, 0)
			//dbinfosss := make([]*protomsg.MultiFeaCache,0)
			//dbinfoss = append(dbinfoss, dbinfosss...)

			JsonDSL := ""
			var source []string
			switch targetType {
			case "face", "FaceDetect":
			source = []string{"id", "targetInfo.feature", "analyServerId"}
			case "track":
			source = []string{"id", "targetInfo.feature", "analyServerId", "targetInfo.attachTarget.feature", "targetInfo.targetLocation", "linkTagInfo.targetInfo.feature", "linkTagInfo.targetInfo.attachTarget.feature", "linkTagInfo.targetInfo.targetLocation"}
			}

			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search?preference=_shards:" + shards + "\|_only_local;scroll=1m"

			var lock sync.RWMutex
			var wg sync.WaitGroup

			for i := 0; i < 48; i++ {
			//请求体
			JsonDSL = ` {
			"slice": {
			"id": "` + strconv.Itoa(i) + `",
			"max": 48
			},
			"size":` + strconv.Itoa(queryNums) + `,
			"query": {
			"bool": {
			"filter": [
			{
			"term": {
			"targetInfo.targetType.raw": "` + targetType + `"
			}
			}
			]
			}
			},
			"_source": ["` + strings.Replace(strings.Trim(fmt.Sprint(source), "[]"), " ", "\",\"", -1) + `"]
			}`
			wg.Add(1)
			go func(reqJsonDSL string) {
			defer wg.Done()

			//fmt.Println(url)
			//fmt.Println(prama)
			//logPrint("url: ",url)
			//logPrint("url: ",reqJsonDSL)
			buf, err := EsReq("POST", url, []byte(reqJsonDSL))
			if err != nil {
			logPrint("EsReq: ", err)
			return
			}

			// 返回 _source 数组
			sources, err := Sourcelistforscroll(buf)
			if err != nil {
			logPrint("EsReq: ", err)
			return
			}
			// 返回所有查询的数据
			ftmpDatas := Parsesources(sources["sourcelist"].([]map[string]interface{}))
			lock.Lock()
			dbinfos = append(dbinfos, ftmpDatas...)
			//logPrint("prsLen: ", len(Parsesources(sources["sourcelist"].([]map[string]interface{}))))
			//logPrint("dbinfosLen: ", len(dbinfos))
			lock.Unlock()

			scroll_id := sources["scroll_id"].(string)

			//scroll请求头
			scroll_url := "http://" + serverIp + ":" + serverPort + "/_search/scroll"
			for {
			next_scroll_id := ""
			if next_scroll_id != "" {
			scroll_id = next_scroll_id
			}
			jsonDSL := `{
			"scroll": "1m",
			"scroll_id" : "` + scroll_id + `"
			}`
			//fmt.Println(scroll_url)
			//fmt.Println(jsonDSL)
			buf, err := EsReq("POST", scroll_url, []byte(jsonDSL))

			if err != nil {
			//fmt.Println("lenth1: ", len(dbinfos))
			return
			}
			nextSources, err := Sourcelistforscroll(buf)

			if nextSources == nil {
			return
			}

			nextM := nextSources["sourcelist"].([]map[string]interface{})
			//fmt.Println("id",nextSources)
			if nextM == nil \|\| len(nextM) == 0 {
			//fmt.Println("lenth: ", len(capturetable))
			return
			}
			tmpDatas := Parsesources(nextM)
			lock.Lock()
			dbinfos = append(dbinfos, tmpDatas...)
			//logPrint("tmpDatasLen: ", len(tmpDatas))
			//logPrint("AdbinfosLen: ", len(dbinfos))
			lock.Unlock()

			next_scroll_id = nextSources["scroll_id"].(string)
			}

			}(JsonDSL)
			}
			wg.Wait()

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

			return dbinfos, nil
			}

			// **********************CORN TASK*****************************
			// 查询日期范围内是否还存在数据
			func QueryAnalyServerData(serverIp string, serverPort string, indexName string, startTime string, endTime string, analyServerId string) (result bool, err error) {
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			deleteJson := `{
			"query":{
			"bool":{
			"filter":[{
			"range":{
			"updateTime":{
			"gte":"` + startTime + `",
			"lte":"` + endTime + `"
			}
			}
			},
			{
			"term":{
			"analyServerId":"` + analyServerId + `"
			}
			}
			]
			}
			}
			} `
			buf, err := EsReq("POST", url, []byte(deleteJson))
			if err != nil {
			return false, errors.New("请求失败")
			}
			resTotal, err := SourceTotal(buf)
			if err != nil {
			return false, errors.New("解码失败")
			}
			if resTotal == -1 \|\| resTotal == 0 {
			result = false
			} else {
			result = true
			}
			return result, nil
			}

			// 按日期范围，服务器Id删除数据
			func DeleteByDocumentNumber(docNumber []string, serverIp string, serverPort string, indexName string) (total int, err error) {

			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_delete_by_query"
			docNumbers := strings.Replace(strings.Trim(fmt.Sprint(docNumber), "[]"), " ", "\",\"", -1)
			deleteJson := `{
			"query":{
			"bool":{
			"filter":[
			{
			"terms":{
			"documentNumber":["` + docNumbers + `"]
			}
			}
			]
			}
			}
			} `
			//fmt.Println(url)
			//fmt.Println(deleteJson)
			//return
			buf, err := EsReq("POST", url, []byte(deleteJson))
			if err != nil {
			return -1, errors.New("请求失败")
			}
			deleteRes, err := SourceDeleted(buf)
			if err != nil {
			return -1, errors.New("解码失败")
			}
			return deleteRes, nil
			}

			//func GetCaptureDaysByDocumentNumber(docNumber string, comId string, indexName string, serverIp string, serverPort string){
			// url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search"
			// queryDSL := ``
			//}

			// 按日期范围，服务器Id删除数据
			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":{
			"bool":{
			"filter":[{
			"range":{
			"updateTime":{
			"gte":"` + startTime + `",
			"lte":"` + endTime + `"
			}
			}
			},
			{
			"term":{
			"analyServerId":"` + analyServerId + `"
			}
			}
			]
			}
			}
			} `
			fmt.Println(url)
			fmt.Println(deleteJson)
			buf, err := EsReq("POST", url, []byte(deleteJson))
			if err != nil {
			return -1, errors.New("请求失败")
			}
			deleteRes, err := SourceDeleted(buf)
			if err != nil {
			return -1, errors.New("解码失败")
			}
			return deleteRes, nil
			}

			// 给所有节点追加删除任务信息
			func AddDelTask(serverIp string, serverPort string, indexName string, startTime string, endTime string, analyServerId string) (result bool, err error) {
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query"
			addJson := `{
			"script": {
			"lang":"painless",
			"inline": "ctx._source.instantTask.add(params.newtask)",
			"params": {
			"newtask": {
			"instantClearId": "` + analyServerId + `",
			"startTime": "` + startTime + `",
			"endTime": "` + endTime + `"
			}
			}
			},
			"query": {
			"bool": {
			"filter": [
			{
			"term": {
			"application": "loopCoverage"
			}
			}
			]
			}
			}
			}`
			buf, err := EsReq("POST", url, []byte(addJson))
			if err != nil {
			return false, errors.New("请求失败")
			}
			updateRes, err := SourceUpdated(buf)
			if err != nil {
			return false, errors.New("解码失败")
			}
			if updateRes == -1 {
			result = false
			} else {
			result = true
			}
			return result, nil
			}

			// 移除已执行完的删除任务
			func DeleteDelTask(serverIp string, serverPort string, indexName string, analyServerId string) (result bool, err error) {
			url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query"
			deleteJson := `{
			"script": {
			"lang":"painless",
			"inline": "ctx._source.instantTask.remove(0)"
			},
			"query": {
			"bool": {
			"filter":[{
			"term":{
			"id":"` + analyServerId + `"
			}
			}]
			}
			}
			}`
			buf, err := EsReq("POST", url, []byte(deleteJson))
			if err != nil {
			return false, errors.New("请求失败")
			}
			updateRes, err := SourceUpdated(buf)
			if err != nil {
			return false, errors.New("解码失败")
			}
			if updateRes == -1 {
			result = false
			} else {
			result = true
			}
			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
			}