package main import ( "encoding/json" "github.com/importcjj/sensitive" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo/options" "gopkg.in/mgo.v2/bson" "log" "regexp" "runtime" "sensitiveWords.udp/util" "strings" "sync" "time" ) var reg_alias = regexp.MustCompile("(税务局|工商行政管理局|文化广播电视新闻出版局|外国专家局|" + "中医药管理局|市场监督管理局|广播电视局|医疗保障局|机关事务管理局|粮食和物资储备局|" + "监狱管理局|畜牧兽医局|食品药品监督管理局|城市管理行政执法局|城市管理局|国家保密局|密码管理局|" + "地方金融监督管理局|住房保障和房屋管理局|质量技术监督局|人力资源与社会保障局|公路管理局|国土资源局|" + "卫生和计划生育局|民事政务局|公众安全局|交通管理局|人力资源和社会保障局|劳动和社会保障局|" + "住房和城乡建设局|就业服务局|文物管理局|环境保护局|粮食和物资储备局|教育体育局|" + "体育局|教育局|招商局|农业局|农机局|水务局|林业局|财政局|审计局|统计局|商务局)$") var reglen *regexp.Regexp = regexp.MustCompile("^(.{1,5}|.{40,})$") var strReg *regexp.Regexp = regexp.MustCompile("^(.{0,3}工程队|.{0,3}总公司|_+|.{0,2}设备安装公司|.{0,2}装[饰修潢]公司|.{0,2}开发公司|.{0,4}有限公司|.{0,4}有限责任公司|.{0,4}设计院|建筑设计研?究?院|省文物考古研究所|经济开发区|省.*|镇人民政府|.{0,2}服务公司|" + ".{0,2}工程质量监督站|.{0,3}经[营销]部|.{0,3}事务所|.{0,4}工程公司|.{0,4}责任公司|.*勘测|.{0,4}研究院|.*能源建|.{0,2}安装工程|.*[市省]{1}|.{0,4}中心|.*区.?|" + ".{0,3}税务局|.{0,3}财政局|.{0,3}商行|.{0,2}公安处|.{0,2}测绘院|.{0,3}开发|.{0,2}建设局|.{0,2}经销部|.{0,3}委员会|.{0,2}分公司|.{0,2}管理站|.{0,2}事务管理局|" + ".*资料|.{0,2}办公用品.{1,2}|.*唯亭|.*设备|.+安装|.{0,2}技术服务|市.+[台院社局司]|城?区.+[府局室院]|县.+[院台局]|.{0,2}发展公司|经济技术开发|" + "发展和改革局|贵州有色地质|铝塑门窗加工|生产力促进中心|特殊普通合伙|工业集团公司|人民调解协会|人民政府办公厅|机电设备公司|房地产开发有限公司|.{0,4}商店|中等专业学校|" + "农村信用联社|.{0,4}经营部|.{0,4}销售部|驾驶员培训学校|.{2}县.{2}镇|保安服务总公司|住房和城乡建设局|地产评估事务所|生产资料门市部|×+|.{0,3}[0-9]{15}|.*[0-9]+|.*路|.*无字号名称.*|.*车|.*[,,]{1}.*|.*个体工商户|.*运输户)$") //非中文开头... var unstart_strReg *regexp.Regexp = regexp.MustCompile("^([\u4e00-\u9fa5])") //开头 var start_strReg *regexp.Regexp = regexp.MustCompile("^([a-zA-Z]{1,2}[\u4e00-\u9fa5]{6,}|省|市|县|区|业绩|资格|中标|项目|预算单位)") //结尾 var end_strReg *regexp.Regexp = regexp.MustCompile("(\\.|\\.\\.|餐馆|店|腻子|肉庄|画社|美发屋|发廊|网吧|网咖|零售点|新街|包子铺|奶茶铺|(株)|先生|女士|小姐|" + "资格|业绩|中标|项目|预算单位|摊位号|号|厅|室|部|点|馆|场|厂|床|所|处|站|行|中心|合作社|ATMS|" + "吧|楼|摊|摊位|廊|茶社|坊|圃|汤锅|园|民宿|美容院|房|排挡|府|庄|栈|队|批发|苑|养殖户|棋牌|农家乐|货运|" + "城|社|基地|会|服务|娱乐|种植|百货|汽修|农家菜|亭|小吃|快餐|粮库|卫生院|书画院|面|门窗|鸡排|屋|橱|堂|肉铺|服务|服饰|/*)$") //包含 var con_strReg *regexp.Regexp = regexp.MustCompile("(\\?|?|%|代码标识|删除|错误|吊销|注销|发起人|待清理|&#|护照号|身份证号|" + "法人| |国家拨入|借款|积累资金|单位自有|认股人|--|、|&|`|美元|[\u4e00-\u9fa5]{2,6}·[\u4e00-\u9fa5]{2,6})|" + "[a-zA-Z]{5,}") var uncon_strReg *regexp.Regexp = regexp.MustCompile("(园|政府|集团|公司|有限|合伙|企|院|学|局|处)") //部署-历史-敏感词库 func initSensitiveWordsData() { log.Println("初始化敏感词-源数据...") gteid, err := primitive.ObjectIDFromHex(YamlConfig.TaskGteId) if err != nil { log.Fatalln(err) } lteid, err := primitive.ObjectIDFromHex(YamlConfig.TaskLteId) if err != nil { log.Fatalln(err) } log.Println("id段落:",util.BsonTOStringId(gteid),util.BsonTOStringId(lteid)) iter := MixDataMgo.GetMgoConn().C("unique_qyxy").Find(map[string]interface{}{ "_id": map[string]interface{}{ "$gte": gteid, "$lte": lteid, }, }).Sort("_id").Iter() Filter = sensitive.New() var initnum uint for tmp := map[string]interface{}{}; iter.Next(&tmp); tmp = map[string]interface{}{} { initnum++ if initnum%10000==0 { log.Println("current index ", initnum,tmp["qy_name"]) } Filter.AddWord(tmp["qy_name"].(string)) } log.Println("init ok", initnum) } //定时增量数据处理---冯 func addTaskSensitiveWordsData() { mmmgo, err := util.InitMgoEn("mongodb://172.17.4.187:27082,172.17.145.163:27083", 20, "fengweiqiang", "fwq@123123") if err != nil { log.Fatalln(err) } con := mmmgo.GetCon() if con == nil { log.Fatalln("mgo con err") } Filter = sensitive.New() tick := time.Tick(time.Hour * 24 * 7)//查询七天前 for {//定时任务 ctime := <-tick cronData := time.Date(ctime.Year(), ctime.Month(), ctime.Day()-7, ctime.Hour(), ctime.Minute(), ctime.Second(), 0, time.Local) findByupdate, err := con.Database("mixdata").Collection("qyxy_std").Find(nil, bson.M{ "updatetime": bson.M{"$gte": cronData.Unix()}, }, options.Find().SetProjection(bson.M{"company_name": 1, "updatetime": 1,"company_type": 1,"company_type_old": 1}).SetSort(bson.M{"_id": 1})) if err != nil { log.Println("tick err", cronData) continue } defer findByupdate.Close(nil) for tmp := make(map[string]interface{}); findByupdate.Next(nil); tmp = map[string]interface{}{} { err := findByupdate.Decode(&tmp) if err == nil { if company_name, ok := tmp["company_name"].(string); ok { if reglen.MatchString(company_name) || strReg.MatchString(company_name) || !uncon_strReg.MatchString(company_name)|| !unstart_strReg.MatchString(company_name)|| start_strReg.MatchString(company_name)|| end_strReg.MatchString(company_name)|| con_strReg.MatchString(company_name) { continue } if strings.Contains(util.ObjToString(tmp["company_type"]),"个人")|| strings.Contains(util.ObjToString(tmp["company_type"]),"个体")|| strings.Contains(util.ObjToString(tmp["company_type_old"]),"个人")|| strings.Contains(util.ObjToString(tmp["company_type_old"]),"个体") { continue } //存mgo con.Database("mixdata").Collection("unique_qyxy").InsertOne(nil, bson.M{ "qy_name": company_name, }) //存敏感词 Filter.AddWord(company_name) //存es=判断+新增 dealWithEsData(company_name,util.BsonTOStringId(tmp["_id"])) } } } log.Println("tick ok", cronData) } } //处理是否新增es func dealWithEsData(name string,tmpid string) { query:= `{"query":{"bool":{"must":[{"term":{"`+es_index+`.name":"`+name+`"}}],"must_not":[],"should":[]}},"from":0,"size":10,"sort":[],"facets":{}}` tmp := make(map[string]interface{}) json.Unmarshal([]byte(query),&tmp) searchResult, err := Client_Es.Search().Index(es_index).Type(es_type).Source(tmp).Do() if err != nil { log.Println("从ES查询出错", err.Error()) }else { data := make(map[string]interface{},0) if searchResult.Hits != nil { for _, hit := range searchResult.Hits.Hits { json.Unmarshal(*hit.Source, &data) } } if len(data)==0{ //log.Println("无数据-新增") _, err := Client_Es.Index().Index(es_index).Type(es_type).Id(tmpid).BodyJson(map[string]interface{}{ "name":name, "name_word":name, }).Do() if err != nil { //log.Println("新增失败:",name,tmpid) } } } } //处理内存分段 func dealWithDataMemory() { iter := MixDataMgo.GetMgoConn().C("unique_qyxy").Find(map[string]interface{}{ "_id": map[string]interface{}{ "$gte": util.BsonTOStringId("1fffffffffffffffffffffff"), "$lte": util.BsonTOStringId("9fffffffffffffffffffffff"), }, }).Sort("_id").Iter() Filter = sensitive.New() var initnum uint saveIdArr ,start_id:= make([]map[string]string,0),"" var m runtime.MemStats for tmp := map[string]interface{}{}; iter.Next(&tmp); tmp = map[string]interface{}{} { if start_id=="" { start_id = util.BsonTOStringId(tmp["_id"]) } Filter.AddWord(tmp["qy_name"].(string)) initnum++ if initnum%100000==0 { runtime.ReadMemStats(&m) men :=util.ToMegaBytes(m.HeapAlloc) log.Printf("current index %d\tos %.2f M",initnum, men) if men>7.5*1024 { //7.5G saveIdArr = append(saveIdArr, map[string]string{ "start":start_id, "end":util.BsonTOStringId(tmp["_id"]), }) runtime.GC() Filter = sensitive.New() start_id = "" time.Sleep(time.Second*5) } } break } saveIdArr = append(saveIdArr, map[string]string{ "start":start_id, "end":"", }) for k,v:=range saveIdArr{ log.Println("第",k,"段",v["start"],v["end"]) } log.Println("memory is ok", initnum) } func temporaryTest() { log.Println("测试......导出数据") q := map[string]interface{}{} sess := MixDataMgo.GetMgoConn() defer MixDataMgo.DestoryMongoConn(sess) //多线程升索引 pool_es := make(chan bool, 20) wg_es := &sync.WaitGroup{} it := sess.DB(MixDataMgo.DbName).C("zk_company_test").Find(&q).Iter() total,isOK:=0,0 for tmp := make(map[string]interface{}); it.Next(&tmp); total++ { if total % 1000 == 0 { log.Println("current index",total,isOK) } name:=util.ObjToString(tmp["name"]) pool_es <- true wg_es.Add(1) go func(name string) { defer func() { <-pool_es wg_es.Done() }() new_name,b :=dealWithNameScoreRules(name) if new_name!="" && b { isOK++ MixDataMgo.Save("zk_words_test", map[string]interface{}{ "name":name, "new_name":new_name, }) } }(name) tmp = make(map[string]interface{}) } wg_es.Wait() log.Println("is over",total) }