Optimize day and night algorithm, format date and time, batch process by date.

sunty

2024-05-31 f6ca7bb43270474fa876ff6ba62c6b2113b045ad

 EsClient.go

@@ -3,60 +3,17 @@
import (
   "basic.com/pubsub/protomsg.git"
   "bytes"
   "encoding/base64"
   "encoding/json"
   "errors"
   "fmt"
   "io"
   "github.com/spf13/viper"
   "io/ioutil"
   "log"
   "math"
   "net/http"
   "strings"
   "time"
)

func GetEsDataReq(url string, parama string, isSource bool) map[string]interface{} {
   //fmt.Println("es 查询请求路径" + url) //  配置信息 获取
   var dat map[string]interface{}
   req, err := http.NewRequest("POST", url, strings.NewReader(parama))
   req.Header.Add("Content-Type", "application/json")
   timeout := time.Duration(10 * time.Second) //超时时间50ms
   client := &http.Client{Timeout: timeout}
   resp, err := client.Do(req)
   if err != nil {
      fmt.Println(err)
      return dat
   }
   defer resp.Body.Close()
   dec := json.NewDecoder(resp.Body)
   if err := dec.Decode(&dat); err == io.EOF {
      fmt.Println(err.Error())
      return dat
   } else if err != nil {
      fmt.Println(err.Error())
      return dat
   }
   // 是否需要 解析 es 返回的 source
   if isSource {
      dat = dat["hits"].(map[string]interface{})
      var data = make(map[string]interface{}, 2)
      data["total"] = dat["total"]
      sources := []interface{}{}
      for _, value := range dat["hits"].([]interface{}) {
         source := value.(map[string]interface{})["_source"].(map[string]interface{})
         //source["id"] = source["id"]
         /*sdkType := source["sdkType"]
         if sdkType != nil {
            sdk, _ := strconv.Atoi(sdkType.(string))
            source["sdkType"] = sdkTypeToValue(sdk)
         }*/
         sources = append(sources, source)
      }
      data["datalist"] = sources
      return data
   } else {
      return dat
   }
}

func Parsesources(sources []map[string]interface{}) (multiInfos []*protomsg.MultiFeaCache) {
   var ok bool
@@ -146,7 +103,7 @@
   return
}

//解析抓拍库人员结构
// 解析抓拍库人员结构
func AIOceanAnalysis(sources []map[string]interface{}) (tmpinfos []protomsg.AIOcean) {
   var ok bool
   for _, source := range sources {
@@ -353,7 +310,7 @@
   return tmpinfos
}

//解析底库人员结构
// 解析底库人员结构
func Dbpersonbyid(sources []map[string]interface{}) (tmpinfos []protomsg.Dbperson) {
   var ok bool

@@ -421,7 +378,7 @@
   return tmpinfos
}

//解析底库结构
// 解析底库结构
func Dbtablebyid(sources []map[string]interface{}) (tmpinfos []protomsg.Dbtable) {
   var ok bool

@@ -476,7 +433,177 @@
   return tmpinfos
}

func SourceAggregations(buf [] byte, thresholdTime float64, queryUseTime float64) (sources map[string]interface{}, err error) {
func FaceSourceAggregations(buf []byte, thresholdTime int, thresholdStayTime int) (sources []map[string]interface{}, err error) {
   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 {
      hitsSources := make([]map[string]interface{}, 0)
      topAttentionHits := inf.(map[string]interface{})["top_attention_hits"].(map[string]interface{})
      middleHits := topAttentionHits["hits"].(map[string]interface{})
      finalHits := middleHits["hits"].([]interface{})
      startTime := ""
      indexLength := len(finalHits)
      point := 0
      for _, in := range finalHits {
         point = point + 1
         tmpHitSource := make(map[string]interface{})
         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)
         if len(tmpTime) > 19 {
            tmpTime = tmpTime[:19]
         }
         mTime, err := time.ParseInLocation("2006-01-02 15:04:05", tmpTime, loc)
         if err != nil {
            return nil, errors.New("时间解析错误")
         }

         sTime := tmpTime
         eTime := mTime.Add(time.Second * 1).Format("2006-01-02 15:04:05")
         stayTime := 1.0
         if startTime != "" && point <= indexLength {
            sinTime, _ := time.ParseInLocation("2006-01-02 15:04:05", startTime, loc)
            passTime := math.Abs(mTime.Sub(sinTime).Seconds())
            hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
            //fmt.Println("passTime:   ", passTime)
            if int(passTime) <= thresholdTime {
               if point == indexLength {
                  hitStartTime := hitsSources[len(hitsSources)-1]["startTime"].(string)
                  realStartTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitStartTime, loc)
                  stayTime = math.Abs(mTime.Sub(realStartTime).Seconds())
                  hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                  startTime = ""
               } else {
                  startTime = tmpTime
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               }
               continue
            } else {
               hitStartTime := hitsSources[len(hitsSources)-1]["startTime"].(string)
               hitEndTime := hitsSources[len(hitsSources)-1]["endTime"].(string)
               realStartTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitStartTime, loc)
               realEndTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitEndTime, loc)
               stayTime = math.Abs(realEndTime.Sub(realStartTime).Seconds())
               if sinTime.Sub(mTime).Seconds() == 0 {
                  sinTime.Add(time.Second * 1)
                  sinTime.Format("2006-01-02 15:04:05")
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                  stayTime = 1
               } else if stayTime == 0 {
                  stayTime = 1
                  hitsSources[len(hitsSources)-1]["endTime"] = realEndTime.Add(time.Second * 1).Format("2006-01-02 15:04:05")
               }
               hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
               if point == indexLength {
                  stayTime = 1
               }
               startTime = ""
            }
         }
         //fmt.Println("========================================================")
         startTime = tmpTime
         tmpHitSource["faceId"] = baseInfo["targetId"].(string)
         if targetInfo["areaId"] == nil {
            continue
         }
         tmpHitSource["areaId"] = targetInfo["areaId"].(string)
         tmpHitSource["startTime"] = sTime
         tmpHitSource["faceImg"] = targetInfo["picSmUrl"].(string)
         tmpHitSource["endTime"] = eTime
         tmpHitSource["stayTime"] = stayTime
         hitsSources = append(hitsSources, tmpHitSource)
      }
      allSource = append(allSource, hitsSources...)
   }
   return allSource, nil
}

func SourceDeduplication(buf []byte) ([]map[string]interface{}, error) {
   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
   }
   faceId := make([]map[string]interface{}, 0)
   for _, in := range buckets {
      tmpInfo := make(map[string]interface{})
      topAttentionHits := in.(map[string]interface{})["top_attention_hits"].(map[string]interface{})
      middleHits := topAttentionHits["hits"].(map[string]interface{})
      finalHits := middleHits["hits"].([]interface{})
      tmpInfo["faceId"] = in.(map[string]interface{})["key"].(map[string]interface{})["faceId"].(string)
      tmpInfo["lastTime"] = finalHits[0].(map[string]interface{})["_source"].(map[string]interface{})["picDate"].(string)
      faceId = append(faceId, tmpInfo)
   }
   return faceId, nil
}

// 解析聚合计数结构
func SourceStatistics(buf []byte) ([]map[string]interface{}, error) {
   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
   }
   resultData := make([]map[string]interface{}, 0)
   for _, pick := range buckets {
      data := make(map[string]interface{}, 0)
      data["areaId"] = pick.(map[string]interface{})["key"].(map[string]interface{})["areaId"].(string)
      data["peopleNum"] = int(pick.(map[string]interface{})["doc_count"].(float64))
      resultData = append(resultData, data)
   }
   return resultData, nil
}

func SourceAggregations(buf []byte, thresholdTime float64, queryUseTime float64) (sources map[string]interface{}, err error) {
   s := make(map[string]interface{})
   loc, err := time.LoadLocation("Asia/Shanghai")
   if err != nil {
@@ -507,7 +634,10 @@
      middleHits := topAttentionHits["hits"].(map[string]interface{})
      finalHits := middleHits["hits"].([]interface{})
      startTime := ""
      indexLength := len(finalHits)
      point := 0
      for _, in := range finalHits {
         point = point + 1
         tmpHitSource := make(map[string]interface{})
         tmpbuf, ok := in.(map[string]interface{})
         if !ok {
@@ -522,40 +652,55 @@
         baseInfo := source["baseInfo"].([]interface{})[0].(map[string]interface{})
         targetInfo := source["targetInfo"].([]interface{})[0].(map[string]interface{})
         tmpTime := source["picDate"].(string)
         if len(tmpTime) > 19 {
            tmpTime = tmpTime[:19]
         }
         mTime, err := time.ParseInLocation("2006-01-02 15:04:05", tmpTime, loc)
         if err != nil {
            return nil, errors.New("时间解析错误")
         }

         sTime := tmpTime
         eTime := tmpTime
         stayTime := 0.0
         if startTime != "" {
         eTime := mTime.Add(time.Second * 1).Format("2006-01-02 15:04:05")
         stayTime := 1.0
         if startTime != "" && point <= indexLength {
            sinTime, _ := time.ParseInLocation("2006-01-02 15:04:05", startTime, loc)
            passTime := math.Abs(mTime.Sub(sinTime).Seconds())
            hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
            //fmt.Println("passTime:   ", passTime)
            if passTime <= thresholdTime {
               startTime = tmpTime
               hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               if point == indexLength {
                  hitStartTime := hitsSources[len(hitsSources)-1]["startTime"].(string)
                  realStartTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitStartTime, loc)
                  stayTime = math.Abs(mTime.Sub(realStartTime).Seconds())
                  hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                  startTime = ""
               } else {
                  startTime = tmpTime
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               }
               continue
            } else {
               hitStartTime := hitsSources[len(hitsSources)-1]["startTime"].(string)
               hitEndTime := hitsSources[len(hitsSources)-1]["endTime"].(string)
               realStartTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitStartTime, loc)
               //fmt.Println("hitStartTime:    ",hitStartTime)
               //fmt.Println("realStartTime:    ",realStartTime)
               //fmt.Println("mTime:    ",mTime)
               stayTime = math.Abs(mTime.Sub(realStartTime).Seconds())
               //fmt.Println("stayTime:    ",stayTime)
               hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               //fmt.Println("stayTime",stayTime)
               realEndTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitEndTime, loc)
               stayTime = math.Abs(realEndTime.Sub(realStartTime).Seconds())
               if sinTime.Sub(mTime).Seconds() == 0 {
                  sinTime.Add(time.Second * 1)
                  sinTime.Format("2006-01-02 15:04:05")
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                  stayTime = stayTime + 1
                  stayTime = 1
               } else if stayTime == 0 {
                  stayTime = 1
                  hitsSources[len(hitsSources)-1]["endTime"] = realEndTime.Add(time.Second * 1).Format("2006-01-02 15:04:05")
               }
               hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
               //fmt.Println(hitsSources[len(hitsSources)-1])
               if point == indexLength {
                  stayTime = 1
               }
               startTime = ""
            }
         }
         //fmt.Println("========================================================")
@@ -584,10 +729,6 @@
         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)
@@ -597,7 +738,7 @@
   return s, nil
}

func SourceAggregationsReturnByGrouped(buf [] byte, thresholdTime float64) (sources []map[string]interface{}, err error) {
func SourceAggregationsReturnByGrouped(buf []byte, thresholdTime float64) (sources []map[string]interface{}, err error) {
   loc, err := time.LoadLocation("Asia/Shanghai")
   if err != nil {
      return nil, errors.New("时区设置错误")
@@ -624,8 +765,12 @@
      topAttentionHits := inf.(map[string]interface{})["top_attention_hits"].(map[string]interface{})
      middleHits := topAttentionHits["hits"].(map[string]interface{})
      finalHits := middleHits["hits"].([]interface{})
      indexLength := len(finalHits)
      point := 0
      startTime := ""
      //fmt.Println("finalHits: ",finalHits)
      for _, in := range finalHits {
         point = point + 1
         tmpHitSource := make(map[string]interface{})
         tmpbuf, ok := in.(map[string]interface{})
         if !ok {
@@ -640,40 +785,55 @@
         baseInfo := source["baseInfo"].([]interface{})[0].(map[string]interface{})
         targetInfo := source["targetInfo"].([]interface{})[0].(map[string]interface{})
         tmpTime := source["picDate"].(string)
         if len(tmpTime) > 19 {
            tmpTime = tmpTime[:19]
         }
         mTime, err := time.ParseInLocation("2006-01-02 15:04:05", tmpTime, loc)
         if err != nil {
            return nil, errors.New("时间解析错误")
         }

         sTime := tmpTime
         eTime := tmpTime
         stayTime := 0.0
         if startTime != "" {
         eTime := mTime.Add(time.Second * 1).Format("2006-01-02 15:04:05")
         stayTime := 1.0
         if startTime != "" && point <= indexLength {
            sinTime, _ := time.ParseInLocation("2006-01-02 15:04:05", startTime, loc)
            passTime := math.Abs(mTime.Sub(sinTime).Seconds())
            hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
            //fmt.Println("passTime:   ", passTime)
            if passTime <= thresholdTime {
               startTime = tmpTime
               hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               if point == indexLength {
                  hitStartTime := hitsSources[len(hitsSources)-1]["startTime"].(string)
                  realStartTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitStartTime, loc)
                  stayTime = math.Abs(mTime.Sub(realStartTime).Seconds())
                  hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                  startTime = ""
               } else {
                  startTime = tmpTime
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               }
               continue
            } else {
               hitStartTime := hitsSources[len(hitsSources)-1]["startTime"].(string)
               hitEndTime := hitsSources[len(hitsSources)-1]["endTime"].(string)
               realStartTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitStartTime, loc)
               //fmt.Println("hitStartTime:    ",hitStartTime)
               //fmt.Println("realStartTime:    ",realStartTime)
               //fmt.Println("mTime:    ",mTime)
               stayTime = math.Abs(mTime.Sub(realStartTime).Seconds())
               //fmt.Println("stayTime:    ",stayTime)
               hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
               //fmt.Println("stayTime",stayTime)
               realEndTime, _ := time.ParseInLocation("2006-01-02 15:04:05", hitEndTime, loc)
               stayTime = math.Abs(realEndTime.Sub(realStartTime).Seconds())
               if sinTime.Sub(mTime).Seconds() == 0 {
                  sinTime.Add(time.Second * 1)
                  sinTime.Format("2006-01-02 15:04:05")
                  hitsSources[len(hitsSources)-1]["endTime"] = tmpTime
                  stayTime = stayTime + 1
                  stayTime = 1
               } else if stayTime == 0 {
                  stayTime = 1
                  hitsSources[len(hitsSources)-1]["endTime"] = realEndTime.Add(time.Second * 1).Format("2006-01-02 15:04:05")
               }
               hitsSources[len(hitsSources)-1]["stayTime"] = stayTime
               //fmt.Println(hitsSources[len(hitsSources)-1])
               if point == indexLength {
                  stayTime = 1
               }
               startTime = ""
            }
         }
         startTime = tmpTime
@@ -707,7 +867,7 @@
   return sources, nil
}

//解析抓拍库人员结构
// 解析抓拍库人员结构
func PerSonAnalysis(preData []map[string]interface{}) (sources []map[string]interface{}, err error) {
   loc, err := time.LoadLocation("Asia/Shanghai")
   if err != nil {
@@ -872,9 +1032,44 @@
   if !ok {
      return -1, errors.New("first total change error!")
   }
   tmp := middle["total"].(float64)
   total = int(tmp)

   tmp, b := middle["total"].(map[string]interface{})
   if b != true {
      v := middle["total"].(float64)
      t := int(v)
      return t, nil
   }
   value := tmp["value"].(float64)
   total = int(value)
   return total, nil
}

func SourceAggregationList(buf []byte) (sources []map[string]interface{}, err error) {
   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!")
   }
   //fmt.Println("本次共匹配条数为: ", out["hits"].(map[string]interface{})["total"].(float64))
   documentAggregations := middle["group_by_documentnumber"].(map[string]interface{})
   buckets := documentAggregations["buckets"].([]interface{})
   if len(buckets) == 0 {
      return nil, nil
   }
   for _, in := range buckets {
      tmpbuf, ok := in.(map[string]interface{})
      if !ok {
         return nil, errors.New("")
      }
      sources = append(sources, tmpbuf)
   }
   return sources, nil
}

func EsReq(method string, url string, parama []byte) (buf []byte, err error) {
@@ -885,6 +1080,7 @@
   }
   request, err := http.NewRequest(method, url, bytes.NewBuffer(parama))
   request.Header.Set("Content-type", "application/json")
   request.Header.Set("Authorization", Token)

   if err != nil {
      fmt.Println("build request fail !")
@@ -917,6 +1113,31 @@
// 赋值时检测是否能够赋值
//func //Isnil(key string, ok bool){
//    if !ok {
//            fmt.Println(key, "is nil can not asign") 
//            fmt.Println(key, "is nil can not asign")
//    }
//}

type account struct {
   Username     string `mapstructure: "username"`
   Userpassword string `mapstructure: "userpassword"`
}

var Account = &account{}

var Token string

func init() {
   v := viper.New()
   v.SetConfigType("yaml")
   v.SetConfigName("pro")
   v.AddConfigPath("../config/")
   v.AddConfigPath("./config/")
   v.AddConfigPath("/opt/vasystem/config/")
   err := v.ReadInConfig()
   if err != nil {
      log.Fatal("err on parsing configuration file!", err)
   }
   v.UnmarshalKey("es.account", Account)

   Token = "Basic " + base64.StdEncoding.EncodeToString([]byte(Account.Username+":"+Account.Userpassword))
}