add data analysis util

sunty

2019-09-27 10758192becb761b34656786ce888a339efd8433

add data analysis util

 EsApi.go

@@ -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,196 @@
    }
    return tabsource
}

//初始化实时抓拍
func InitRealTimeCapture(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 {
        queryStr = `"query":{
        "match_all":{}
    },`
    } else {
        queryStr = `"query":{
        "bool":{
            "filter":[
                {
                    "term":{
                        "isAlarm":1
                    }
                }
            ]
        }
    },`
    }
    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"]
    }`

    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) (total int, err error){
    url := "http://" + serverIp + ":" + serverPort +
        "/" + indexName + "/_search"
    DSLJson := `{
    "size":0,
    "query":{
        "bool":{
            "filter":[{
                "range":{
                    "picDate":{
                        "gte":"now+8H/d"
                        }
                    }
                }]
            }
        }
    }`
    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) (sources map[string]int,err error){
    url := "http://" + serverIp + ":" + serverPort +
        "/videopersons,personaction/_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]int    erface{})
    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!")
    }
    var source = make(map[string]int,0)
    for _, in := range sdkName_status["buckets"].([]interface{}){
        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[sdkName] = count
        //fmt.Println("in",in)
        //sources[in["key"].(string)] = int(in["doc_count"].(float64))
    }
    //fmt.Println("sources",source)
    return source,nil
}