jy-data-extract/qyxy/qyxy_std/task.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/cron"
	"github.com/wangbin/jiebago"
	"go.mongodb.org/mongo-driver/bson"
	"qfw/util"
	"qfw/util/redis"
	"regexp"
	"strings"
	"sync"
	"time"
)

var (
	AreaFiled				 = []string{"credit_no", "company_code"}
	WordsArr 				= []string{"研发", "研制", "开发", "生产", "制造", "制作", "加工", "种植"}

	seg 					jiebago.Segmenter
	regPre, _ 				= regexp.Compile(`^(.+[省|市|区|县|州])?(.+)`)
)

//	企业基本信息
var company_base = []string{"company_name", "company_code", "credit_no", "org_code", "legal_person", "company_status",
	"authority", "establish_date", "issue_date", "operation_startdate", "operation_enddate", "capital", "company_type",
	"company_status", "company_address", "business_scope", "cancel_date", "cancel_reason", "revoke_date", "revoke_reason",
	"legal_person_type", "real_capital", "en_name", "list_code", "tax_code", "use_flag",
}

var province_map = map[string]string{
	"BJ": "北京", "TJ": "天津", "SH": "上海", "CQ": "重庆", "HB": "河北", "SX": "山西", "NMG": "内蒙", "LN": "辽宁", "JL": "吉林",
	"HLJ": "黑龙江", "JS": "江苏", "ZJ": "浙江", "AH": "安徽", "FJ": "福建", "JX": "江西", "SD": "山东", "HEN": "河南", "HUB": "湖北",
	"HUN": "湖南", "GD": "广东", "GX": "广西", "HAIN": "海南", "SC": "四川", "GZ": "贵州", "YN": "云南", "XZ": "西藏", "SAX": "陕西",
	"GS": "甘肃", "QH": "青海", "NX": "宁夏", "XJ": "新疆",
}

func TimeTask() {
	c := cron.New()
	//cronstr := "0 0 " + fmt.Sprint(TaskTime) + " * * ?"			//每天TaskTime跑一次
	cronstrPa := "0 0 15 ? * WED"								//凭安增量数据每周二跑一次
	_ = c.AddFunc(cronstrPa, func() {
		TaskFun()
	})
	c.Start()
}

func TaskFun() {
	defer util.Catch()
	sess := MongoTool1.GetMgoConn()
	defer MongoTool1.DestoryMongoConn(sess)

	pool := make(chan bool, 10)
	wg := &sync.WaitGroup{}

	//q := bson.M{"_id": bson.M{"$gt": lastId}}
	//q := bson.M{"_id": 262454280}
	it := sess.DB("mixdata").C("qyxy_tmp").Find(nil).Select(nil).Iter()
	count := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		lastId = util.Int64All(tmp["_id"])
		if count%20000 == 0 {
			util.Debug("current:", count)
		}
		pool <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-pool
				wg.Done()
			}()
			IncStd(tmp)
		}(tmp)
	}

	util.Debug("over ---", count)
}

// IncStd 增量数据	临时表qyxy_tmp
func IncStd(tmp map[string]interface{}) {
	update := make(map[string]interface{})
	save := make(map[string]interface{})
	push := make(map[string]interface{})
	// company_base
	if tmp["company_base"] != nil {
		companyArr := util.ObjArrToMapArr(tmp["company_base"].([]interface{}))
		company := companyArr[len(companyArr)-1]
		for _, v := range company_base{
			if company[v] == nil {
				continue
			}
			// company_type
			if v == "company_type" {
				save["company_type_old"] = company[v]
				if text := util.ObjToString(company["company_type"]); text != "" {
					if strings.Contains(text, "个体") || strings.Contains(text, "非公司")  {
						save["company_type"] = "个体工商户"
					} else {
						text = strings.ReplaceAll(text, "（", "(")
						text = strings.ReplaceAll(text, "）", ")")
						if stype := QyStypeMap[text]; stype != "" {
							save["company_type"] = stype
						} else {
							save["company_type"] = "其他"
						}
					}
				}
				// company_status
			}else if v == "company_status" {
				save["company_status_old"] = company[v]
				if text := util.ObjToString(company["company_status"]); text != "" {
					text = strings.ReplaceAll(text, "（", "(")
					text = strings.ReplaceAll(text, "）", ")")
					if status := CompanyStatusMap[text]; status != "" {
						save["company_status"] = status
					} else {
						save["company_status"] = "其他"
					}
				}
			}else if v == "capital" {
				// capital/currency
				text := util.ObjToString(company[v])
				if currency := GetCurrency(text); currency != "" {
					save["currency"] = currency //币种
				}
				capital := ObjToMoney(text)
				capital = capital / 10000
				if capital != 0 {
					save[v] = capital
				}
			}else if v == "use_flag" {
				save[v] = util.IntAll(company[v])
			}else {
				save[v] = company[v]
			}
		}
		// mysql create_time/update_time
		save["create_time_msql"] = company["create_time"]
		save["update_time_msql"] = company["update_time"]
		// company_area/company_city/company_district
		pshort := util.ObjToString(company["province_short"])
		save["company_area"] = province_map[pshort]
		for i, field := range AreaFiled {
			if company[field] == nil {
				continue
			}
			if code := fmt.Sprint(company[field]); code != "" {
				if i == 0 && len(code) >= 8 { //credit_no企业信用代码
					code = code[2:8]
				} else if i == 1 && len(code) >= 6 { //company_code注册号
					code = code[:6]
				}
				if city := AddressMap[code]; city != nil { //未作废中取
					if city.Province != "" && city.Province == save["company_area"] {
						if city.City != "" {
							save["company_city"] = city.City //市
						}
						if city.District != "" {
							save["company_district"] = city.District //县
						}
						break
					}
				} else { //作废中取
					if city := AddressOldMap[code]; city != nil {
						if city.Province != "" && city.Province == save["company_area"] {
							if city.City != "" {
								save["company_city"] = city.City //市
							}
							if city.District != "" {
								save["company_district"] = city.District //县
							}
						}
						break
					}
				}
			}
		}
	}
	save["_id"] = tmp["_id"]
	save["updatetime"] = time.Now().Unix()
	if save["company_type"] != "个体工商户" {
		// history_name
		if tmp["company_history_name"] != nil {
			history_name := util.ObjArrToMapArr(tmp["company_history_name"].([]interface{}))
			var names []string
			for _, h := range history_name{
				names = append(names, util.ObjToString(h["history_name"]))
			}
			save["history_name"] =  strings.Join(names, ",")
		}
		// company_employee
		if tmp["company_employee"] != nil {
			employees := util.ObjArrToMapArr(tmp["company_employee"].([]interface{}))
			var names []string
			var arr []map[string]interface{}
			for _, v := range employees{
				tmp := make(map[string]interface{})
				tmp["employee_name"] = v["employee_name"]
				tmp["position"] = v["position"]
				tmp["is_history"] = v["is_history"]
				arr = append(arr, tmp)
				names = append(names, util.ObjToString(v["employee_name"]))
			}
			//save["employees"] = arr
			push["employees"] = bson.M{"$each": arr}
			save["employee_name"] =  strings.Join(names, ",")
		}
		// company_partner
		if tmp["company_partner"] != nil {
			partners := util.ObjArrToMapArr(tmp["company_partner"].([]interface{}))
			var names []string
			var arr []map[string]interface{}
			tmp["partners"] =  partners
			for _, v := range partners{
				if util.IntAll(tmp["is_history"]) == 0 {
					tmp := make(map[string]interface{})
					tmp["stock_capital"] = v["stock_capital"]
					tmp["stock_name"] = v["stock_name"]
					tmp["identify_no"] = v["identify_no"]
					tmp["stock_realcapital"] = v["stock_realcapital"]
					tmp["is_history"] = v["is_history"]
					tmp["is_personal"] = v["is_personal"]
					tmp["stock_type"] = v["stock_type"]
					tmp["identify_type"] = v["identify_type"]
					arr = append(arr, tmp)
					names = append(names, util.ObjToString(v["stock_name"]))
				}
			}
			//save["partners"] =  arr
			push["partners"] = bson.M{"$each": arr}
			save["stock_name"] =  strings.Join(names, ",")
		}
		// annual_report_base
		if tmp["annual_report_base"] != nil {
			reports := util.ObjArrToMapArr(tmp["annual_report_base"].([]interface{}))
			var arr []map[string]interface{}
			year := 0
			phone, email := "", ""
			for _, v := range reports{
				tmp := make(map[string]interface{})
				tmp["operator_name"] = v["operator_name"]
				tmp["report_year"] = v["report_year"]
				tmp["zip_code"] = v["zip_code"]
				tmp["employee_no"] = v["employee_no"]
				tmp["company_phone"] = v["company_phone"]
				tmp["company_email"] = v["company_email"]
				arr = append(arr, tmp)
				if year < util.IntAll(v["report_year"]) {
					year = util.IntAll(v["report_year"])
					phone = util.ObjToString(v["company_phone"])
					email = util.ObjToString(v["company_email"])
				}
			}
			//save["annual_reports"] =  arr
			push["annual_reports"] = bson.M{"$each": arr}
			if year != 0 {
				save["company_phone"] = phone
				save["company_email"] = email
			}
		}
		// website_url
		if tmp["annual_report_website"] != nil {
			webs := util.ObjArrToMapArr(tmp["annual_report_website"].([]interface{}))
			year := 0
			web := ""
			for _, v := range webs{
				if year < util.IntAll(v["report_year"]) && util.IntAll(v["is_history"]) == 0 {
					year = util.IntAll(v["report_year"])
					web = util.ObjToString(v["website_url"])
				}
			}
			if year != 0 {
				save["website_url"] =  web
			}
		}
		// bid_contracttype
		var types []string
		if phone := util.ObjToString(save["company_phone"]); phone != "" {
			if len(phone) == 11 {
				types = append(types, "手机号")
			}else {
				types = append(types, "固定电话")
			}
		}
		if util.ObjToString(save["company_email"]) != "" {
			types = append(types, "邮箱")
		}

		companyName := util.ObjToString(save["company_name"])
		// redis bid_unittype、bid_purchasing、bid_projectname
		if b, err := redis.Exists("qyxy_winner", companyName); err == nil && b {
			maps := make(map[string]interface{})
			text := redis.GetStr("qyxy_winner", companyName)
			err1 := json.Unmarshal([]byte(text), &maps)
			if err1 != nil {
				util.Debug(companyName, "winner-----map解析异常")
			}else {
				for k := range maps{
					if k == "bid_contracttype" {
						t1 := util.ObjArrToStringArr(maps[k].([]interface{}))
						types = append(types, t1...)
					}else {
						save[k] = maps[k]
					}
				}
			}
		}
		if len(types) == 0 {
			types = append(types, "不存在")
		}
		save["bid_contracttype"] = types
		// bid_unittype
		flag := false
		for _, v := range WordsArr{
			if strings.Contains(util.ObjToString(save["business_scope"]), v) {
				flag = true
				break
			}
		}
		if flag {
			save["bid_unittype"] = []string{"厂商"}
		}
		// search_type
		if t := util.ObjToString(save["company_type"]); t != "" {
			if t != "个体工商户" && t != "其他" {
				t1 := util.ObjToString(save["company_type_old"])
				name := util.ObjToString(save["company_name"])
				if strings.Contains(t1, "有限合伙") {
					save["search_type"] = "有限合伙"
				}else if strings.Contains(t1, "合伙") {
					save["search_type"] = "普通合伙"
				}else if strings.Contains(name, "股份") ||
					(strings.Contains(t1, "上市") && !strings.Contains(t1, "非上市")) {
					save["search_type"] = "股份有限公司"
				}else {
					save["search_type"] = "有限责任公司"
				}
			}
		}
		// company_shortname
		if m := getStName(util.ObjToString(save["company_name"])); m != "" {
			save["company_shortname"] = m
		}
	}

	update["$set"] = save
	if len(push) > 0 {
		update["$push"] = push
	}
	updataInfo := []map[string]interface{}{
		{"_id": save["_id"]},
		update,
	}
	updatePool <- updataInfo
}

// InfoStd 存量全量数据
func InfoStd(tmp map[string]interface{})  {
	save := make(map[string]interface{})
	for _, v := range company_base{
		if tmp[v] == nil {
			continue
		}
		// company_type
		if v == "company_type" {
			save["company_type_old"] = tmp[v]
			if text := util.ObjToString(tmp["company_type"]); text != "" {
				if strings.Contains(text, "个体") || strings.Contains(text, "非公司")  {
					save["company_type"] = "个体工商户"
				} else {
					text = strings.ReplaceAll(text, "（", "(")
					text = strings.ReplaceAll(text, "）", ")")
					if stype := QyStypeMap[text]; stype != "" {
						save["company_type"] = stype
					} else {
						save["company_type"] = "其他"
					}
				}
			}
		// company_status
		}else if v == "company_status" {
			save["company_status_old"] = tmp[v]
			if text := util.ObjToString(tmp["company_status"]); text != "" {
				text = strings.ReplaceAll(text, "（", "(")
				text = strings.ReplaceAll(text, "）", ")")
				if status := CompanyStatusMap[text]; status != "" {
					save["company_status"] = status
				} else {
					save["company_status"] = "其他"
				}
			}
		}else if v == "capital" {
			// capital/currency
			text := util.ObjToString(tmp[v])
			if currency := GetCurrency(text); currency != "" {
				save["currency"] = currency //币种
			}
			capital := ObjToMoney(text)
			capital = capital / 10000
			if capital != 0 {
				save[v] = capital
			}
		}else if v == "use_flag" {
			save[v] = util.IntAll(tmp[v])
		}else {
			save[v] = tmp[v]
		}
	}
	save["_id"] = tmp["company_id"]
	// mysql create_time/update_time
	save["create_time_msql"] = tmp["create_time"]
	save["update_time_msql"] = tmp["update_time"]
	save["createtime"] = time.Now().Unix()
	save["updatetime"] = time.Now().Unix()
	// company_area/company_city/company_district
	pshort := util.ObjToString(tmp["province_short"])
	save["company_area"] = province_map[pshort]
	for i, field := range AreaFiled {
		if tmp[field] == nil {
			continue
		}
		if code := fmt.Sprint(tmp[field]); code != "" {
			if i == 0 && len(code) >= 8 { //credit_no企业信用代码
				code = code[2:8]
			} else if i == 1 && len(code) >= 6 { //company_code注册号
				code = code[:6]
			}
			if city := AddressMap[code]; city != nil { //未作废中取
				if city.Province != "" && city.Province == save["company_area"] {
					if city.City != "" {
						save["company_city"] = city.City //市
					}
					if city.District != "" {
						save["company_district"] = city.District //县
					}
					break
				}
			} else { //作废中取
				if city := AddressOldMap[code]; city != nil {
					if city.Province != "" && city.Province == save["company_area"] {
						if city.City != "" {
							save["company_city"] = city.City //市
						}
						if city.District != "" {
							save["company_district"] = city.District //县
						}
					}
					break
				}
			}
		}
	}
	if tmp["company_type"] != "个体工商户" {
		// history_name
		historyNameFun(save)
		// company_employee
		employeeFun(save)
		// company_partner
		partnerFun(save)
		// annual_report_base
		reportFun(save)
		// website_url
		websiteFun(save)
		// bid_contracttype
		var types []string
		if phone := util.ObjToString(save["company_phone"]); phone != "" {
			if len(phone) == 11 {
				types = append(types, "手机号")
			}else {
				types = append(types, "固定电话")
			}
		}
		if util.ObjToString(save["company_email"]) != "" {
			types = append(types, "邮箱")
		}

		companyName := util.ObjToString(tmp["company_name"])
		// redis bid_unittype、bid_purchasing、bid_projectname
		if b, err := redis.Exists("qyxy_winner", companyName); err == nil && b {
			maps := make(map[string]interface{})
			text := redis.GetStr("qyxy_winner", companyName)
			err1 := json.Unmarshal([]byte(text), &maps)
			if err1 != nil {
				util.Debug(companyName, "winner-----map解析异常")
			}else {
				for k := range maps{
					if k == "bid_contracttype" {
						t1 := util.ObjArrToStringArr(maps[k].([]interface{}))
						types = append(types, t1...)
					}else {
						save[k] = maps[k]
					}
				}
			}
		}
		if len(types) == 0 {
			types = append(types, "不存在")
		}
		save["bid_contracttype"] = types
		// bid_unittype
		flag := false
		for _, v := range WordsArr{
			if strings.Contains(util.ObjToString(tmp["business_scope"]), v) {
				flag = true
				break
			}
		}
		if flag {
			save["bid_unittype"] = []string{"厂商"}
		}
		// search_type
		if t := util.ObjToString(save["company_type"]); t != "" {
			if t != "个体工商户" && t != "其他" {
				t1 := util.ObjToString(save["company_type_old"])
				name := util.ObjToString(save["company_name"])
				if strings.Contains(t1, "有限合伙") {
					save["search_type"] = "有限合伙"
				}else if strings.Contains(t1, "合伙") {
					save["search_type"] = "普通合伙"
				}else if strings.Contains(name, "股份") ||
					(strings.Contains(t1, "上市") && !strings.Contains(t1, "非上市")) {
					save["search_type"] = "股份有限公司"
				}else {
					save["search_type"] = "有限责任公司"
				}
			}
		}
		// company_shortname
		if m := getStName(util.ObjToString(tmp["company_name"])); m != "" {
			save["company_shortname"] = m
		}
	}

	saveInfo := []map[string]interface{}{
		{"_id": tmp["company_id"]},
		{"$set": save},
	}
	//savePool <- save
	updatePool <- saveInfo
}

// 曾用名
func historynamefunMysql(tmp map[string]interface{}) {
	query := "SELECT history_name FROM company_history_name WHERE company_id=?"
	info := MysqlTool.SelectBySql(query, util.ObjToString(tmp["_id"]))
	if len(*info) > 0 {
		var names []string
		for _, v := range *info{
			names = append(names, util.ObjToString(v["history_name"]))
		}
		tmp["history_name"] =  strings.Join(names, ",")
	}
}
func historyNameFun(tmp map[string]interface{}) {
	field := bson.M{"history_name": 1}
	info, b := MongoTool1.Find("company_history_name", bson.M{"company_id": tmp["_id"]}, nil, field, false, -1, -1)
	if b && len(*info) > 0 {
		var names []string
		for _, v := range *info{
			names = append(names, util.ObjToString(v["history_name"]))
		}
		tmp["history_name"] =  strings.Join(names, ",")
	}
}

// company_employee 高管
func employeeFunMySql(tmp map[string]interface{})  {
	query := "SELECT employee_name, position, is_history FROM company_employee WHERE company_id=?"
	info := MysqlTool.SelectBySql(query, util.ObjToString(tmp["_id"]))
	if len(*info) > 0 {
		var names []string
		tmp["employees"] =  *info
		for _, v := range *info{
			names = append(names, util.ObjToString(v["employee_name"]))
		}
		tmp["employee_name"] =  strings.Join(names, ",")
	}
}
func employeeFun(tmp map[string]interface{})  {
	field := bson.M{"employee_name": 1, "position": 1, "is_history": 1}
	info, b := MongoTool1.Find("company_employee", bson.M{"company_id": tmp["_id"]}, nil, field, false, -1, -1)
	if b && len(*info) > 0 {
		var names []string
		tmp["employees"] =  *info
		for _, v := range *info{
			names = append(names, util.ObjToString(v["employee_name"]))
		}
		tmp["employee_name"] =  strings.Join(names, ",")
	}
}

// company_partner 合伙人
func partnerFunMySql(tmp map[string]interface{}) {
	query := "SELECT stock_name, stock_type, is_personal, identify_type, identify_no, stock_capital, stock_realcapital, is_history FROM company_partner WHERE company_id=?"
	info := MysqlTool.SelectBySql(query, util.ObjToString(tmp["_id"]))
	if len(*info) > 0 {
		var names []string
		tmp["partners"] =  *info
		for _, v := range *info{
			if util.IntAll(tmp["is_history"]) == 0 {
				names = append(names, util.ObjToString(v["stock_name"]))
			}
		}
		tmp["stock_name"] =  strings.Join(names, ",")
	}
}
func partnerFun(tmp map[string]interface{}) {
	field := bson.M{"stock_name": 1, "stock_type": 1, "is_personal": 1, "identify_type": 1, "identify_no": 1, "stock_capital": 1, "stock_realcapital": 1, "is_history": 1}
	info, b := MongoTool1.Find("company_partner", bson.M{"company_id": tmp["_id"]}, nil, field, false, -1, -1)
	if b && len(*info) > 0 {
		var names []string
		tmp["partners"] =  *info
		for _, v := range *info{
			if util.IntAll(tmp["is_history"]) == 0 {
				names = append(names, util.ObjToString(v["stock_name"]))
			}
		}
		tmp["stock_name"] =  strings.Join(names, ",")
	}
}

// annual_report_base	年报信息
func reportFunMysql(tmp map[string]interface{}) {
	query := "SELECT report_year, company_phone, company_email, zip_code, employee_no, operator_name FROM annual_report_base WHERE company_id=?"
	info := MysqlTool.SelectBySql(query, util.ObjToString(tmp["_id"]))
	if len(*info) > 0 {
		tmp["annual_reports"] =  *info
		year := 0
		phone, email := "", ""
		for _, v := range *info{
			if year < util.IntAll(v["report_year"]) {
				year = util.IntAll(v["report_year"])
				phone = util.ObjToString(v["company_phone"])
				email = util.ObjToString(v["company_email"])
			}
		}
		if year != 0 {
			tmp["company_phone"] = phone
			tmp["company_email"] = email
		}
	}
}
func reportFun(tmp map[string]interface{}) {
	field := bson.M{"report_year": 1, "company_phone": 1, "company_email": 1, "zip_code": 1, "employee_no": 1, "operator_name": 1}
	info, b := MongoTool1.Find("annual_report_base", bson.M{"company_id": tmp["_id"]}, nil, field, false, -1, -1)
	if b && len(*info) > 0 {
		tmp["annual_reports"] =  *info
		year := 0
		phone, email := "", ""
		for _, v := range *info{
			if year < util.IntAll(v["report_year"]) {
				year = util.IntAll(v["report_year"])
				phone = util.ObjToString(v["company_phone"])
				email = util.ObjToString(v["company_email"])
			}
		}
		if year != 0 {
			tmp["company_phone"] = phone
			tmp["company_email"] = email
		}
	}
}

// website_url
func websiteFunMySql(tmp map[string]interface{}) {
	query := "SELECT website_url, website_name, website_type, report_year, is_history FROM annual_report_website WHERE company_id=?"
	info := MysqlTool.SelectBySql(query, util.ObjToString(tmp["_id"]))
	if len(*info) > 0 {
		year := 0
		web := ""
		for _, v := range *info{
			if year < util.IntAll(v["report_year"]) && util.IntAll(tmp["is_history"]) == 0 {
				year = util.IntAll(v["report_year"])
				web = util.ObjToString(v["website_url"])
			}
		}
		if year != 0 {
			tmp["website_url"] =  web
		}
	}
}
func websiteFun(tmp map[string]interface{}) {
	field := bson.M{"website_url": 1, "website_name": 1, "website_type": 1, "report_year": 1, "is_history": 1}
	info, b := MongoTool1.Find("annual_report_website", bson.M{"company_id": tmp["_id"]}, nil, field, false, -1, -1)
	if b && len(*info) > 0 {
		year := 0
		web := ""
		for _, v := range *info{
			if year < util.IntAll(v["report_year"]) && util.IntAll(tmp["is_history"]) == 0 {
				year = util.IntAll(v["report_year"])
				web = util.ObjToString(v["website_url"])
			}
		}
		if year != 0 {
			tmp["website_url"] =  web
		}
	}
}

// company_shortname 企业简称
func getStName(name string) string {
	regnames := regPre.FindStringSubmatch(name)
	lenth := len(regnames)
	if lenth < 1 {
		return ""
	}
	newstr := regnames[lenth-1]

	ch := seg.Cut(newstr, false)
	val := []string{}
	for word := range ch {
		val = append(val, word)
	}
	name2 := ""
	for _, v := range val {
		name2 = name2 + v
		if len([]rune(name2)) >= 4 {
			break
		}
	}
	return name2
}