es_index_data/udpcreateindex/init.go

package main

import (
	util "app.yhyue.com/data_processing/common_utils"
	"app.yhyue.com/data_processing/common_utils/elastic"
	"app.yhyue.com/data_processing/common_utils/mongodb"
	"app.yhyue.com/data_processing/common_utils/udp"
	"createindex/oss"
	"fmt"
	"net"
	"strings"
	"time"
)

var (
	Sysconfig map[string]interface{}

	biddingMgo  *mongodb.MongodbSim
	extractMgo  *mongodb.MongodbSim
	qyxyMgo     *mongodb.MongodbSim
	projectMgo  *mongodb.MongodbSim
	standardMgo *mongodb.MongodbSim

	Es1 *elastic.Elastic
	Es2 *elastic.Elastic

	currentColl string
	udpclient   udp.UdpClient //udp对象
	updport     string
	multiIndex  []string

	biddingMgoFields      []string
	biddingEsFields       map[string]interface{}
	projectinfoFields     map[string]interface{}
	purchasinglistFields  map[string]interface{}
	procurementlisFields  map[string]interface{}
	winnerorderlistFields map[string]interface{}

	updateBiddingPool chan []map[string]interface{}
	updateExtractPool chan []map[string]interface{}
	saveEsPool        chan map[string]interface{}
	saveEsAllPool     chan map[string]interface{}
	saveEsElsePool    chan map[string]interface{}
	saveProjectEsPool chan map[string]interface{}
	updateBiddingSp   chan bool
	updateExtractSp   chan bool
	saveEsSp          chan bool
	saveEsAllSp       chan bool
	saveEsElseSp      chan bool
	saveProjectSp     chan bool

	JyUdpAddr *net.UDPAddr

	esAddr, esNode string

	FilterKeyword []string              //正文竟品关键词过滤
	ProvinceDict  map[string][]Province //省份-map
	CityDict      map[string][]City     //城市-map
	DistrictDict  map[string][]District //区县-map

	bidding, extract, qyxy, project, standard, buyer, winner, biddingIndex map[string]interface{}

	MgoBulkSize  = 200   // mgo批量保存大小
	EsBulkSize   = 200   // es批量保存大小
	detailLength = 50000 // es保存detail长度
	fileLength   = 50000 // es保存附件文本长度
	pscopeLength = 32766 // projectscope长度
)

var StopFlag = false // 程序生索引停止标志

func init() {
	util.ReadConfig(&Sysconfig)
	updport, _ = Sysconfig["updport"].(string)

	bidding, _ = Sysconfig["mgo_bidding"].(map[string]interface{})
	extract, _ = Sysconfig["mgo_extract"].(map[string]interface{})
	qyxy, _ = Sysconfig["mgo_qyxy"].(map[string]interface{})
	project, _ = Sysconfig["mgo_project"].(map[string]interface{})
	standard, _ = Sysconfig["standard"].(map[string]interface{})
	buyer, _ = Sysconfig["buyer"].(map[string]interface{})
	winner, _ = Sysconfig["winner"].(map[string]interface{})

	currentColl = util.ObjToString(bidding["collect"])
	biddingIndex, _ = Sysconfig["bidding_index"].(map[string]interface{})

	biddingMgo = &mongodb.MongodbSim{
		MongodbAddr: bidding["addr"].(string),
		Size:        util.IntAllDef(bidding["pool"], 5),
		DbName:      bidding["db"].(string),
		UserName:    bidding["uname"].(string),
		Password:    bidding["upwd"].(string),
	}
	biddingMgo.InitPool()
	extractMgo = &mongodb.MongodbSim{
		MongodbAddr: extract["addr"].(string),
		Size:        util.IntAllDef(extract["pool"], 5),
		DbName:      extract["db"].(string),
	}
	extractMgo.InitPool()
	qyxyMgo = &mongodb.MongodbSim{
		MongodbAddr: qyxy["addr"].(string),
		Size:        util.IntAllDef(qyxy["pool"], 5),
		DbName:      qyxy["db"].(string),
	}
	qyxyMgo.InitPool()
	projectMgo = &mongodb.MongodbSim{
		MongodbAddr: project["addr"].(string),
		Size:        util.IntAllDef(project["pool"], 5),
		DbName:      project["db"].(string),
		//UserName:    bidding["uname"].(string),
		//Password:    bidding["upwd"].(string),
	}
	projectMgo.InitPool()
	standardMgo = &mongodb.MongodbSim{
		MongodbAddr: standard["addr"].(string),
		Size:        util.IntAllDef(standard["pool"], 5),
		DbName:      standard["db"].(string),
		UserName:    bidding["uname"].(string),
		Password:    bidding["upwd"].(string),
	}
	standardMgo.InitPool()

	econf1 := Sysconfig["elastic_1"].(map[string]interface{})
	Es1 = &elastic.Elastic{
		S_esurl: econf1["addr"].(string),
		I_size:  util.IntAllDef(econf1["pool"], 5),
	}
	Es1.InitElasticSize()

	econf2 := Sysconfig["elastic_2"].(map[string]interface{})
	Es2 = &elastic.Elastic{
		S_esurl: econf2["addr"].(string),
		I_size:  util.IntAllDef(econf2["pool"], 5),
	}
	Es2.InitElasticSize()

	if mi := util.ObjToString(biddingIndex["multiIndex"]); mi != "" {
		multiIndex = strings.Split(mi, ",")
	}
	util.Debug(multiIndex)
	biddingMgoFields = strings.Split(biddingIndex["mgofields"].(string), ",")
	biddingEsFields = biddingIndex["esfieldsmap"].(map[string]interface{})             // bidding es字段
	projectinfoFields = biddingIndex["projectinfomap"].(map[string]interface{})        // projectinfo
	purchasinglistFields = biddingIndex["purchasinglistmap"].(map[string]interface{})  //采购清单
	procurementlisFields = biddingIndex["procurementlistmap"].(map[string]interface{}) //采购意向
	winnerorderlistFields = biddingIndex["winnerordermap"].(map[string]interface{})    //中标候选

	FilterKeyword = util.ObjArrToStringArr(Sysconfig["filter-keyword"].([]interface{}))
	initCheckCity()
	//初始化oss
	oss.InitOss()
	InitFileInfo()

	m := Sysconfig["jyfb_udp"].(map[string]interface{})
	JyUdpAddr = &net.UDPAddr{
		IP:   net.ParseIP(m["addr"].(string)),
		Port: util.IntAll(m["port"]),
	}

	updateBiddingPool = make(chan []map[string]interface{}, 5000)
	updateBiddingSp = make(chan bool, 5)
	updateExtractPool = make(chan []map[string]interface{}, 5000)
	updateExtractSp = make(chan bool, 5)
	saveEsPool = make(chan map[string]interface{}, 5000)
	saveEsSp = make(chan bool, 5)
	saveEsAllPool = make(chan map[string]interface{}, 5000)
	saveEsAllSp = make(chan bool, 5)
	saveEsElsePool = make(chan map[string]interface{}, 5000)
	saveEsElseSp = make(chan bool, 5)
	saveProjectEsPool = make(chan map[string]interface{}, 5000)
	saveProjectSp = make(chan bool, 5)
}

func NewTk(m map[string]interface{}) *TaskInfo {
	p := &TaskInfo{
		stype:       m["stype"].(string),
		currentTime: time.Now().Unix(),
	}
	return p
}

type TaskInfo struct {
	stype         string
	thread        int
	currentTime   int64
	bidding_count int
	extract_count int
	es_count      int
}

type Province struct {
	P_Name string
}
type City struct {
	P_Name string
	C_Name string
}
type District struct {
	P_Name string
	C_Name string
	D_Name string
}

//初始化城市
func initCheckCity() {
	//初始化-城市配置
	ProvinceDict = make(map[string][]Province, 0)
	CityDict = make(map[string][]City, 0)
	DistrictDict = make(map[string][]District, 0)

	q := map[string]interface{}{
		"town_code": map[string]interface{}{
			"$exists": 0,
		},
	}
	sess := standardMgo.GetMgoConn()
	defer standardMgo.DestoryMongoConn(sess)
	it := sess.DB(standardMgo.DbName).C(util.ObjToString(standard["coll_area"])).Find(&q).Iter()
	total := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
		district_code := util.IntAll(tmp["district_code"])
		city_code := util.IntAll(tmp["city_code"])
		if district_code > 0 {
			province := util.ObjToString(tmp["province"])
			city := util.ObjToString(tmp["city"])
			district := util.ObjToString(tmp["district"])
			data := District{province, city, district}
			if DistrictDict[district] == nil {
				DistrictDict[district] = []District{data}
			} else {
				arr := DistrictDict[district]
				arr = append(arr, data)
				DistrictDict[district] = arr
			}
		} else {
			if city_code > 0 {
				province := util.ObjToString(tmp["province"])
				city := util.ObjToString(tmp["city"])
				data := City{province, city}
				if CityDict[city] == nil {
					CityDict[city] = []City{data}
				} else {
					arr := CityDict[city]
					arr = append(arr, data)
					CityDict[city] = arr
				}
			} else {
				province := util.ObjToString(tmp["province"])
				data := Province{province}
				if ProvinceDict[province] == nil {
					ProvinceDict[province] = []Province{data}
				} else {
					arr := ProvinceDict[province]
					arr = append(arr, data)
					ProvinceDict[province] = arr
				}
			}
		}
		tmp = make(map[string]interface{})
	}
	util.Debug(fmt.Sprintf("城市配置加载完毕...省~%d 市~%d 区~%d", len(ProvinceDict), len(CityDict), len(DistrictDict)))
}