添加常住人口和重点人员比对

zhangzengfei

2024-07-07 250f2e68becab56cc009216496dfa0d7d8d1a22f

添加常住人口和重点人员比对

 cache/cache.go

@@ -20,44 +20,49 @@
    PRE_DBTABLE = "dbTable_"
)

type CmapItem struct {
type AreaMapItem struct {
    sync.Mutex
    Area map[string]*shardmap.ShardMap
}

var CacheMap *CmapItem
var CaptureDbMap *AreaMapItem
var RealNameDbMap = shardmap.New(uint8(*threadnum))
var KeyPersonDbMap = shardmap.New(uint8(*threadnum))

var doOnce sync.Once

func ReInitDbTablePersonsCache() {
    CacheMap.Lock()
    defer CacheMap.Unlock()
    if CacheMap == nil {
        CacheMap = &CmapItem{
            Area: make(map[string]*shardmap.ShardMap),
        }
    }
    for tableId, _ := range CacheMap.Area {
        delete(CacheMap.Area, tableId)
    }

    initDbTablePersonsCache()
}

func InitDbTablePersons() {
func InitCache() {
    doOnce.Do(func() {
        flag.Parse()

        CacheMap = &CmapItem{
        CaptureDbMap = &AreaMapItem{
            Area: make(map[string]*shardmap.ShardMap),
        }

        // 初始化未分类, 没有小区id的档案
        CacheMap.Area[Unfiled] = shardmap.New(uint8(*threadnum))
        CaptureDbMap.Area[Unfiled] = shardmap.New(uint8(*threadnum))
    })

    initDbTablePersonsCache()
    initRealNamePersonsCache()
    initKeyPersonsCache()
}

func ReInitDbTablePersonsCache() {
    CaptureDbMap.Lock()
    defer CaptureDbMap.Unlock()

    if CaptureDbMap == nil {
        CaptureDbMap = &AreaMapItem{
            Area: make(map[string]*shardmap.ShardMap),
        }
    }
    for tableId, _ := range CaptureDbMap.Area {
        delete(CaptureDbMap.Area, tableId)
    }

    initDbTablePersonsCache()
}
func initDbTablePersonsCache() {
    // 缓存底库中的全部人员信息
    var dbpApi db.DbPersons
@@ -66,7 +71,7 @@
    var psApi db.PersonStatus
    accessAreas, _ := psApi.GetPersonAccessedAreas()

    logger.Debugf("所有底库共有%d条记录", total)
    logger.Debugf("抓拍档案库共有%d条记录", total)
    if e == nil && total > 0 {
        queryEachNum := *querynum
        qn := int(total) / *threadnum
@@ -91,35 +96,63 @@
                    return
                }
                logger.Debugf(" eachNum:%d, 获取%d条人员信息", queryEachNum, len(dbPersons))
                CacheMap.Lock()
                CaptureDbMap.Lock()

                areaId := ""
                for _, value := range dbPersons {
                    areaId = value.AreaId
                    // 没有小区id的人员
                    if areaId == "" {
                        CacheMap.Area[Unfiled].Set(value.Id, value)
                        CacheMap.Area[Unfiled].Settime()
                        CaptureDbMap.Area[Unfiled].Set(value.Id, value)
                        CaptureDbMap.Area[Unfiled].Settime()
                        continue
                    }

                    for _, areaId := range accessAreas[value.Id] {
                        if _, ok := CacheMap.Area[areaId]; !ok {
                            CacheMap.Area[areaId] = shardmap.New(uint8(*threadnum))
                        if _, ok := CaptureDbMap.Area[areaId]; !ok {
                            CaptureDbMap.Area[areaId] = shardmap.New(uint8(*threadnum))
                        }

                        CacheMap.Area[areaId].Set(value.Id, value)
                        CacheMap.Area[areaId].Settime()
                        CaptureDbMap.Area[areaId].Set(value.Id, value)
                        CaptureDbMap.Area[areaId].Settime()
                    }
                }

                CacheMap.Unlock()
                CaptureDbMap.Unlock()

            }(j)
        }
        wg.Wait()
        logger.Debug("底库人员缓存完成用时:", time.Since(temptime))
        logger.Debug("抓拍档案库人员缓存完成用时:", time.Since(temptime))
    }
}

func initRealNamePersonsCache() {
    var dbApi db.Layouts
    dbPersons, err := dbApi.GetRealNamePersonList()
    if err != nil {
        logger.Error("init real-name persons error,", err.Error())
    }

    for _, value := range dbPersons {
        RealNameDbMap.Set(value.Id, value)
    }

    logger.Debugf("常住人口共有%d条记录", len(dbPersons))
}

func initKeyPersonsCache() {
    var dbApi db.Layouts
    dbPersons, err := dbApi.GetKeyPersonList()
    if err != nil {
        logger.Error("init real-name persons error,", err.Error())
    }

    for _, value := range dbPersons {
        KeyPersonDbMap.Set(value.Id, value)
    }

    logger.Debugf("重点人员共有%d条记录", len(dbPersons))
}

// UpdateDbPersonsCacheById 更新缓存中的全部人员信息
@@ -131,30 +164,27 @@
        return
    }
    if info != nil && info.AreaId != "" {
        CacheMap.Lock()
        defer CacheMap.Unlock()
        if _, ok := CacheMap.Area[info.AreaId]; !ok {
            CacheMap.Area[info.AreaId] = shardmap.New(uint8(*threadnum))
        if _, ok := CaptureDbMap.Area[info.AreaId]; !ok {
            CaptureDbMap.Lock()
            defer CaptureDbMap.Unlock()
            CaptureDbMap.Area[info.AreaId] = shardmap.New(uint8(*threadnum))
        }
        CacheMap.Area[info.AreaId].Set(info.Id, info)
        CacheMap.Area[info.AreaId].Settime()
        CaptureDbMap.Area[info.AreaId].Set(info.Id, info)
        CaptureDbMap.Area[info.AreaId].Settime()
    }
}

func DeleteDbPersonsCacheById(id string) {
    CacheMap.Lock()
    defer CacheMap.Unlock()

    for key, _ := range CacheMap.Area {
        CacheMap.Area[key].Del(id)
    for key, _ := range CaptureDbMap.Area {
        CaptureDbMap.Area[key].Del(id)
    }
}

func RealTimeAddPersonInfoToCache(tableId string, id string, faceFeature string, enable int32, carNo string) {
    CacheMap.Lock()
    defer CacheMap.Unlock()
    if _, ok := CacheMap.Area[tableId]; !ok {
        CacheMap.Area[tableId] = shardmap.New(uint8(*threadnum))
    CaptureDbMap.Lock()
    defer CaptureDbMap.Unlock()
    if _, ok := CaptureDbMap.Area[tableId]; !ok {
        CaptureDbMap.Area[tableId] = shardmap.New(uint8(*threadnum))
    }
    var ei = protomsg.Esinfo{
        Id:          id,
@@ -163,16 +193,16 @@
        Enable:      enable,
        CarNo:       carNo,
    }
    CacheMap.Area[tableId].Set(id, &ei)
    CaptureDbMap.Area[tableId].Set(id, &ei)
    logger.Debug("id:", id, ",tableId:", ",len(faceFeature):", len(faceFeature), ",tableId:", tableId, ",enable:", enable)
}

func RealTimeDelPersonFromCache(tableId string, id string) {
    logger.Debug("DelPersonFromCache,tableId:", tableId, ",id:", id)
    CacheMap.Lock()
    defer CacheMap.Unlock()
    if _, ok := CacheMap.Area[tableId]; ok {
        CacheMap.Area[tableId].Del(id)
    CaptureDbMap.Lock()
    defer CaptureDbMap.Unlock()
    if _, ok := CaptureDbMap.Area[tableId]; ok {
        CaptureDbMap.Area[tableId].Del(id)
        logger.Debug("DelPerson ok success")
    } else {
        logger.Error("tableId:", tableId, " not exist")
@@ -181,27 +211,27 @@

func RealTimeDelTable(tableId string) {
    logger.Debug("RealTimeDelTable tableId:", tableId)
    CacheMap.Lock()
    defer CacheMap.Unlock()
    CaptureDbMap.Lock()
    defer CaptureDbMap.Unlock()

    if dtM, ok := CacheMap.Area[PRE_DBTABLE]; ok {
    if dtM, ok := CaptureDbMap.Area[PRE_DBTABLE]; ok {
        dtM.Del(tableId)
    }
    if _, ok := CacheMap.Area[tableId]; ok {
        delete(CacheMap.Area, tableId)
    if _, ok := CaptureDbMap.Area[tableId]; ok {
        delete(CaptureDbMap.Area, tableId)
    }
}

// 使底库生效,将底库中的所有生效状态的人特征添加到缓存
func RealTimeUpdateTable(tableId string, enable int32) {
    logger.Debug("RealTimeUpdateTable tableId:", tableId, ",enable:", enable)
    CacheMap.Lock()
    defer CacheMap.Unlock()
    CaptureDbMap.Lock()
    defer CaptureDbMap.Unlock()

    if _, ok := CacheMap.Area[PRE_DBTABLE]; !ok {
        CacheMap.Area[PRE_DBTABLE] = shardmap.New(uint8(*threadnum))
    if _, ok := CaptureDbMap.Area[PRE_DBTABLE]; !ok {
        CaptureDbMap.Area[PRE_DBTABLE] = shardmap.New(uint8(*threadnum))
    }
    CacheMap.Area[PRE_DBTABLE].Set(tableId, enable == 1)
    CaptureDbMap.Area[PRE_DBTABLE].Set(tableId, enable == 1)
}

func UpdateCache(changeMsg *protomsg.EsPersonCacheChange) {

 compare/capture.go

New file
@@ -0,0 +1,75 @@
package compare

import (
    "sdkCompare/cache"

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

func capturePersonsCompere(args protomsg.CompareArgs, fFeature []float32, baseScore float32) protomsg.SdkCompareResult {
    var scr protomsg.SdkCompareResult
    var walkedArea = make(map[string]struct{}, 0)

    // 比对传入的小区id
    if args.TreeNodes != nil && len(args.TreeNodes) > 0 {
        for _, id := range args.TreeNodes {
            if _, ok := cache.CaptureDbMap.Area[id]; !ok {
                continue
            }

            targets := cache.CaptureDbMap.Area[id].Walk(DoSdkCompare, fFeature, baseScore)
            if len(targets) > 0 {
                // 比对结果去重, 同一个人到访过多个小区, 缓存数据内会有多条记录
                for idx, t := range targets {
                    var isRepeat bool
                    for _, r := range scr.CompareResult {
                        if t.Id == r.Id {
                            isRepeat = true
                            break
                        }
                    }
                    if !isRepeat {
                        scr.CompareResult = append(scr.CompareResult, targets[idx])
                    }
                }
            }

            walkedArea[id] = struct{}{}
        }

        if len(scr.CompareResult) > 0 || !args.IsCompareAll {
            return scr
        }
    }

    // 比对全部小区
    if !args.IsCompareAll && len(args.TreeNodes) > 0 {
        baseScore += 20
    }

    for key, val := range cache.CaptureDbMap.Area {
        if _, ok := walkedArea[key]; ok {
            continue
        }

        targets := val.Walk(DoSdkCompare, fFeature, baseScore)
        if len(targets) > 0 {
            // 比对结果去重, 同一个人到访过多个小区, 缓存数据内会有多条记录
            for idx, t := range targets {
                var isRepeat bool
                for _, r := range scr.CompareResult {
                    if t.Id == r.Id {
                        isRepeat = true
                        break
                    }
                }
                if !isRepeat {
                    scr.CompareResult = append(scr.CompareResult, targets[idx])
                }
            }
        }
        // todo 添加小区外的关联关系, 下次优先比对
    }

    return scr
}

 compare/compare.go

@@ -2,10 +2,9 @@

import (
    "fmt"
    "sdkCompare/util"
    "strconv"

    "sdkCompare/cache"
    "sdkCompare/util"

    "basic.com/pubsub/protomsg.git"
    "basic.com/valib/logger.git"
@@ -14,12 +13,12 @@

const thresholdLimit = float32(30)

func GetComparePersonBaseInfo(args protomsg.CompareArgs) []byte {
func Walk(args protomsg.CompareArgs) []byte {
    if args.FaceFeature == nil {
        return nil
    }

    floatFeat := util.ByteSlice2float32Slice(args.FaceFeature)
    fFeature := util.ByteSlice2float32Slice(args.FaceFeature)

    //指定最低分
    baseScore := thresholdLimit
@@ -31,79 +30,27 @@
        baseScore = 0
    }

    var scResult protomsg.SdkCompareResult
    var response protomsg.SdkCompareResult

    var walkedArea = make(map[string]struct{}, 0)
    // 优先比对传入的小区id
    if args.TreeNodes != nil && len(args.TreeNodes) > 0 {
        for _, id := range args.TreeNodes {
            if _, ok := cache.CacheMap.Area[id]; ok {
                targets := cache.CacheMap.Area[id].Walk(DoSdkCompare, floatFeat, baseScore)
                if len(targets) > 0 {
                    // 比对结果去重, 同一个人到访过多个小区, 缓存数据内会有多条记录
                    for idx, t := range targets {
                        var isRepeat bool
                        for _, r := range scResult.CompareResult {
                            if t.Id == r.Id {
                                isRepeat = true
                                break
                            }
                        }
                        if !isRepeat {
                            scResult.CompareResult = append(scResult.CompareResult, targets[idx])
                        }
                    }
    // 抓拍库比对
    switch args.CompareTarget {
    case "realName":
        response = realNamePersonsCompere(args, fFeature, baseScore)
    case "keyPerson":
        response = keyPersonsCompere(args, fFeature, baseScore)
    default:
        response = capturePersonsCompere(args, fFeature, baseScore)
                }

                walkedArea[id] = struct{}{}
            }
    logger.Debugf("Compare Target %s result len %d", args.CompareTarget, len(response.CompareResult))

    if len(response.CompareResult) > 0 {
        logger.Debugf("Compare result %+v", response.CompareResult)
        }

        if len(scResult.CompareResult) > 0 {
            goto done
        }
    }

    // 比对以外的小区
    if !args.IsCompareAll && len(args.TreeNodes) > 0 {
        baseScore += 20
    }

    for key, val := range cache.CacheMap.Area {
        if _, ok := walkedArea[key]; ok {
            continue
        }

        targets := val.Walk(DoSdkCompare, floatFeat, baseScore)
        if len(targets) > 0 {
            if len(targets) > 0 {
                // 比对结果去重, 同一个人到访过多个小区, 缓存数据内会有多条记录
                for idx, t := range targets {
                    var isRepeat bool
                    for _, r := range scResult.CompareResult {
                        if t.Id == r.Id {
                            isRepeat = true
                            break
                        }
                    }
                    if !isRepeat {
                        scResult.CompareResult = append(scResult.CompareResult, targets[idx])
                    }
                }
            }
            // todo 添加小区外的关联关系, 下次优先比对
        }
    }

done:
    logger.Debugf("比对结果 %d条", len(scResult.CompareResult))
    if len(scResult.CompareResult) > 0 {
        logger.Debugf("比对结果%+v", scResult.CompareResult)
    }

    buf, err := proto.Marshal(&scResult)
    buf, err := proto.Marshal(&response)
    if err != nil {
        logger.Error("scResult Marshal error!", err)
        logger.Error("response Marshal error!", err)
        return nil
    }


 compare/keyPerson.go

New file
@@ -0,0 +1,18 @@
package compare

import (
    "basic.com/pubsub/protomsg.git"
    "sdkCompare/cache"
)

func keyPersonsCompere(args protomsg.CompareArgs, fFeature []float32, baseScore float32) protomsg.SdkCompareResult {
    var scr protomsg.SdkCompareResult
    targets := cache.KeyPersonDbMap.Walk(DoSdkCompare, fFeature, baseScore)
    if len(targets) > 0 {
        for idx, _ := range targets {
            scr.CompareResult = append(scr.CompareResult, targets[idx])
        }
    }

    return scr
}

 compare/realName.go

New file
@@ -0,0 +1,18 @@
package compare

import (
    "basic.com/pubsub/protomsg.git"
    "sdkCompare/cache"
)

func realNamePersonsCompere(args protomsg.CompareArgs, fFeature []float32, baseScore float32) protomsg.SdkCompareResult {
    var scr protomsg.SdkCompareResult
    targets := cache.RealNameDbMap.Walk(DoSdkCompare, fFeature, baseScore)
    if len(targets) > 0 {
        for idx, _ := range targets {
            scr.CompareResult = append(scr.CompareResult, targets[idx])
        }
    }

    return scr
}

 db/realPersons.go

New file
@@ -0,0 +1,75 @@
package db

import (
    "encoding/base64"
    "sdkCompare/util"
)

type Layouts struct {
    Id         string `gorm:"column:id"`
    TableId    string `gorm:"column:tableId"`
    Name       string `gorm:"column:name"`
    Phone      string `gorm:"column:phone"`
    Address    string `gorm:"column:address"`
    Age        string `gorm:"column:age"`
    Gender     string `gorm:"column:gender"`
    IdCard     string `gorm:"column:id_card"`
    PersonType string `gorm:"column:person_type"` //'1：常驻人口，2：前科人员，3：涉毒人员，4：涉稳人员；5：在逃人员，6:流动人口，7:寄住人口，',
    Feature    string `gorm:"column:feature"`
}

func (l *Layouts) TableName() string {
    return "layouts"
}

func (l *Layouts) GetRealNamePersonList() (arr []*FeatureCacheBase, err error) {
    var persons []Layouts
    sql := "select id, tableId, person_type, feature from layouts where person_type = '1'"
    err = db.Raw(sql).Find(&persons).Error
    if err != nil {
        return nil, nil
    }

    for _, p := range persons {
        if p.Feature != "" {
            byteFeat, err := base64.StdEncoding.DecodeString(p.Feature)
            if err != nil {
                continue
            }

            arr = append(arr, &FeatureCacheBase{
                Id:          p.Id,
                TableId:     p.TableId,
                FaceFeature: util.ByteSlice2float32Slice(byteFeat),
            })
        }
    }

    return
}

func (l *Layouts) GetKeyPersonList() (arr []*FeatureCacheBase, err error) {
    var persons []Layouts
    sql := "select id, id_card, person_type, feature from layouts where person_type != '1'"
    err = db.Raw(sql).Find(&persons).Error
    if err != nil {
        return nil, nil
    }

    for _, p := range persons {
        if p.Feature != "" {
            byteFeat, err := base64.StdEncoding.DecodeString(p.Feature)
            if err != nil {
                continue
            }

            arr = append(arr, &FeatureCacheBase{
                Id:          p.Id,
                TableId:     p.IdCard,
                FaceFeature: util.ByteSlice2float32Slice(byteFeat),
            })
        }
    }

    return
}

 main.go

@@ -42,7 +42,7 @@
        return
    }

    cache.InitDbTablePersons()
    cache.InitCache()

    serveUrl := "tcp://0.0.0.0:"
    serveUrl = serveUrl + strconv.Itoa(config.MainConf.ServePort)
@@ -50,7 +50,6 @@
    logger.Infof("%s serve url:%s", procName, serveUrl)

    var ctx, cancel = context.WithCancel(context.Background())

    serve.Start(ctx, serveUrl, config.MainConf.WorkerNum)

    quit := make(chan os.Signal)

 serve/serve.go

@@ -41,7 +41,7 @@
                var compareArgInfo protomsg.CompareArgs
                if err = proto.Unmarshal(request.Payload, &compareArgInfo); err == nil {
                    timeStart := time.Now()
                    result = compare.GetComparePersonBaseInfo(compareArgInfo)
                    result = compare.Walk(compareArgInfo)
                    logger.Debug("用时：", time.Since(timeStart))
                } else {
                    logger.Warn("CompareArgs or EsPersonCacheChange json unmarshal error")
@@ -83,10 +83,12 @@
        logger.Error("can't get new rep socket: %s", err)
        return
    }

    if err = sock.SetOption(mangos.OptionRaw, true); err != nil {
        logger.Error("can't set raw mode: %s", err)
        return
    }

    sock.AddTransport(ipc.NewTransport())
    sock.AddTransport(tcp.NewTransport())
    if err = sock.Listen(url); err != nil {