jy-data-extract/udpdataclear/udpSensitiveWords/grpc_server/data.go

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"
	"strings"
	"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("(\\?|？|%|代码标识|删除|错误|吊销|注销|发起人|待清理|&#|护照号|身份证号|" +
	"法人|&nbsp|国家拨入|借款|积累资金|单位自有|认股人|--|、|&|`|美元|[\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段落：",BsonTOStringId(gteid),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 := 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) {
						continue
					}
					if strings.Contains(ObjToString(tmp["company_type"]),"个人")||
						strings.Contains(ObjToString(tmp["company_type"]),"个体")||
						strings.Contains(ObjToString(tmp["company_type_old"]),"个人")||
						strings.Contains(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,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": BsonTOStringId("1fffffffffffffffffffffff"),
			"$lte":  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 = BsonTOStringId(tmp["_id"])
		}
		Filter.AddWord(tmp["qy_name"].(string))
		initnum++
		if initnum%100000==0 {
			runtime.ReadMemStats(&m)
			men :=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":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)
}