add buckets

sunty

2020-08-20 66d9e889d745b573ad8d67ac39e1d05335424f5c

add buckets

 EsApi.go

@@ -159,6 +159,139 @@

}

//根据时间范围，摄像机列表，分组聚合人脸列表
func GetfaceDataBucketsBycameraIdAndTime(cameraId []string, startTime string, endTime string, serverIp string, ServerPort string, indexName string) (buckersDate map[string]interface{},err error){
    esCameraId := strings.Replace(strings.Trim(fmt.Sprint(cameraId), "[]"), " ", "\",\"", -1)
    var buckersUrl = "http://" + serverIp + ":" + ServerPort + "/" + indexName + "/_search"
    var buckersBody = `{
    "query": {
        "bool": {
            "filter": [
                {
                    "range": {
                        "picDate": {
                            "gte": "` + startTime + `",
                            "lte": "` + endTime + `"
                        }
                    }
                },
                {
                    "term": {
                        "targetInfo.targetType.raw": "FaceDetect"
                    }
                },
                {
                    "terms": {
                        "cameraId": ["` + esCameraId + `"]
                    }
                }
            ]
        }
    },
    "size": 0,
    "aggs": {
        "buckets_aggs": {
            "composite": {
                "sources": [
                    {
                        "baseInfo.targetId": {
                            "terms": {
                                "field": "baseInfo.targetId"
                            }
                        }
                    },
                    {
                        "cameraId": {
                            "terms": {
                                "field": "cameraId"
                            }
                        }
                    }
                ]
            },
            "aggs":{
                "top_attention_hits":{
                    "top_hits":{
                        "size": 100,
                        "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"]
                        }
                    }
                }
            }
        }
    }
}`
    buf, err := EsReq("POST", buckersUrl, []byte(buckersBody))
    if err != nil {
        return nil, err
    }

    sources, err := SourceAggregations(buf)
    if err != nil {
        return nil, err
    }
    //println(sources)
    return sources,nil
}

//根据抓拍人员id更新（picurl）图片地址---预开发
func UpdatePicUrlById(id string, picurl string, indexName string, serverIp string, serverPort string) (statu int, err error) {

    var info interface{}
    url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_update_by_query?refresh=true"
    sourceStr := "ctx._source.videoUrl='" + picurl + "'"
    var videoUrlInfo = `
        {
          "script": {
            "source": "` + sourceStr + `"	
          },
          "query": {
            "term": {
              "id": "` + id + `"
            }
          }
        }
        `
    //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)
    if !ok {
        logPrint("first updated change error!")
        statu = 500
        return statu, errors.New("first updated change error!")
    }
    if middle == 1 {
        statu = 200
        return statu, nil
    }
    if middle == 0 {
        statu = 201
        return statu, errors.New("已经修改")
    }
    return statu, nil
}

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

@@ -770,7 +903,7 @@
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)
    dbinfos := make([]*protomsg.MultiFeaCache, 0)
    //dbinfosss := make([]*protomsg.MultiFeaCache,0)
    //dbinfoss = append(dbinfoss, dbinfosss...)

@@ -819,20 +952,20 @@
            //logPrint("url: ",reqJsonDSL)
            buf, err := EsReq("POST", url, []byte(reqJsonDSL))
            if err != nil {
                logPrint("EsReq: ",err)
                logPrint("EsReq: ", err)
                return
            }

            // 返回 _source 数组
            sources, err := Sourcelistforscroll(buf)
            if err != nil {
                logPrint("EsReq: ",err)
                logPrint("EsReq: ", err)
                return
            }
            // 返回所有查询的数据
            ftmpDatas :=  Parsesources(sources["sourcelist"].([]map[string]interface{}))
            ftmpDatas := Parsesources(sources["sourcelist"].([]map[string]interface{}))
            lock.Lock()
            dbinfos = append(dbinfos,ftmpDatas...)
            dbinfos = append(dbinfos, ftmpDatas...)
            //logPrint("prsLen: ", len(Parsesources(sources["sourcelist"].([]map[string]interface{}))))
            //logPrint("dbinfosLen: ", len(dbinfos))
            lock.Unlock()
@@ -921,14 +1054,13 @@
    if err != nil {
        return false, errors.New("解码失败")
    }
    if resTotal == -1 || resTotal == 0{
    if resTotal == -1 || resTotal == 0 {
        result = false
    } else {
        result = true
    }
    return result, nil
}


//按日期范围，服务器Id删除数据
func DeleteAnalyServerData(serverIp string, serverPort string, indexName string, startTime string, endTime string, analyServerId string) (result bool, err error) {

 EsClient.go

@@ -8,6 +8,7 @@
    "fmt"
    "io"
    "io/ioutil"
    "math"
    "net/http"
    "strings"
    "time"
@@ -475,6 +476,107 @@
    return tmpinfos
}

func SourceAggregations(buf [] byte) (sources map[string]interface{}, err error) {
    s := make(map[string]interface{})
    loc, err := time.LoadLocation("Asia/Shanghai")
    if err != nil {
        return nil, errors.New("时区设置错误")
    }
    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!")
    }
    bucketsAggs := middle["buckets_aggs"].(map[string]interface{})
    buckets := bucketsAggs["buckets"].([]interface{})
    if len(buckets) == 0 {
        return nil, nil
    }
    allSource := make([]map[string]interface{},0)
    for _, inf := range buckets {
        //tmpSources := make(map[string]interface{}, 0)
        hitsSources := make([]map[string]interface{}, 0)
        //groupKey := inf.(map[string]interface{})["key"].(map[string]interface{})
        //docCount := int(inf.(map[string]interface{})["doc_count"].(float64))
        topAttentionHits := inf.(map[string]interface{})["top_attention_hits"].(map[string]interface{})
        middleHits := topAttentionHits["hits"].(map[string]interface{})
        finalHits := middleHits["hits"].([]interface{})
        tmpHitSource := make(map[string]interface{})
        startTime := ""
        for _, in := range finalHits {
            tmpbuf, ok := in.(map[string]interface{})
            if !ok {
                fmt.Println("change to source error!")
                continue
            }
            source, ok := tmpbuf["_source"].(map[string]interface{})
            if !ok {
                fmt.Println("change _source error!")
                continue
            }
            baseInfo := source["baseInfo"].([]interface{})[0].(map[string]interface{})
            targetInfo := source["targetInfo"].([]interface{})[0].(map[string]interface{})
            tmpTime := source["picDate"].(string)
            mTime, err := time.ParseInLocation("2006-01-02 15:04:05", tmpTime, loc)
            if err != nil {
                return nil, errors.New("时间解析错误")
            }

            sTime := tmpTime
            eTime := tmpTime
            if startTime != "" {
                sinTime, _ := time.ParseInLocation("2006-01-02 15:04:05", startTime, loc)
                if math.Abs(sinTime.Sub(mTime).Seconds()) <= 20 {
                    startTime = tmpTime
                    hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                    continue
                } else {
                    if sinTime.Sub(mTime).Seconds() == 0{
                        sinTime.Add(time.Second*1)
                        sinTime.Format("2006-01-02 15:04:05")
                        hitsSources[len(hitsSources)-1]["endTime"] = sinTime
                    }
                }
            }
            startTime = tmpTime
            tmpHitSource["personId"] = baseInfo["targetId"].(string)
            tmpHitSource["cameraId"] = source["cameraId"].(string)
            tmpHitSource["cameraName"] = source["cameraName"].(string)
            tmpHitSource["cameraAddr"] = source["cameraAddr"].(string)
            tmpHitSource["targetScore"] = int(targetInfo["targetScore"].(float64))
            tmpHitSource["properties"] = source["showLabels"].(string)
            tmpHitSource["tableId"] = baseInfo["tableId"].(string)
            tmpHitSource["tableName"] = baseInfo["tableName"].(string)
            tmpHitSource["bwType"] = baseInfo["bwType"].(string)
            tmpHitSource["personName"] = baseInfo["targetName"].(string)
            tmpHitSource["compareScore"] = int(baseInfo["compareScore"].(float64))
            tmpHitSource["startTime"] = sTime
            tmpHitSource["startBackGroundPicUrl"] = source["picMaxUrl"].([]interface{})
            tmpHitSource["startFacePicUrl"] = targetInfo["picSmUrl"].(string)
            tmpHitSource["endTime"] = eTime
            tmpHitSource["endTBackGroundPicUrl"] = source["picMaxUrl"].([]interface{})
            tmpHitSource["endTFacePicUrl"] = targetInfo["picSmUrl"].(string)
            hitsSources = append(hitsSources, tmpHitSource)
        }
        allSource = append(allSource,hitsSources...)
    //    tmpSources["groupKey"] = groupKey
    //    tmpSources["doc_count"] = docCount
    //    tmpSources["hits_sources"] = hitsSources
    //    sources = append(sources, tmpSources)
    }
    count := len(allSource)
    fmt.Println(count)
    s["count"] = count
    s["allSource"] = allSource

    return s, nil
}

func Sourcelist(buf []byte) (sources []map[string]interface{}, err error) {
    var info interface{}
    json.Unmarshal(buf, &info)