func GetAllLocalVideopersonsId get all when analyServerId is null

liuxiaolong

2019-10-26 86655db5ef2cca9d7cf4eedae879c3a7c4464b9f

 EsApi.go

@@ -6,8 +6,9 @@
    "fmt"
    "strconv"
    "strings"
   "sync"

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

// 查询底库人员信息
@@ -83,7 +84,8 @@
                        }
                     }]
                  }
               }
               },
                "size":1000000
            }
        `
    buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfoRequest))
@@ -98,6 +100,41 @@

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

// 根据抓拍人员id查询抓拍人员信息
func Videopersonsinfosbyid (id []string, indexName string, serverIp string, serverPort string) ([]protomsg.Videopersons, error) {
   var videopersonsInfo []protomsg.Videopersons
   videopersonsPersonId := strings.Replace(strings.Trim(fmt.Sprint(id), "[]"), " ", "\",\"", -1)
   var dbinfoRequest = `
            {
               "query": {
                  "bool": {
                     "filter": [{
                        "terms": {
                           "id": [
                              "`+ videopersonsPersonId +`"
                           ]
                        }
                     }]
                  }
               },
                "size":1000000
            }
        `
   buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfoRequest))
   if err != nil {
      return videopersonsInfo,  err
   }

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

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

// 根据底库人员id查询底库人员信息
@@ -116,7 +153,8 @@
                        }
                     }]
                  }
               }
               },
                "size":1000000
            }
        `
    buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfoRequest))
@@ -132,6 +170,34 @@
    dbperson := Dbpersonbyid(sources)
    println(dbperson)
    return dbperson,nil
}
//根据抓拍库人员id查询特征值
func GetVideoPersonFaceFeatureById (id string, indexName string, serverIp string, serverPort string) (string, error) {
    var jsonDSL = `
            {
               "query": {
                  "bool": {
                     "filter": [{
                        "term": {
                           "id":"`+ id +`"
                        }
                     }]
                  }
               },
                "_source":["faceFeature"]
            }
        `
    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
}

// 根据tableid 查询tablename
@@ -150,7 +216,8 @@
        },
        "_source": [
            "tableName"
        ]
        ],
        "size":1000000
    }
    `
    buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(dbinfotable))
@@ -161,7 +228,6 @@
    if err != nil {
          return "",err 
    }

    for _, source := range sources {
        if name, ok := source["tableName"].(string); ok {
            tablename = name
@@ -173,7 +239,7 @@


//根据抓拍人员id更新（videourl）摄像机地址
func updateVideourlById(id string, videoUrl string, indexName string, serverIp string, serverPort string)(statu int){
func UpdateVideourlById(id string, videoUrl string, indexName string, serverIp string, serverPort string)(statu int){
    var info interface{}
    var videoUrlInfo = `
        {
@@ -184,10 +250,11 @@
            "term": {
              "id": "` +id+ `"
            }
          }
          },
            "size":1000000
        }
        `
    buf, err := EsReq("POST", "http://"+serverIp+":"+serverPort+"/"+indexName+"/_search",[]byte(videoUrlInfo))
    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
@@ -201,7 +268,7 @@
        return
    }

    middle, ok := out["updated"].(int)
    middle, ok := out["updated"].(float64)
    if !ok {
        fmt.Println("first updated change error!")
        statu = 500
@@ -216,4 +283,478 @@
        return
    }
    return statu
}
}

//获取当前节点抓拍库所有人员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 + "\"]}},"
   }

   //判断库表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底层机制处理分页

   analyServerFilterStr := ""
   analyServerId := compareArgs.AnalyServerId
   if analyServerId != "" {
      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)
         }


         fmt.Println(len(capturetable))

      }()
   }
   wg.Wait()

   //fmt.Println("lenth_all: ", len(capturetable))
   //fmt.Println("耗时：", 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   ,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
}

//聚合任务列表，taskId+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

}