From 86655db5ef2cca9d7cf4eedae879c3a7c4464b9f Mon Sep 17 00:00:00 2001
From: liuxiaolong <736321739@qq.com>
Date: 星期六, 26 十月 2019 16:10:08 +0800
Subject: [PATCH] func GetAllLocalVideopersonsId get all when analyServerId is null

---
 EsApi.go |  285 +++++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 278 insertions(+), 7 deletions(-)

diff --git a/EsApi.go b/EsApi.go
index aabdc97..391f9e6 100644
--- a/EsApi.go
+++ b/EsApi.go
@@ -6,8 +6,9 @@
     "fmt"
     "strconv"
     "strings"
+	"sync"
 
-    "basic.com/pubsub/protomsg.git"
+	"basic.com/pubsub/protomsg.git"
         )
 
 // 鏌ヨ搴曞簱浜哄憳淇℃伅
@@ -286,7 +287,6 @@
 
 //鑾峰彇褰撳墠鑺傜偣鎶撴媿搴撴墍鏈変汉鍛業D
 func GetAllLocalVideopersonsId(compareArgs protomsg.CompareArgs, indexName string, serverIp string, serverPort string) (capturetable []string) {
-	ts := time.Now()
 	queryStr := ""
 	queryBody := compareArgs.InputValue
 	//妫�绱㈡
@@ -327,12 +327,12 @@
 
 	//浣跨敤es搴曞眰鏈哄埗澶勭悊鍒嗛〉
 
+	analyServerFilterStr := ""
 	analyServerId := compareArgs.AnalyServerId
-	if analyServerId == "" {
-		fmt.Println("no analyServerId")
-		return
+	if analyServerId != "" {
+		analyServerFilterStr = "{\"term\":{\"analyServerId\":\"" + analyServerId + "\"}},"
 	}
-	analyServerFilterStr := "{\"term\":{\"analyServerId\":\"" + analyServerId + "\"}},"
+
 
 	//棣栨璇锋眰澶�
 	url := "http://" + serverIp + ":" + serverPort +
@@ -469,7 +469,7 @@
     url := "http://" + serverIp + ":" + serverPort +
         "/" + indexName + "/_search?search_type=dfs_query_then_fetch"
     fmt.Println(url)
-    fmt.Println(prama)
+    fmt.Println(prama)		
     buf, err := EsReq("POST", url,[]byte(prama))
     if err != nil {
         fmt.Println("http request videoUrlInfo info is err!")
@@ -487,3 +487,274 @@
     }
     return tabsource
 }
+
+//鍒濆鍖栧疄鏃舵姄鎷�
+func InitRealTimeCapture(serverIp string, serverPort string, indexName string, isAlarm bool	,quantity int) ([]protomsg.Videopersons, error){
+	var videopersonsInfo []protomsg.Videopersons
+	url := "http://" + serverIp + ":" + serverPort +
+		"/" + indexName + "/_search"
+	queryStr := ""
+	if isAlarm == true {
+		queryStr = `"query":{
+		"match_all":{}
+	},`
+	} else {
+		queryStr = `"query":{
+		"bool":{
+			"filter":[
+				{
+					"term":{
+						"isAlarm":1
+					}
+				}
+			]
+		}
+	},`
+	}
+	DSLJson := `{
+	"size":`+strconv.Itoa(quantity)+`,
+	`+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"]
+	}`
+	buf, err := EsReq("POST", url, []byte(DSLJson))
+	if err != nil {
+		return videopersonsInfo, err
+	}
+
+	sources, err := Sourcelist(buf)
+	if err != nil {
+		return videopersonsInfo, err
+	}
+
+	videoperson := Videopersonsbyid(sources)
+	//fmt.Println(len(videoperson))
+	return videoperson, nil
+}
+
+//瀹炴椂鎶撴媿
+func RealTimeCapture(serverIp string, serverPort string, indexName string, isAlarm bool	) ([]protomsg.Videopersons, error){
+	var videopersonsInfo []protomsg.Videopersons
+	url := "http://" + serverIp + ":" + serverPort +
+		"/" + indexName + "/_search"
+	queryStr := ""
+	if isAlarm == true {
+		fmt.Println("continue")
+	} else {
+		queryStr = `
+				{
+					"term":{
+						"isAlarm":1
+					}
+				}
+					`
+	}
+	DSLJson := `{
+	"size":20,
+	"query":{
+		"bool":{
+			"filter":[
+				{
+					"range":{
+						"picDate":{
+							"gte":"now+8h-30s",
+							"lt":"now+8h"	
+						}		
+					}
+				},
+			`+queryStr+`
+			]
+		}
+	},
+	"_source": ["baseInfo", "alarmRules", "sex", "analyServerName", "sdkName", "ageDescription", "content", "id", "cameraAddr", "picMaxUrl", "picDate", "race", "videoUrl", "picSmUrl", "taskName", "personIsHub", "isAlarm", "analyServerIp", "cameraId", "isAckAlarm"]
+	}`
+
+	buf, err := EsReq("POST", url, []byte(DSLJson))
+	if err != nil {
+		return videopersonsInfo, err
+	}
+
+	sources, err := Sourcelist(buf)
+	if err != nil {
+		return videopersonsInfo, err
+	}
+
+	videoperson := Videopersonsbyid(sources)
+	fmt.Println(len(videoperson))
+	return videoperson, nil
+}
+
+//缁煎悎缁熻
+func StatisticsComprehensive(serverIp string, serverPort string, indexName string, isAlarm bool) (total int, err error){
+	isAlarmStr := ""
+	if isAlarm == true {
+		isAlarmStr = `,{"term":{"isAlarm":1}}`
+	}
+	url := "http://" + serverIp + ":" + serverPort +
+		"/" + indexName + "/_search"
+	DSLJson := `{
+	"size":0,
+	"query":{
+		"bool":{
+			"filter":[{
+				"range":{
+					"picDate":{
+						"gte":"now+8h/d"
+						}
+					}
+				}
+				`+isAlarmStr+`
+				]
+			}
+		}
+	}`
+	//fmt.Println(DSLJson)
+	buf, err := EsReq("POST",url,[]byte(DSLJson))
+	if err != nil {
+		return total, err
+	}
+	var info interface{}
+	json.Unmarshal(buf, &info)
+	out, ok := info.(map[string]interface{})
+	if !ok {
+		return total, errors.New("http response interface can not change map[string]interface{}")
+	}
+	middle, ok := out["hits"].(map[string]interface{})
+	if !ok {
+		return total, errors.New("first hits change error!")
+	}
+	total = int(middle["total"].(float64))
+	//fmt.Println(total)
+	return total,nil
+}
+
+//瀹炴椂鎶ヨ浠诲姟姣旂巼
+func RealTimeAlarmTaskRate(serverIp string, serverPort string, indexName string) (sources []map[string]interface{},err error){
+	url := "http://" + serverIp + ":" + serverPort +
+		"/"+indexName+"/_search"
+	DSLJson := `{
+	"size":0,
+	"query":{
+		"bool":{
+			"filter":[{
+				"range":{
+					"picDate":{
+						"gte":"now+8h/d"
+					}
+				}
+			}]
+		}
+	},
+	"aggs":{
+		"sdkName_status":{
+			"terms":{
+				"field":"taskName.raw"
+			}
+		}
+	}
+}`
+	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!")
+	}
+	sdkName_status, ok := middle["sdkName_status"].(map[string]interface{})
+	if !ok {
+		return nil, errors.New("first hits change error!")
+	}
+
+	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!")
+			continue
+		}
+		sdkName := tmpbuf["key"].(string)
+		count := int(tmpbuf["doc_count"].(float64))
+		source["name"] = sdkName
+		source["value"] = count
+		sources = append(sources, source)
+	}
+	//fmt.Println("tmpSource",sources)
+	return sources,nil
+}
+
+//鑱氬悎浠诲姟鍒楄〃锛宼askId+taskName
+func AggregateTaskList(serverIp string, serverPort string, indexName string) (sources []map[string]interface{},err error){
+	url := "http://" + serverIp + ":" + serverPort +
+		"/"+indexName+"/_search"
+	DSLJson := `{
+    "size": 0,
+    "aggs": {
+        "task_status": {
+            "composite": {
+                "sources": [
+                    {
+                        "taskId": {
+                            "terms": {
+                                "field": "taskId"
+                            }
+                        }
+                    },
+                    {
+                        "taskName": {
+                            "terms": {
+                                "field": "taskName.raw"
+                            }
+                        }
+                    }
+                ],
+                "size":"1000"
+            }
+        }
+    }
+}`
+	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 {
+			fmt.Println("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)
+	}
+	//fmt.Println("tmpSource",sources)
+	return sources,nil
+
+}

--
Gitblit v1.8.0