|
|
@@ -210,16 +210,6 @@ func QyxyStandard() bool {
|
|
|
esMap[field] = "其他"
|
|
|
}
|
|
|
}
|
|
|
- } else if field == "history_name" {
|
|
|
- var nameArr []string
|
|
|
- for _, v := range strings.Split(qu.ObjToString(tmp[field]), "；") {
|
|
|
- if v != "" {
|
|
|
- nameArr = append(nameArr, v)
|
|
|
- }
|
|
|
- }
|
|
|
- if len(nameArr) > 0 {
|
|
|
- esMap[field] = nameArr
|
|
|
- }
|
|
|
} else if strings.Contains(field, "date") || strings.Contains(field, "time") { //时间处理
|
|
|
if tmp[field] != nil {
|
|
|
if timeTmp, ok := tmp[field].(primitive.DateTime); ok {
|
|
|
@@ -384,7 +374,6 @@ func QyxyStandard() bool {
|
|
|
continue
|
|
|
}
|
|
|
mgoMap[k] = v
|
|
|
-
|
|
|
}
|
|
|
//es数据过滤
|
|
|
EsSaveFlag := true
|
|
|
@@ -398,6 +387,12 @@ func QyxyStandard() bool {
|
|
|
EsSaveFlag = false
|
|
|
}
|
|
|
}
|
|
|
+ if EsSaveFlag {
|
|
|
+ status := qu.ObjToString(esMap["company_status"])
|
|
|
+ if status != "正常" {
|
|
|
+ EsSaveFlag = false
|
|
|
+ }
|
|
|
+ }
|
|
|
if EsSaveFlag {
|
|
|
credit_no := strings.TrimSpace(qu.ObjToString(esMap["credit_no"]))
|
|
|
company_code := strings.TrimSpace(qu.ObjToString(esMap["company_code"]))
|
|
|
@@ -405,14 +400,24 @@ func QyxyStandard() bool {
|
|
|
EsSaveFlag = false
|
|
|
}
|
|
|
}
|
|
|
- // qu.Debug("mgoMap---", mgoMap)
|
|
|
- // return
|
|
|
lock.Lock()
|
|
|
if EsSaveFlag {
|
|
|
+ if esMap["history_name"] != nil {
|
|
|
+ var nameArr []string
|
|
|
+ for _, v := range strings.Split(qu.ObjToString(esMap["history_name"]), "；") {
|
|
|
+ if v != "" {
|
|
|
+ nameArr = append(nameArr, v)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if len(nameArr) > 0 {
|
|
|
+ esMap["history_name"] = nameArr
|
|
|
+ }
|
|
|
+ }
|
|
|
EsSaveCache <- esMap //过滤后数据保存
|
|
|
}
|
|
|
EsSaveAllCache <- esMap //所有数据保存
|
|
|
update = append(update, map[string]interface{}{"$set": mgoMap})
|
|
|
+ SaveHistoryName(tmp)
|
|
|
if len(update) == 2 {
|
|
|
arr = append(arr, update)
|
|
|
}
|
|
|
@@ -436,7 +441,7 @@ func QyxyStandard() bool {
|
|
|
}
|
|
|
|
|
|
//所有企业数据标准化
|
|
|
-func HistoryQyxyStandard() bool {
|
|
|
+func HistoryQyxyStandard() {
|
|
|
qu.Debug("--------History--------")
|
|
|
defer qu.Catch()
|
|
|
sess := Mgo.GetMgoConn()
|
|
|
@@ -445,16 +450,13 @@ func HistoryQyxyStandard() bool {
|
|
|
pool := make(chan bool, 20) //控制线程数
|
|
|
wg := &sync.WaitGroup{}
|
|
|
lock := &sync.Mutex{} //控制读写
|
|
|
- // arr := [][]map[string]interface{}{}
|
|
|
- // count, _ := sess.DB(Dbname).C(Dbcoll).Find(nil).Count()
|
|
|
- // log.Println("共查询:", count, "条")
|
|
|
- // if count == 0 {
|
|
|
- // return false
|
|
|
- // }
|
|
|
+ arr := [][]map[string]interface{}{}
|
|
|
+ //count, _ := sess.DB(Dbname).C(Dbcoll).Find(nil).Count()
|
|
|
+ //log.Println("共查询:", count, "条")
|
|
|
it := sess.DB(Dbname).C(Dbcoll).Find(nil).Iter()
|
|
|
sum := 0
|
|
|
for tmp := make(map[string]interface{}); it.Next(&tmp); sum++ {
|
|
|
- if sum%10000 == 0 {
|
|
|
+ if sum%100 == 0 {
|
|
|
log.Println("current:", sum)
|
|
|
}
|
|
|
pool <- true
|
|
|
@@ -657,8 +659,8 @@ func HistoryQyxyStandard() bool {
|
|
|
tmpArrMgo := []map[string]interface{}{}
|
|
|
if annual_reports, ok := tmp["annual_reports"].([]interface{}); ok && len(annual_reports) > 0 {
|
|
|
for _, annual_report := range annual_reports {
|
|
|
- tmpMapMgo := map[string]interface{}{}
|
|
|
- tmpMap := map[string]interface{}{}
|
|
|
+ tmpMapMgo := map[string]interface{}{} //记录每个年报信息标准化到mgo的数据
|
|
|
+ tmpMap := map[string]interface{}{} //只记录每个年报信息的company_email和company_phone
|
|
|
report_year := ""
|
|
|
m := annual_report.(map[string]interface{})
|
|
|
for i, tmpArr := range AnnualReportsArr {
|
|
|
@@ -717,15 +719,75 @@ func HistoryQyxyStandard() bool {
|
|
|
esMap[k] = v
|
|
|
}
|
|
|
}
|
|
|
+ //合并
|
|
|
+ for k, v := range esMap {
|
|
|
+ if k == "partners" {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ mgoMap[k] = v
|
|
|
+ }
|
|
|
+ //es数据过滤
|
|
|
+ EsSaveFlag := true
|
|
|
+ company_name := qu.ObjToString(esMap["company_name"])
|
|
|
+ if len([]rune(company_name)) < 8 {
|
|
|
+ EsSaveFlag = false
|
|
|
+ }
|
|
|
+ if EsSaveFlag {
|
|
|
+ company_type := qu.ObjToString(esMap["company_type"])
|
|
|
+ if company_type == "" || company_type == "个体工商户" {
|
|
|
+ EsSaveFlag = false
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if EsSaveFlag {
|
|
|
+ status := qu.ObjToString(esMap["company_status"])
|
|
|
+ if status != "正常" {
|
|
|
+ EsSaveFlag = false
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if EsSaveFlag {
|
|
|
+ credit_no := strings.TrimSpace(qu.ObjToString(esMap["credit_no"]))
|
|
|
+ company_code := strings.TrimSpace(qu.ObjToString(esMap["company_code"]))
|
|
|
+ if credit_no == "" && company_code == "" {
|
|
|
+ EsSaveFlag = false
|
|
|
+ }
|
|
|
+ }
|
|
|
lock.Lock()
|
|
|
- EsSaveCache <- esMap
|
|
|
+ if EsSaveFlag {
|
|
|
+ if esMap["history_name"] != nil {
|
|
|
+ var nameArr []string
|
|
|
+ for _, v := range strings.Split(qu.ObjToString(esMap["history_name"]), "；") {
|
|
|
+ if v != "" {
|
|
|
+ nameArr = append(nameArr, v)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if len(nameArr) > 0 {
|
|
|
+ esMap["history_name"] = nameArr
|
|
|
+ }
|
|
|
+ }
|
|
|
+ EsSaveCache <- esMap //过滤后数据保存
|
|
|
+ }
|
|
|
+ EsSaveAllCache <- esMap //所有数据保存
|
|
|
+ SaveHistoryName(tmp)
|
|
|
+ update = append(update, map[string]interface{}{"$set": mgoMap})
|
|
|
+ if len(update) == 2 {
|
|
|
+ arr = append(arr, update)
|
|
|
+ }
|
|
|
+ if len(arr) > 500 {
|
|
|
+ tmps := arr
|
|
|
+ Mgo.UpSertBulk(Savecoll, tmps...)
|
|
|
+ arr = [][]map[string]interface{}{}
|
|
|
+ }
|
|
|
lock.Unlock()
|
|
|
}(tmp)
|
|
|
tmp = make(map[string]interface{})
|
|
|
}
|
|
|
wg.Wait()
|
|
|
+ lock.Lock()
|
|
|
+ if len(arr) > 0 {
|
|
|
+ Mgo.UpSertBulk(Savecoll, arr...)
|
|
|
+ }
|
|
|
+ lock.Unlock()
|
|
|
log.Println("Run Over...Count:", sum)
|
|
|
- return true
|
|
|
}
|
|
|
|
|
|
//过滤后数据存库
|
|
|
@@ -765,6 +827,34 @@ func SaveEs() {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+func SaveHistoryName(tmp map[string]interface{}) {
|
|
|
+ if qu.ObjToString(tmp["company_name"]) != "" {
|
|
|
+ set := make(map[string]interface{})
|
|
|
+ set["company_name"] = qu.ObjToString(tmp["company_name"])
|
|
|
+ set["credit_no"] = tmp["credit_no"]
|
|
|
+ set["org_code"] = tmp["org_code"]
|
|
|
+ set["company_id"] = tmp["company_id"]
|
|
|
+ set["company_type"] = tmp["company_type"]
|
|
|
+ set["company_status"] = tmp["company_status"]
|
|
|
+ set["company_code"] = tmp["company_code"]
|
|
|
+ Mgo.Update("qyxy_historyname", map[string]interface{}{"company_name": qu.ObjToString(tmp["company_name"])}, map[string]interface{}{"$set": set}, true, false)
|
|
|
+ if qu.ObjToString(tmp["history_name"]) != "" {
|
|
|
+ for _, v := range strings.Split(qu.ObjToString(tmp["history_name"]), "；") {
|
|
|
+ if v != "" {
|
|
|
+ set["company_name"] = v
|
|
|
+ set["credit_no"] = tmp["credit_no"]
|
|
|
+ set["org_code"] = tmp["org_code"]
|
|
|
+ set["company_id"] = tmp["company_id"]
|
|
|
+ set["company_type"] = tmp["company_type"]
|
|
|
+ set["company_status"] = tmp["company_status"]
|
|
|
+ set["company_code"] = tmp["company_code"]
|
|
|
+ Mgo.Update("qyxy_historyname", map[string]interface{}{"company_name": v}, map[string]interface{}{"$set": set}, true, false)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
//所有数据存库
|
|
|
func SaveAllEs() {
|
|
|
log.Println("Es SaveAll...")
|
|
|
@@ -806,13 +896,13 @@ func InitAddress() {
|
|
|
log.Println("Init Address...")
|
|
|
AddressMap = map[string]*City{}
|
|
|
AddressOldMap = map[string]*City{}
|
|
|
- address, _ := Mgo.Find("address", nil, nil, nil, false, -1, -1)
|
|
|
+ address, _ := Mgo.Find("address_new_2020", nil, nil, nil, false, -1, -1)
|
|
|
for _, tmp := range *address {
|
|
|
code := qu.ObjToString(tmp["code"])
|
|
|
- // codeLen := len(code)
|
|
|
- // if t_code := CodeMap[codeLen]; t_code != "" { //新的address表补齐code
|
|
|
- // code = code + t_code
|
|
|
- // }
|
|
|
+ codeLen := len(code)
|
|
|
+ if t_code := CodeMap[codeLen]; t_code != "" { //新的address表补齐code
|
|
|
+ code = code + t_code
|
|
|
+ }
|
|
|
remark := fmt.Sprint(tmp["Remarks"])
|
|
|
city := &City{}
|
|
|
tmpjson, err := json.Marshal(tmp)
|
