jy-data-extract/udpfilterdup/src2

{

"udpport": ":19097",
"dupdays": 7,
"mongodb": {
    "addr": "172.17.4.85:27080",
    "pool": 10,
    "db": "qfw",
    "extract": "result_20200715",
    "extract_back": "result_20200714",
    "site": {
        "dbname": "qfw",
        "coll": "site"
    }
},
"task_mongodb": {
    "task_addrName": "172.17.4.187:27081",
    "task_dbName": "qfw",
    "task_collName": "ocr_flie_over",
    "pool": 5

},
"jkmail": {
    "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
    "api": "http://10.171.112.160:19281/_send/_mail"
},
"nextNode": [
    {
        "addr": "172.17.4.194",
        "port": 1782,
        "stype": "project",
        "memo": "合并项目"
    },
    {
        "addr": "127.0.0.1",
        "port": 1783,
        "stype": "bidding",
        "memo": "创建招标数据索引new"
    }
],
"threads": 1,
"isMerger": false,
"lowHeavy":true,
"timingTask":false,
"timingSpanDay": 5,
"timingPubScope": 720,
"specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
"specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
"specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
"specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
"beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"

}

mgo = &MongodbSim{

    MongodbAddr: "172.17.4.187:27083",
    DbName:      "qfw",
    Size:        10,
}

mgo.InitPool()

return

func moveTimeoutData() {

log.Println("部署迁移定时任务")
c := cron.New()
c.AddFunc("0 0 0 * * ?", func() { moveOnceTimeOut() })
c.Start()

}

func moveOnceTimeOut() {

log.Println("执行一次迁移超时数据")

sess := mgo.GetMgoConn()
defer mgo.DestoryMongoConn(sess)
now:=time.Now()
move_time := time.Date(now.Year()-2, now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).Unix()
q := map[string]interface{}{
    "comeintime": map[string]interface{}{
        "$lt": move_time,
    },
}
log.Println(q)
it := sess.DB(mgo.DbName).C("result_20200714").Find(&q).Iter()
index := 0
for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
    if index%10000 == 0 {
        log.Println("index", index)
    }
    del_id:=BsonTOStringId(tmp["_id"])
    mgo.Save("result_20200713", tmp)
    mgo.DeleteById("result_20200714",del_id)
    tmp = map[string]interface{}{}
}
log.Println("save and delete", " ok index", index)

}

{

"udpport": ":1785",
"dupdays": 7,
"mongodb": {
    "addr": "172.17.4.85:27080",
    "pool": 10,
    "db": "qfw",
    "extract": "result_20200715",
    "extract_back": "result_20200714",
    "site": {
        "dbname": "qfw",
        "coll": "site"
    }
},
"jkmail": {
    "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
    "api": "http://172.17.145.179:19281/_send/_mail"
},
"nextNode": [
    {
        "addr": "127.0.0.1",
        "port": 1783,
        "stype": "bidding",
        "memo": "创建招标数据索引new"
    }
],
"threads": 1,
"isMerger": false,
"lowHeavy":true,
"timingTask":false,
"timingSpanDay": 5,
"timingPubScope": 720,
"specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
"specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
"specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
"specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
"beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"

}

func repairHistory() {

defer util.Catch()
log.Println("执行修复程序")
sess := mgo.GetMgoConn()
defer mgo.DestoryMongoConn(sess)
q:=map[string]interface{}{}
between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
//开始判重
q = map[string]interface{}{
    "_id": map[string]interface{}{
        "$gt": StringTOBsonId("5f15bf800000000000000000"),
        "$lte": StringTOBsonId("5f2375b2a120e23754be1039"),
    },
}
log.Println("历史判重查询条件：",q,"时间：", between_time)

it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
num,oknum, deterTime:= int64(0),int64(0),int64(0) //计数
updateExtract := [][]map[string]interface{}{}//批量更新mongo数组
pendAllArr:=[][]map[string]interface{}{}//待处理数组
dayArr := []map[string]interface{}{}
for tmp := make(map[string]interface{}); it.Next(&tmp); num++ {
    if num%10000 == 0 {
        log.Println("正序遍历:", num)
    }
    source := util.ObjToMap(tmp["jsondata"])
    if util.IntAll((*source)["sourcewebsite"]) == 1 {
        updateExtract = append(updateExtract, []map[string]interface{}{
            map[string]interface{}{
                "_id": tmp["_id"],
            },
            map[string]interface{}{
                "$set": map[string]interface{}{
                    "repeat": 1,
                    "dataging": 0,
                    "repeat_reason": "sourcewebsite为1 重复",
                },
            },
        })
        if len(updateExtract) > 50 {
            mgo.UpSertBulk(extract, updateExtract...)
            updateExtract = [][]map[string]interface{}{}
        }
        tmp = make(map[string]interface{})
        continue
    }

    //取-符合-发布时间X年内的数据
    if util.IntAll(tmp["dataging"]) == 1 {
        pubtime := util.Int64All(tmp["publishtime"])
        if pubtime > 0 && pubtime >= between_time {
            oknum++
            if deterTime==0 {
                log.Println("找到第一条符合条件的数据")
                deterTime = util.Int64All(tmp["publishtime"])
                dayArr = append(dayArr,tmp)
            }else {
                if pubtime-deterTime >timingSpanDay*86400 {
                    //新数组重新构建，当前组数据加到全部组数据
                    pendAllArr = append(pendAllArr,dayArr)
                    dayArr = []map[string]interface{}{}
                    deterTime = util.Int64All(tmp["publishtime"])
                    dayArr = append(dayArr,tmp)
                }else {
                    dayArr = append(dayArr,tmp)
                }
            }
        }else {
            //不在两年内的也清标记
            updateExtract = append(updateExtract, []map[string]interface{}{
                map[string]interface{}{
                    "_id": tmp["_id"],
                },
                map[string]interface{}{
                    "$set": map[string]interface{}{
                        "dataging": 0,
                    },
                },
            })
            if len(updateExtract) > 50 {
                mgo.UpSertBulk(extract, updateExtract...)
                updateExtract = [][]map[string]interface{}{}
            }

        }
    }
    tmp = make(map[string]interface{})
}


//批量更新标记
if len(updateExtract) > 0 {
    mgo.UpSertBulk(extract, updateExtract...)
    updateExtract = [][]map[string]interface{}{}
}

if len(dayArr)>0 {
    pendAllArr = append(pendAllArr,dayArr)
    dayArr = []map[string]interface{}{}
}

log.Println("查询数量：",num,"符合条件：",oknum)

if len(pendAllArr) <= 0 {
    log.Println("没找到dataging==1的数据")
}

//测试分组数量是否正确
testNum:=0
for k,v:=range pendAllArr {
    log.Println("第",k,"组--","数量：",len(v))
    testNum = testNum+len(v)
}
log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)

n, repeateN := 0, 0
pool := make(chan bool, 2)
wg := &sync.WaitGroup{}
for k,v:=range pendAllArr { //每组结束更新一波数据
    pool <- true
    wg.Add(1)
    go func(k int, v []map[string]interface{}) {
        defer func() {
            <-pool
            wg.Done()
        }()
        //每组临时数组 -  互不干扰
        groupUpdateExtract := [][]map[string]interface{}{}
        //构建当前组的数据池
        log.Println("构建第", k, "组---（数据池）")
        //当前组的第一个发布时间
        first_pt := util.Int64All(v[len(v)-1]["publishtime"])
        curTM := TimedTaskDatamap(dupdays+int(timingSpanDay)+1, first_pt+86400, int(k))
        log.Println("开始遍历判重第", k, "组  共计数量：", len(v))
        n = n + len(v)
        log.Println("统计目前总数量：", n, "重复数量：", repeateN)
        for _, tmp := range v {
            info := NewInfo(tmp)
            if !LowHeavy { //是否进行低质量数据判重
                if invalidData(info.buyer, info.projectname, info.projectcode, info.contractnumber) {
                    log.Println("无效数据")
                    groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
                        map[string]interface{}{
                            "_id": tmp["_id"],
                        },
                        map[string]interface{}{
                            "$set": map[string]interface{}{
                                "repeat":   -1, //无效数据标签
                                "dataging": 0,
                            },
                        },
                    })
                    if len(groupUpdateExtract) > 50 {
                        mgo.UpSertBulk(extract, groupUpdateExtract...)
                        groupUpdateExtract = [][]map[string]interface{}{}
                    }
                    return
                }
            }
            b, source, reason := curTM.check(info)
            if b { //有重复,生成更新语句，更新抽取和更新招标
                repeateN++
                //重复数据打标签
                groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
                    map[string]interface{}{
                        "_id": tmp["_id"],
                    },
                    map[string]interface{}{
                        "$set": map[string]interface{}{
                            "repeat":        1,
                            "repeat_reason": reason,
                            "repeat_id":     source.id,
                            "dataging":      0,
                        },
                    },
                })
            } else {
                groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
                    map[string]interface{}{
                        "_id": tmp["_id"],
                    },
                    map[string]interface{}{
                        "$set": map[string]interface{}{
                            "dataging": 0, //符合条件的都为dataging==0
                        },
                    },
                })
            }
            if len(groupUpdateExtract) > 50 {
                mgo.UpSertBulk(extract, groupUpdateExtract...)
                groupUpdateExtract = [][]map[string]interface{}{}
            }
        }
        //每组数据结束-更新数据
        if len(groupUpdateExtract) > 0 {
            mgo.UpSertBulk(extract, groupUpdateExtract...)
        }
    }(k, v)

}

wg.Wait()


time.Sleep(30 * time.Second)
log.Println("this repair over.", n, "repeateN:", repeateN,gtid,lteid)
log.Println("修复结束")

}

//if !LowHeavy { //是否进行低质量数据判重

        //  if invalidData(info.buyer, info.projectname, info.projectcode, info.contractnumber) {
        //      updateExtract = append(updateExtract, []map[string]interface{}{
        //          map[string]interface{}{
        //              "_id": tmp["_id"],
        //          },
        //          map[string]interface{}{
        //              "$set": map[string]interface{}{
        //                  "repeat": -1, //无效数据标签
        //              },
        //          },
        //      })
        //      if len(updateExtract) >= 200 {
        //          mgo.UpSertBulk(extract, updateExtract...)
        //          updateExtract = [][]map[string]interface{}{}
        //      }
        //      return
        //  }
        //}




        if !LowHeavy { //是否进行低质量数据判重
                                if invalidData(info.buyer, info.projectname, info.projectcode, info.contractnumber) {
                                    log.Println("无效数据")
                                    updatelock.Lock()

                                    groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
                                        map[string]interface{}{
                                            "_id": tmp["_id"],
                                        },
                                        map[string]interface{}{
                                            "$set": map[string]interface{}{
                                                "repeat":   -1, //无效数据标签
                                                "dataging": 0,
                                            },
                                        },
                                    })
                                    if len(groupUpdateExtract) > 200 {
                                        mgo.UpSertBulk(extract, groupUpdateExtract...)
                                        groupUpdateExtract = [][]map[string]interface{}{}
                                    }

                                    updatelock.Unlock()

                                    return
                                }
                            }

 //是否合并-低质量数据不合并
                if isMerger && !strings.Contains(reason,"低质量"){
                    newData, update_map ,isReplace := mergeDataFields(source, info)
                    if isReplace {//替换-数据池
                        fmt.Println("合并更新的id：",source.id)
                        //数据池 - 替换
                        DM.replacePoolData(newData)
                        //mongo更新 - 具体字段 - merge
                        mgo.UpdateById(extract,source.id,update_map)
                        //发udp  更新索引
                        //for _, to := range nextNode {
                        //  key := source.id + "-" + source.id + "-" + util.ObjToString(to["stype"])
                        //  by, _ := json.Marshal(map[string]interface{}{
                        //      "gtid":  source.id,
                        //      "lteid": source.id,
                        //      "stype": "biddingall",
                        //      "key":   key,
                        //  })
                        //  addr := &net.UDPAddr{
                        //      IP:   net.ParseIP(to["addr"].(string)),
                        //      Port: util.IntAll(to["port"]),
                        //  }
                        //  node := &udpNode{by, addr, time.Now().Unix(), 0}
                        //  udptaskmap.Store(key, node)
                        //  udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
                        //}
                    }
                }