pubsub/esutil.git

			@@ -285,87 +285,158 @@
			}

			//获取当前节点抓拍库所有人员ID
			func GetAllLocalVideopersonsId(compareArgs protomsg.CompareArgs,indexName string, serverIp string, serverPort 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\"]," +
			"\"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}}],"
			}
			gteDate := compareArgs.SearchTime[0]
			lteDate := compareArgs.SearchTime[1]
			//判断任务ID
			taskIdStr := ""
			taskId := compareArgs.Tasks
			if taskId != nil && len(taskId) > 0 {
			esTaskId := strings.Replace(strings.Trim(fmt.Sprint(taskId), "[]"), " ", "\",\"", -1)
			taskIdStr = "{\"terms\":{\"taskId\":[\"" + esTaskId + "\"]}},"
			}
			//判断摄像机ID
			cameraIdStr := ""
			cameraId := compareArgs.TreeNodes
			if cameraId != nil && len(cameraId) > 0 {
			esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1)
			cameraIdStr = "{\"terms\":{\"cameraId\":[\"" + esCameraId + "\"]}},"
			}
			func GetAllLocalVideopersonsId(compareArgs protomsg.CompareArgs, indexName string, serverIp string, serverPort string) (capturetable []string) {
			ts := time.Now()
			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\"]," +
			"\"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}}],"
			}
			gteDate := compareArgs.SearchTime[0]
			lteDate := compareArgs.SearchTime[1]
			//判断任务ID
			taskIdStr := ""
			taskId := compareArgs.Tasks
			if taskId != nil && len(taskId) > 0 {
			esTaskId := strings.Replace(strings.Trim(fmt.Sprint(taskId), "[]"), " ", "\",\"", -1)
			taskIdStr = "{\"terms\":{\"taskId\":[\"" + esTaskId + "\"]}},"
			}
			//判断摄像机ID
			cameraIdStr := ""
			cameraId := compareArgs.TreeNodes
			if cameraId != nil && len(cameraId) > 0 {
			esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1)
			cameraIdStr = "{\"terms\":{\"cameraId\":[\"" + esCameraId + "\"]}},"
			}

			//判断库表ID
			tableId := compareArgs.Tabs
			esTableId := ""
			esTableIdStr := ""
			if tableId != nil && len(tableId) > 0 {
			esTableId = strings.Replace(strings.Trim(fmt.Sprint(tableId), "[]"), " ", "\",\"", -1)
			esTableIdStr = "{\"terms\":{\"baseInfo.tableId\":[\"" + esTableId + "\"]}},"
			}
			isCollectStr := ""
			isCollect := compareArgs.Collection
			if isCollect != "" {
			isCollectStr = "{\"term\":{\"isCollect\":\"" + isCollect + "\"}},"
			}
			//判断库表ID
			tableId := compareArgs.Tabs
			esTableId := ""
			esTableIdStr := ""
			if tableId != nil && len(tableId) > 0 {
			esTableId = strings.Replace(strings.Trim(fmt.Sprint(tableId), "[]"), " ", "\",\"", -1)
			esTableIdStr = "{\"terms\":{\"baseInfo.tableId\":[\"" + esTableId + "\"]}},"
			}
			isCollectStr := ""
			isCollect := compareArgs.Collection
			if isCollect != "" {
			isCollectStr = "{\"term\":{\"isCollect\":\"" + isCollect + "\"}},"
			}

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

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

			//首次请求头
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search?search_type=dfs_query_then_fetch;scroll=1m"

			var lock sync.RWMutex
			var wg sync.WaitGroup

			for i := 0; i < 32; i++ {
			//请求体
			prama := "{" +
			"\"slice\":{\"id\":" + strconv.Itoa(i) + ",\"max\":48}," +
			"\"size\":\"1000\"," +
			"\"query\":{\"bool\":{" + queryStr +
			"\"filter\":[" +
			cameraIdStr +
			taskIdStr +
			isCollectStr +
			esTableIdStr +
			analyServerFilterStr +
			"{\"range\":{\"picDate\":{\"from\":\"" + gteDate + "\",\"to\":\"" + lteDate + "\",\"include_lower\":true,\"include_upper\":true,\"boost\":1}}}]}}," +
			"\"_source\":[\"id\"]" +
			"}"
			wg.Add(1)
			go func() {
			defer wg.Done()

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

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

			sources, err := Sourcelistforscroll(buf)

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

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

			//scroll请求头
			scroll_url := "http://" + serverIp + ":" + serverPort + "/_search/scroll"

			for {
			var tmpList []string
			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(capturetable))
			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
			}
			//fmt.Println("id")
			for _, source := range nextM {
			tmpList = append(tmpList, source["id"].(string))
			}
			//fmt.Println("tmpList: ", len(tmpList))
			lock.Lock()
			capturetable = append(capturetable, tmpList...)
			lock.Unlock()

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


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

			//请求头
			url := "http://" + serverIp + ":" + serverPort +
			"/" + indexName + "/_search?search_type=dfs_query_then_fetch"
			}()
			}
			wg.Wait()

			//请求体
			prama := "{" +
			"\"size\":\"100000000\"," +
			"\"query\":{\"bool\":{" + queryStr +
			"\"filter\":[" +
			cameraIdStr +
			taskIdStr +
			isCollectStr +
			esTableIdStr +
			analyServerFilterStr +
			"{\"range\":{\"picDate\":{\"from\":\"" + gteDate + "\",\"to\":\"" + lteDate + "\",\"include_lower\":true,\"include_upper\":true,\"boost\":1}}}]}}," +
			"\"_source\":[\"id\"]" +
			"}"
			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
			}
			for _, source := range sources{
			capturetable = append(capturetable, source["id"].(string))
			}
			return capturetable
			//fmt.Println("lenth_all: ", len(capturetable))
			//fmt.Println("耗时：", time.Since(ts))
			return capturetable
			}

			//获取底库人员ID
			@@ -415,4 +486,4 @@
			tabsource[tableId] = append(tabsource[tableId], id)
			}
			return tabsource
			}
			}