| | |
| | | |
| | | //***********************重庆Start**********************************// |
| | | |
| | | type activeHourFormat struct { |
| | | startTime string |
| | | endTime string |
| | | startHour int |
| | | endHour int |
| | | } |
| | | |
| | | //按需求(activeHourFormat结构体)格式化时间数据 |
| | | func formatActiveHour(activeHour string) (activeHourFormat, error) { |
| | | hours := strings.Split(activeHour, "-") |
| | | |
| | | if len(hours) == 2 { |
| | | startHour := hours[0] |
| | | endHour := hours[1] |
| | | |
| | | // 解析开始时间的小时和分钟 |
| | | startParts := strings.Split(startHour, ":") |
| | | startHourInt, _ := strconv.Atoi(startParts[0]) |
| | | |
| | | // 解析结束时间的小时和分钟 |
| | | endParts := strings.Split(endHour, ":") |
| | | endHourInt, _ := strconv.Atoi(endParts[0]) |
| | | |
| | | // 输出开始时间的小时 |
| | | fmt.Println("开始时间的小时:", startHourInt) |
| | | |
| | | // 输出结束时间的小时 + 1 |
| | | endHourPlusOne := (endHourInt + 1) % 24 // 取余确保不超过24小时 |
| | | fmt.Println("结束时间的小时 + 1:", endHourPlusOne) |
| | | activeHourFormat := activeHourFormat{startTime: startHour, endTime: endHour, startHour: startHourInt, endHour: endHourPlusOne} |
| | | return activeHourFormat, nil |
| | | } |
| | | return activeHourFormat{}, errors.New("错误:无法解析开始时间和结束时间") |
| | | |
| | | } |
| | | |
| | | //判断时间是否再范围之内 |
| | | func isTimeInRange(timeStr, startStr, endStr string) bool { |
| | | layout := "15:04:05" |
| | |
| | | |
| | | if startTime.After(endTime) { |
| | | // 跨越日期的情况 |
| | | //fmt.Println("跨日期",timeStamp, timeStamp.After(startTime), timeStamp.Before(endTime)) |
| | | return timeStamp.After(startTime) || timeStamp.Before(endTime) |
| | | } else { |
| | | // 不跨越日期的情况 |
| | | //fmt.Println("不跨日期",timeStamp, timeStamp.After(startTime), timeStamp.Before(endTime)) |
| | | return timeStamp.After(startTime) && timeStamp.Before(endTime) |
| | | } |
| | | } |
| | |
| | | // formattedTime := parsedTime.Format("15:04:05") |
| | | // return formattedTime, nil |
| | | //} |
| | | func resetDataId(dataId []string, id, dDate, dTime string, sDate *string, sTime *string) []string { |
| | | dataId = make([]string, 0) |
| | | *sDate = dDate |
| | | *sTime = dTime |
| | | dataId = append(dataId, id) |
| | | return dataId |
| | | } |
| | | |
| | | func decodeActivityId(aHFormat activeHourFormat, frequency int, intervalInMinutes int, source []map[string]interface{}) ([]map[string]interface{}, error) { |
| | | func decodeActivityId(frequency int, intervalInMinutes int, source []map[string]interface{}) ([]map[string]interface{}, error) { |
| | | docInfo := make([]map[string]interface{}, 0) |
| | | for _, info := range source { |
| | | tmpInfo := make(map[string]interface{}) |
| | | activeId := make([]string, 0) |
| | | sDate := "" |
| | | sTime := "" |
| | | documentNumber := info["key"].(string) |
| | | tmpInfo["documentNumber"] = documentNumber |
| | | //fmt.Println("documentNumber: ", documentNumber) |
| | |
| | | hits := topHits["hits"].(map[string]interface{}) |
| | | hitsResult := hits["hits"].([]interface{}) |
| | | dataId := make([]string, 0) |
| | | for sIndex, sourceInfo := range hitsResult { |
| | | picUrl := "" |
| | | if hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["baseInfo"] != nil { |
| | | baseInfo := hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["baseInfo"] |
| | | if v, ok := baseInfo.([]interface{}); ok { |
| | | picUrl = v[0].(map[string]interface{})["targetPicUrl"].(string) |
| | | } else if v, ok := baseInfo.(map[string]interface{}); ok { |
| | | picUrl = v["targetPicUrl"].(string) |
| | | } |
| | | } else { |
| | | if hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["targetInfo"] != nil { |
| | | picUrl = hitsResult[0].(map[string]interface{})["_source"].(map[string]interface{})["targetInfo"].([]interface{})[0].(map[string]interface{})["picSmUrl"].(string) |
| | | } |
| | | } |
| | | tmpInfo["picUrl"] = picUrl |
| | | lastCaptureTime := "" |
| | | for _, sourceInfo := range hitsResult { |
| | | rSourceInfo := sourceInfo.(map[string]interface{}) |
| | | source := rSourceInfo["_source"].(map[string]interface{}) |
| | | captureTime := source["picDate"].(string) |
| | | dDate := strings.Split(captureTime, " ")[0] |
| | | dTime := strings.Split(captureTime[:19], " ")[1] |
| | | //fmt.Println(captureTime, dDate, dTime) |
| | | id := source["id"].(string) |
| | | //fmt.Println("sindex: ", sIndex, "documentNumber", tmpInfo["documentNumber"], "id: ", id, "captureTime: ", captureTime) |
| | | if !isTimeInRange(dTime, aHFormat.startTime, aHFormat.endTime) { |
| | | if sDate != "" && len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | dataId = resetDataId(dataId, id, dDate, dTime, &sDate, &sTime) |
| | | } |
| | | continue |
| | | } |
| | | if sDate == "" { |
| | | sDate = dDate |
| | | sTime = dTime |
| | | dataId = append(dataId, id) |
| | | if len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | dataId = resetDataId(dataId, id, dDate, dTime, &sDate, &sTime) |
| | | } |
| | | continue |
| | | } |
| | | if checkTimeDifference(sDate+" "+sTime, captureTime[:19], intervalInMinutes) { |
| | | if len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | dataId = resetDataId(dataId, id, dDate, dTime, &sDate, &sTime) |
| | | } |
| | | continue |
| | | } |
| | | //fmt.Println(daysBetweenDates(sDate, dDate)) |
| | | if aHFormat.startHour < aHFormat.endHour && daysBetweenDates(sDate, dDate) == 0 { |
| | | dataId = append(dataId, id) |
| | | } else if aHFormat.startHour > aHFormat.endHour { |
| | | if daysBetweenDates(sDate, dDate) == 0 { |
| | | if compareTimes(dTime, aHFormat.startTime) == compareTimes(sTime, aHFormat.startTime) { |
| | | // ||compareTimes(dTime,aHFormat.endTime) == compareTimes(sTime, aHFormat.endTime){ |
| | | dataId = append(dataId, id) |
| | | } |
| | | } else if daysBetweenDates(sDate, dDate) == 1 { |
| | | //初始时间戳在结束范围之前 |
| | | if compareTimes(sTime, aHFormat.endTime) == -1 { |
| | | if len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | dataId = resetDataId(dataId, id, dDate, dTime, &sDate, &sTime) |
| | | } |
| | | //初始时间戳在开始范围之后 |
| | | } else if compareTimes(sTime, aHFormat.endTime) == 1 { |
| | | //next时间戳在结束范围之前 |
| | | if compareTimes(dTime, aHFormat.endTime) == -1 { |
| | | dataId = append(dataId, id) |
| | | //next时间戳在开始范围之后 |
| | | } else if compareTimes(dTime, aHFormat.startTime) == 1 { |
| | | if len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | dataId = resetDataId(dataId, id, dDate, dTime, &sDate, &sTime) |
| | | } |
| | | } |
| | | } |
| | | } else if daysBetweenDates(sDate, dDate) >= 1 { |
| | | //fmt.Println(len(dataId)) |
| | | if len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | dataId = resetDataId(dataId, id, dDate, dTime, &sDate, &sTime) |
| | | } |
| | | if lastCaptureTime != "" { |
| | | if checkTimeDifference(lastCaptureTime[:19], captureTime[:19], intervalInMinutes) { |
| | | //fmt.Println("时间小于连续阈值", lastCaptureTime, captureTime) |
| | | continue |
| | | } |
| | | } |
| | | if sIndex == len(hitsResult)-1 { |
| | | if len(dataId) >= frequency { |
| | | activeId = append(activeId, dataId...) |
| | | } |
| | | } |
| | | lastCaptureTime = captureTime |
| | | dataId = append(dataId, id) |
| | | } |
| | | if len(activeId) > 0 { |
| | | tmpInfo["id"] = activeId |
| | | |
| | | if len(dataId) > frequency { |
| | | tmpInfo["id"] = dataId |
| | | docInfo = append(docInfo, tmpInfo) |
| | | } |
| | | } |
| | | return docInfo, nil |
| | | } |
| | | |
| | | func DayNightActivityQuery(communityId string, documentNumber string,startTime string, endTime string, activeHour string, frequency int, |
| | | func DayNightActivityQuery(comIds []string, cameraIds []string, docNumber string, startDateStr string, endDateStr string, activeHour string, frequency int, |
| | | intervalInMinutes int, indexName string, serverIp string, serverPort string) (map[string]interface{}, error) { |
| | | esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" |
| | | |
| | | aHFormat, err := formatActiveHour(activeHour) |
| | | hours := strings.Split(activeHour, "-") |
| | | startTimeStr := hours[0] |
| | | endTimeStr := hours[1] |
| | | layoutTime := "15:04:05" |
| | | startTime, err := time.Parse(layoutTime, startTimeStr) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | endTime, err := time.Parse(layoutTime, endTimeStr) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | layoutDate := "2006-01-02" |
| | | startDate, err := time.Parse(layoutDate, startDateStr) |
| | | if err != nil { |
| | | fmt.Println("Error parsing start time:", err) |
| | | return nil, err |
| | | } |
| | | |
| | | endDate, err := time.Parse(layoutDate, endDateStr) |
| | | if err != nil { |
| | | fmt.Println("Error parsing end time:", err) |
| | | return nil, err |
| | | } |
| | | rangePicDate := make([]string, 0) |
| | | if startTime.Before(endTime) { |
| | | for !startDate.After(endDate) { |
| | | sDateTime := startDate.Format(layoutDate) + " " + startTimeStr |
| | | eDateTime := startDate.Format(layoutDate) + " " + endTimeStr |
| | | rangePicDate = append(rangePicDate, ` |
| | | { |
| | | "range": { |
| | | "picDate": { |
| | | "gte": "`+sDateTime+`", |
| | | "lt": "`+eDateTime+`" |
| | | } |
| | | } |
| | | }`) |
| | | startDate = startDate.AddDate(0, 0, 1) |
| | | } |
| | | } else if startTime.After(endTime) { |
| | | if startDate.Year() == endDate.Year() && startDate.Month() == endDate.Month() && startDate.Day() == endDate.Day() { |
| | | fmt.Println("跨日期不支持单日查询") |
| | | return nil, errors.New("跨日期不支持单日查询!") |
| | | } |
| | | for !startDate.After(endDate) { |
| | | sDateTime := startDate.Format(layoutDate) + " " + startTimeStr |
| | | startDate = startDate.AddDate(0, 0, 1) |
| | | //fendDate := startDate.AddDate(0, 0, 1) |
| | | if startDate.Year() == endDate.Year() && startDate.Month() == endDate.Month() && startDate.Day() == endDate.Day() { |
| | | eDateTime := startDate.Format(layoutDate) + " " + endTimeStr |
| | | rangePicDate = append(rangePicDate, ` |
| | | { |
| | | "range": { |
| | | "picDate": { |
| | | "gte": "`+sDateTime+`", |
| | | "lt": "`+eDateTime+`" |
| | | } |
| | | } |
| | | }`) |
| | | //fmt.Println("startDate: ", startDate, "/t endDate: ", endDate) |
| | | break |
| | | } |
| | | eDateTime := startDate.Format(layoutDate) + " " + endTimeStr |
| | | rangePicDate = append(rangePicDate, ` |
| | | { |
| | | "range": { |
| | | "picDate": { |
| | | "gte": "`+sDateTime+`", |
| | | "lt": "`+eDateTime+`" |
| | | } |
| | | } |
| | | }`) |
| | | } |
| | | } |
| | | //fmt.Println(rangePicDate) |
| | | //return nil, nil |
| | | filterDocIdAttr := "" |
| | | if documentNumber != ""{ |
| | | filterDocIdAttr = "{\"term\": {\""+documentNumber+"\": \"\"}}," |
| | | if docNumber != "" { |
| | | filterDocIdAttr = "{\"term\": {\"documentNumber\": \"" + docNumber + "\"}}," |
| | | } |
| | | comIdsStr := "" |
| | | if comIds != nil && len(comIds) > 0 { |
| | | esComIds := strings.Replace(strings.Trim(fmt.Sprint(comIds), "[]"), " ", "\",\"", -1) |
| | | comIdsStr = "{\"terms\":{\"communityId\":[\"" + esComIds + "\"]}}," |
| | | } |
| | | cameraIdsStr := "" |
| | | if comIds != nil && len(cameraIds) > 0 { |
| | | esCameraIds := strings.Replace(strings.Trim(fmt.Sprint(cameraIds), "[]"), " ", "\",\"", -1) |
| | | cameraIdsStr = "{\"terms\":{\"cameraId\":[\"" + esCameraIds + "\"]}}," |
| | | } |
| | | |
| | | var result = make(map[string]interface{}) |
| | | docInfos := make([]map[string]interface{}, 0) |
| | | for _, rpd := range rangePicDate { |
| | | //if comIdsStr != "" || filterDocIdAttr != "" || cameraIdsStr != "" { |
| | | // rpd += "," |
| | | //} |
| | | queryDSL := ` |
| | | { |
| | | "size": 0, |
| | | "query": { |
| | | "bool": { |
| | | "filter": [ |
| | | ` + comIdsStr + ` |
| | | ` + cameraIdsStr + ` |
| | | ` + filterDocIdAttr + ` |
| | | ` + rpd + ` |
| | | ], |
| | | "must_not": [ |
| | | { |
| | | "term": { |
| | | "documentNumber": "" |
| | | } |
| | | } |
| | | ] |
| | | } |
| | | }, |
| | | "aggs": { |
| | | "group_by_documentnumber": { |
| | | "terms": { |
| | | "field": "documentNumber", |
| | | "size": 100000 |
| | | }, |
| | | "aggs": { |
| | | "top_hits": { |
| | | "top_hits": { |
| | | "_source": [ |
| | | "id", |
| | | "picDate", |
| | | "baseInfo.targetPicUrl", |
| | | "targetInfo.picSmUrl" |
| | | ], |
| | | "size": 100000, |
| | | "sort": [ |
| | | { |
| | | "picDate": { |
| | | "order": "asc" |
| | | } |
| | | } |
| | | ] |
| | | } |
| | | } |
| | | } |
| | | } |
| | | } |
| | | }` |
| | | fmt.Println(esURL) |
| | | fmt.Println(queryDSL) |
| | | buf, err := EsReq("POST", esURL, []byte(queryDSL)) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | source, err := SourceAggregationList(buf) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | //fmt.Println(source) |
| | | //docResult 本次查询结果 |
| | | docResult, err := decodeActivityId(frequency, intervalInMinutes, source) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | if len(docResult) == 0 { |
| | | continue |
| | | } |
| | | if len(docInfos) == 0 { |
| | | docInfos = append(docInfos, docResult...) |
| | | continue |
| | | } |
| | | for _, dr := range docResult { |
| | | var found bool |
| | | for _, di := range docInfos { |
| | | if dr["documentNumber"].(string) == di["documentNumber"].(string) { |
| | | // 找到了相同的documentNumber,将id合并到di["id"]中 |
| | | ids, ok := di["id"].([]string) |
| | | if !ok { |
| | | // 如果id不是字符串切片类型,则进行初始化 |
| | | ids = []string{} |
| | | } |
| | | ids = append(ids, dr["id"].([]string)...) |
| | | di["id"] = ids |
| | | //docInfos[diIndex] = di // 更新原始docInfos中的值 |
| | | found = true |
| | | break |
| | | } |
| | | } |
| | | if !found { |
| | | // 没有找到相同的documentNumber,将dr添加到docInfos中 |
| | | docInfos = append(docInfos, dr) |
| | | } |
| | | } |
| | | } |
| | | if len(docInfos) == 0 { |
| | | return nil, nil |
| | | } |
| | | DataInfos, err := GetInfosByIds(docInfos[0]["id"].([]string), indexName, serverIp, serverPort) |
| | | result["documentNumbers"] = docInfos |
| | | result["datalist"] = DataInfos |
| | | return result, nil |
| | | } |
| | | |
| | | type acmInfo struct { |
| | | documentNumber string |
| | | camerasInfos []camerasInfo |
| | | } |
| | | |
| | | type camerasInfo struct { |
| | | cameraId string |
| | | captureInfos []captureInfo |
| | | } |
| | | |
| | | type captureInfo struct { |
| | | id string |
| | | picDate string |
| | | } |
| | | |
| | | func addSecondsToTimestamp(timestamp string, seconds int) (string, error) { |
| | | parsedTime, err := time.Parse("2006-01-02 15:04:05", timestamp) |
| | | if err != nil { |
| | | return "", err |
| | | } |
| | | newTime := parsedTime.Add(time.Second * time.Duration(seconds)) |
| | | newTimestamp := newTime.Format("2006-01-02 15:04:05") |
| | | return newTimestamp, nil |
| | | } |
| | | |
| | | func decodeInfo(intervalInMinutes int, source []map[string]interface{}) ([]acmInfo, error) { |
| | | acmInfos := make([]acmInfo, 0) |
| | | for _, info := range source { |
| | | var aInfo acmInfo |
| | | documentNumber := info["key"].(string) |
| | | aInfo.documentNumber = documentNumber |
| | | groupByCameraId := info["group_by_cameraId"].(map[string]interface{}) |
| | | cameraBuckets := groupByCameraId["buckets"].([]interface{}) |
| | | for _, cameraInfo := range cameraBuckets { |
| | | var camsInfo camerasInfo |
| | | cInfo := cameraInfo.(map[string]interface{}) |
| | | cameraId := cInfo["key"].(string) |
| | | camsInfo.cameraId = cameraId |
| | | dataBuckets := cInfo["top_hits"].(map[string]interface{})["hits"].(map[string]interface{})["hits"].([]interface{}) |
| | | markTime := "" |
| | | for _, dataInfo := range dataBuckets { |
| | | var capInfo captureInfo |
| | | dInfo := dataInfo.(map[string]interface{}) |
| | | dSource := dInfo["_source"].(map[string]interface{}) |
| | | id := dSource["id"].(string) |
| | | picDate := dSource["picDate"].(string) |
| | | //addFlag := false |
| | | if markTime == "" { |
| | | //addFlag = true |
| | | markTime = picDate |
| | | } else { |
| | | if checkTimeDifference(markTime, picDate, intervalInMinutes) { |
| | | //fmt.Println(markTime, picDate) |
| | | markTime = picDate |
| | | continue |
| | | } |
| | | markTime = picDate |
| | | } |
| | | capInfo.id = id |
| | | capInfo.picDate = picDate |
| | | camsInfo.captureInfos = append(camsInfo.captureInfos, capInfo) |
| | | } |
| | | aInfo.camerasInfos = append(aInfo.camerasInfos, camsInfo) |
| | | } |
| | | acmInfos = append(acmInfos, aInfo) |
| | | } |
| | | return acmInfos, nil |
| | | } |
| | | |
| | | type addResultIds struct { |
| | | documentNumber string |
| | | unionIds []unionId |
| | | } |
| | | |
| | | type unionId struct { |
| | | baseId string |
| | | targetId string |
| | | } |
| | | |
| | | func addResultInfo(source []map[string]interface{}, targetAddResultIds *[]addResultIds, bId string) { |
| | | found := false |
| | | for _, info := range source { |
| | | documentNumber := info["key"].(string) |
| | | dataBuckets := info["top_hits"].(map[string]interface{})["hits"].(map[string]interface{})["hits"].([]interface{}) |
| | | id := dataBuckets[0].(map[string]interface{})["_source"].(map[string]interface{})["id"].(string) |
| | | //fmt.Println("documentNumber: ", documentNumber, "\tid: ", id) |
| | | for i, docInfo := range *targetAddResultIds { |
| | | if documentNumber == docInfo.documentNumber { |
| | | //fmt.Println("新更新") |
| | | (*targetAddResultIds)[i].unionIds = append((*targetAddResultIds)[i].unionIds, unionId{baseId: bId, targetId: id}) |
| | | found = true |
| | | break |
| | | } |
| | | } |
| | | if !found { |
| | | //fmt.Println("新添加") |
| | | var targetAddResultId addResultIds |
| | | targetAddResultId.documentNumber = documentNumber |
| | | targetAddResultId.unionIds = append(targetAddResultId.unionIds, unionId{baseId: bId, targetId: id}) |
| | | *targetAddResultIds = append(*targetAddResultIds, targetAddResultId) |
| | | } |
| | | |
| | | } |
| | | } |
| | | |
| | | func removeDuplicates(nums []string) []string { |
| | | result := make([]string, 0) |
| | | seen := make(map[string]bool) |
| | | |
| | | for _, num := range nums { |
| | | if !seen[num] { |
| | | result = append(result, num) |
| | | seen[num] = true |
| | | } |
| | | } |
| | | |
| | | return result |
| | | } |
| | | |
| | | func findAnalyzeCoordinatedMovementsInfos(infos []acmInfo, docNumber string, beforeTime int, afterTime int, frequency int, |
| | | indexName string, serverIp string, serverPort string) (map[string]interface{}, error) { |
| | | //baseAddResultIds := make([]addResultIds, 0) |
| | | targetAddResultIds := make([]addResultIds, 0) |
| | | esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" |
| | | for _, info := range infos { |
| | | for _, cInfo := range info.camerasInfos { |
| | | for _, pInfo := range cInfo.captureInfos { |
| | | gteDate, err := addSecondsToTimestamp(pInfo.picDate[:19], beforeTime) |
| | | if err != nil { |
| | | fmt.Println(err) |
| | | } |
| | | lteDate, err := addSecondsToTimestamp(pInfo.picDate[:19], afterTime) |
| | | if err != nil { |
| | | fmt.Println(err) |
| | | } |
| | | queryDSL := ` |
| | | { |
| | | "size": 0, |
| | | "query": { |
| | | "bool": { |
| | | "filter": [ |
| | | { |
| | | "range": { |
| | | "picDate": { |
| | | "gte": "` + gteDate + `", |
| | | "lte": "` + lteDate + `" |
| | | } |
| | | } |
| | | }, |
| | | { |
| | | "term": { |
| | | "cameraId": "` + cInfo.cameraId + `" |
| | | } |
| | | } |
| | | ], |
| | | "must_not": [ |
| | | { |
| | | "term": { |
| | | "documentNumber": "` + docNumber + `" |
| | | } |
| | | } |
| | | ] |
| | | } |
| | | }, |
| | | "aggs": { |
| | | "group_by_documentnumber": { |
| | | "terms": { |
| | | "field": "documentNumber", |
| | | "size": 100000 |
| | | }, |
| | | "aggs": { |
| | | "top_hits": { |
| | | "top_hits": { |
| | | "_source": [ |
| | | "id", |
| | | "cameraId", |
| | | "picDate" |
| | | ], |
| | | "size": 10000, |
| | | "sort": [ |
| | | { |
| | | "picDate": { |
| | | "order": "asc" |
| | | } |
| | | } |
| | | ] |
| | | } |
| | | } |
| | | } |
| | | } |
| | | } |
| | | }` |
| | | //fmt.Println(esURL, queryDSL) |
| | | buf, err := EsReq("POST", esURL, []byte(queryDSL)) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | source, err := SourceAggregationList(buf) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | //fmt.Println("pInfo.id: ", pInfo.id) |
| | | addResultInfo(source, &targetAddResultIds, pInfo.id) |
| | | //fmt.Println("targetAddResultIds: ", targetAddResultIds) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | //fmt.Println(source) |
| | | } |
| | | } |
| | | } |
| | | //fmt.Println("targetAddResultIds: ", targetAddResultIds) |
| | | baseIds := make([]string, 0) |
| | | targetIds := make([]string, 0) |
| | | for _, tAIdInfo := range targetAddResultIds { |
| | | if len(tAIdInfo.unionIds) >= frequency { |
| | | for _, unionId := range tAIdInfo.unionIds { |
| | | baseIds = append(baseIds, unionId.baseId) |
| | | targetIds = append(targetIds, unionId.targetId) |
| | | } |
| | | } |
| | | } |
| | | |
| | | rdbaseIds := removeDuplicates(baseIds) |
| | | rdtargetIds := removeDuplicates(targetIds) |
| | | baseInfos, err := GetInfosByIds(rdbaseIds, indexName, serverIp, serverPort) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | targetInfos, err := GetInfosByIds(rdtargetIds, indexName, serverIp, serverPort) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | docNumberMap := make(map[string][]interface{}) |
| | | for _, tinfo := range targetInfos { |
| | | docNumber := tinfo["documentNumber"].(string) |
| | | docNumberMap[docNumber] = append(docNumberMap[docNumber], tinfo) |
| | | } |
| | | targetRecordInfos := make([]map[string]interface{}, 0) |
| | | for docNumber, infos := range docNumberMap { |
| | | ifs := make(map[string]interface{}) |
| | | ifs["documentNumber"] = docNumber |
| | | ifs["recordInfos"] = infos |
| | | targetRecordInfos = append(targetRecordInfos, ifs) |
| | | } |
| | | result := make(map[string]interface{}) |
| | | result["baseRecordInfo"] = baseInfos |
| | | result["targetRecordInfo"] = targetRecordInfos |
| | | return result, nil |
| | | } |
| | | |
| | | func AnalyzeCoordinatedMovements(comIds []string, docNumber string, startDate string, endDate string, beforeTime int, afterTime int, frequency int, |
| | | intervalInMinutes int, indexName string, serverIp string, serverPort string) (map[string]interface{}, error) { |
| | | esURL := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" |
| | | //判断社区IDs |
| | | comIdsStr := "" |
| | | if comIds == nil || len(comIds) > 0 { |
| | | esComIds := strings.Replace(strings.Trim(fmt.Sprint(comIds), "[]"), " ", "\",\"", -1) |
| | | comIdsStr = "{\"terms\":{\"communityId\":[\"" + esComIds + "\"]}}," |
| | | } |
| | | queryDSL := ` |
| | | { |
| | |
| | | { |
| | | "range": { |
| | | "picDate": { |
| | | "gte": "` + startTime + `", |
| | | "lt": "` + endTime + `" |
| | | "gte": "` + startDate + `", |
| | | "lte": "` + endDate + `" |
| | | } |
| | | } |
| | | }, |
| | | `+filterDocIdAttr+` |
| | | ` + comIdsStr + ` |
| | | { |
| | | "term": { |
| | | "communityId": "` + communityId + `" |
| | | } |
| | | }, |
| | | { |
| | | "script": { |
| | | "script": { |
| | | "source": "doc['picDate'].value.hourOfDay >= ` + strconv.Itoa(aHFormat.startHour) + ` || doc['picDate'].value.hourOfDay < ` + strconv.Itoa(aHFormat.endHour) + `", |
| | | "lang": "painless" |
| | | } |
| | | } |
| | | } |
| | | ], |
| | | "must_not": [ |
| | | { |
| | | "term": { |
| | | "documentNumber": "" |
| | | "documentNumber": "` + docNumber + `" |
| | | } |
| | | } |
| | | ] |
| | |
| | | "field": "documentNumber", |
| | | "size": 100000 |
| | | }, |
| | | "aggs": { |
| | | "top_hits": { |
| | | "top_hits": { |
| | | "_source": [ |
| | | "id", |
| | | "picDate" |
| | | ], |
| | | "size": 100000, |
| | | "sort": [ |
| | | { |
| | | "picDate": { |
| | | "order": "asc" |
| | | } |
| | | "aggs": { |
| | | "group_by_cameraId": { |
| | | "terms": { |
| | | "field": "cameraId", |
| | | "size": 10000 |
| | | }, |
| | | "aggs": { |
| | | "top_hits": { |
| | | "top_hits": { |
| | | "_source": [ |
| | | "id", |
| | | "cameraId", |
| | | "picDate" |
| | | ], |
| | | "size": 10000, |
| | | "sort": [ |
| | | { |
| | | "picDate": { |
| | | "order": "asc" |
| | | } |
| | | } |
| | | ] |
| | | } |
| | | ] |
| | | } |
| | | } |
| | | } |
| | | } |
| | | } |
| | | } |
| | | } |
| | | } |
| | | }` |
| | | //fmt.Println(esURL) |
| | | //fmt.Println(queryDSL) |
| | | var result = make(map[string]interface{}) |
| | | buf, err := EsReq("POST", esURL, []byte(queryDSL)) |
| | | if err != nil { |
| | | return nil, err |
| | |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | //fmt.Println(source) |
| | | docResult, err := decodeActivityId(aHFormat, frequency, intervalInMinutes, source) |
| | | docResult, err := decodeInfo(intervalInMinutes, source) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | //result, _ := decodeDocumentInfos(source) |
| | | //return result, nil |
| | | if len(docResult) ==0 { |
| | | return result,nil |
| | | //fmt.Println(docResult) |
| | | result, err := findAnalyzeCoordinatedMovementsInfos(docResult, docNumber, beforeTime, afterTime, frequency, indexName, serverIp, serverPort) |
| | | if err != nil { |
| | | return nil, err |
| | | } |
| | | DataInfos, err := GetInfosByIds(docResult[0]["id"].([]string), indexName, serverIp, serverPort) |
| | | result["documentNumbers"] = docResult |
| | | result["datalist"] = DataInfos |
| | | //fmt.Println(result) |
| | | return result, nil |
| | | } |
| | | |
| | |
| | | } |
| | | |
| | | // 按日期范围,服务器Id删除数据 |
| | | func DeleteByDocumentNumber(docNumber []string, serverIp string, serverPort string, indexName string) (total int, err error) { |
| | | |
| | | url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_delete_by_query" |
| | | docNumbers := strings.Replace(strings.Trim(fmt.Sprint(docNumber), "[]"), " ", "\",\"", -1) |
| | | deleteJson := `{ |
| | | "query":{ |
| | | "bool":{ |
| | | "filter":[ |
| | | { |
| | | "terms":{ |
| | | "documentNumber":["` + docNumbers + `"] |
| | | } |
| | | } |
| | | ] |
| | | } |
| | | } |
| | | } ` |
| | | //fmt.Println(url) |
| | | //fmt.Println(deleteJson) |
| | | //return |
| | | buf, err := EsReq("POST", url, []byte(deleteJson)) |
| | | if err != nil { |
| | | return -1, errors.New("请求失败") |
| | | } |
| | | deleteRes, err := SourceDeleted(buf) |
| | | if err != nil { |
| | | return -1, errors.New("解码失败") |
| | | } |
| | | return deleteRes, nil |
| | | } |
| | | |
| | | //func GetCaptureDaysByDocumentNumber(docNumber string, comId string, indexName string, serverIp string, serverPort string){ |
| | | // url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_search" |
| | | // queryDSL := `` |
| | | //} |
| | | |
| | | // 按日期范围,服务器Id删除数据 |
| | | func DeleteAnalyServerData(serverIp string, serverPort string, indexName string, startTime string, endTime string, analyServerId string) (total int, err error) { |
| | | url := "http://" + serverIp + ":" + serverPort + "/" + indexName + "/_delete_by_query" |
| | | deleteJson := `{ |