es_index_data/createEsIndex/buyertask.go

package main

import (
	"context"
	"database/sql"
	"esindex/config"
	"fmt"
	"go.uber.org/zap"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
	"sync"
	"time"
)

func buyerTest() {

	names := []string{"海南东方新丝路职业学院", "海口旅游职业学院", "海南卫生健康职业学院", "海南科技职业大学", "海南省旅游学校"}

	areaMap, err := loadCodeRegionMap(Mysql.DB)
	if err != nil {
		fmt.Println(err, areaMap)
	}
	for _, na := range names {

		query := fmt.Sprintf(`
 SELECT
               b.name, 
               b.area_code, 
               b.city_code, 
			   b.seo_id,
               t.id,
               t.name_id, 
			   b.company_id,
               t.createtime,
               t.updatetime,
               class.name AS buyerclass 
              
           FROM 
               dws_f_ent_tags AS t 
               LEFT JOIN code_buyerclass AS class ON class.code = t.labelvalues
               LEFT JOIN dws_f_ent_baseinfo AS b ON b.name_id = t.name_id

			WHERE  b.name = '%v' 
        ;
      `, na)

		ctx := context.Background()
		rows, err := Mysql.DB.QueryContext(ctx, query)
		if err != nil {
			log.Info("dealData", zap.Any("QueryContext err", err))
		}

		columns, err := rows.Columns()
		if err != nil {
			log.Info("buyerOnce", zap.Any("rows.Columns", err))
		}

		for rows.Next() {
			scanArgs := make([]interface{}, len(columns))
			values := make([]interface{}, len(columns))
			ret := make(map[string]interface{})
			//MySQL 更新
			update := map[string]interface{}{}

			for k := range values {
				scanArgs[k] = &values[k]
			}

			err = rows.Scan(scanArgs...)
			if err != nil {
				log.Info("buyerOnce", zap.Any("rows.Scan", err))
				break
			}

			for i, col := range values {
				if v, ok := col.([]uint8); ok {
					ret[columns[i]] = string(v)
				} else {
					ret[columns[i]] = col
				}
			}

			name := util.ObjToString(ret["name"])
			//company_id ！= ‘’ 或者在凭安 特殊企业或者在爬虫采购单位里
			if util.ObjToString(ret["company_id"]) != "" || specialNames[name] {
				update["status"] = 1
			} else if ruleBuyer(name) { //不符合条件，排除
				update["status"] = -1
			} else { //默认2，认为可信
				update["status"] = 2
			}

			//1.更新MySQL
			where := map[string]interface{}{
				"name_id": ret["name_id"],
			}
			if len(update) > 0 {
				Mysql.Update("dws_f_ent_tags", where, update)
			}

			//2.生索引,status = 1或者2 才生索引
			if util.IntAll(update["status"]) == 1 || util.IntAll(update["status"]) == 2 {
				data := make(map[string]interface{}, 0)
				data["name"] = name
				data["name_id"] = ret["name_id"]
				if ret["seo_id"] != nil {
					data["seo_id"] = ret["seo_id"]
				}
				data["id"] = ret["name_id"]
				data["buyer_name"] = name
				areaCode := util.ObjToString(ret["area_code"])
				data["province"] = areaMap[areaCode].Province
				cityCode := util.ObjToString(ret["city_code"])
				data["city"] = areaMap[cityCode].City
				data["buyerclass"] = ret["buyerclass"]
				if ret["createtime"] != nil {
					if createtime, ok := ret["createtime"].(time.Time); ok {
						data["createtime"] = createtime.Unix()
						if ret["updatetime"] != nil {
							if updatetime, ok := ret["updatetime"].(time.Time); ok {
								data["updatetime"] = updatetime.Unix()
							}
						} else {
							data["updatetime"] = createtime.Unix()
						}
					}
				}

				sql := fmt.Sprintf(`select count(id) from dws_f_ent_contact where name_id = '%v'`, ret["name_id"])
				counts := Mysql.CountBySql(sql)
				if counts > 0 {
					data["is_contact"] = true
				} else {
					data["is_contact"] = false
				}
				arrEs := make([]map[string]interface{}, 0) //最终生索引数据
				arrEs = append(arrEs, data)
				err = Es.InsertOrUpdate(config.Conf.DB.Es.IndexBuyer, arrEs)
				if err != nil {
					log.Info("buyerOnce", zap.Any("InsertOrUpdate err", err))
				}
				// 华为云新集群，存储标讯、项目、凭安数据
				if config.Conf.DB.Es.Addr3 != "" {
					err = Es3.InsertOrUpdate(config.Conf.DB.Es.IndexBuyer, arrEs)
					if err != nil {
						log.Info("buyerOnce", zap.Any("Addr3 InsertOrUpdate err", err))
					}
				}
			}
		}

		rows.Close()

		if err := rows.Err(); err != nil {
			log.Info("buyerOnce", zap.Any("err", err))
		}

	}

}

// buyerOnce  处理增量数据
func buyerOnce() {
	if len(specialNames) < 1 {
		initSpecialNames()
	}
	areaMap, err := loadCodeRegionMap(Mysql.DB)
	if err != nil {
		log.Info("buyerOnce", zap.Error(err), zap.Any("areaMap", areaMap))
	}

	rowsPerPage := 1000
	now := time.Now()
	tarTime := time.Date(now.Year(), now.Month(), now.Day()-1, 00, 00, 00, 00, time.Local)
	curTime := tarTime.Format("2006-01-02")

	countSql := fmt.Sprintf(`
	 SELECT
              count(t.id)

          FROM
              dws_f_ent_tags AS t
              LEFT JOIN dws_f_ent_baseinfo AS b ON b.name_id = t.name_id

          WHERE  t.createtime >= '%v' OR t.updatetime >= '%v'

          ORDER BY t.id DESC  LIMIT 1
`, curTime, curTime)

	dataCounts := Mysql.CountBySql(countSql)
	if dataCounts > 0 {
		log.Info("buyerOnce", zap.Any(fmt.Sprintf("总数:%s", curTime), dataCounts))
	} else {
		log.Info("buyerOnce", zap.String(curTime, "没有更新数据"))
		return
	}

	lastid, total := 0, 0
	realCount := 0

	for {
		query := fmt.Sprintf(`
 SELECT
               b.name, 
               b.area_code, 
               b.city_code, 
			   b.seo_id,
               t.id,
               t.name_id, 
			   b.company_id,
               t.createtime,
               t.updatetime,
               c.area, 
               c.city, 
               class.name AS buyerclass 
              
           FROM 
               dws_f_ent_tags AS t 
               LEFT JOIN code_buyerclass AS class ON class.code = t.labelvalues
               LEFT JOIN dws_f_ent_baseinfo AS b ON b.name_id = t.name_id 

			WHERE  (t.createtime >= '%v' OR t.updatetime >= '%v') and ( t.id > %d ) 
           ORDER BY t.id ASC
          LIMIT %d;
      `, curTime, curTime, lastid, rowsPerPage)

		ctx := context.Background()
		rows, err := Mysql.DB.QueryContext(ctx, query)
		if err != nil {
			log.Info("dealData", zap.Any("QueryContext err", err))
		}

		if total >= int(dataCounts) {
			log.Info("buyerOnce over", zap.Any("total", total), zap.Any("lastid", lastid))
			break
		}

		columns, err := rows.Columns()
		if err != nil {
			log.Info("buyerOnce", zap.Any("rows.Columns", err))
		}

		for rows.Next() {
			scanArgs := make([]interface{}, len(columns))
			values := make([]interface{}, len(columns))
			ret := make(map[string]interface{})
			//MySQL 更新
			update := map[string]interface{}{}

			for k := range values {
				scanArgs[k] = &values[k]
			}

			err = rows.Scan(scanArgs...)
			if err != nil {
				log.Info("buyerOnce", zap.Any("rows.Scan", err))
				break
			}

			for i, col := range values {
				if v, ok := col.([]uint8); ok {
					ret[columns[i]] = string(v)
				} else {
					ret[columns[i]] = col
				}
			}

			total++
			if total%100 == 0 {
				log.Info("buyerOnce", zap.Int("current total", total))
			}

			lastid = util.IntAll(ret["id"])
			name := util.ObjToString(ret["name"])
			//company_id ！= ‘’ 或者在凭安 特殊企业或者在爬虫采购单位里
			if util.ObjToString(ret["company_id"]) != "" || specialNames[name] {
				realCount++
				update["status"] = 1
			} else if ruleBuyer(name) { //不符合条件，排除
				update["status"] = -1
			} else { //默认2，认为可信
				realCount++
				update["status"] = 2
			}

			//1.更新MySQL
			where := map[string]interface{}{
				"name_id": ret["name_id"],
			}
			if len(update) > 0 {
				Mysql.Update("dws_f_ent_tags", where, update)
			}

			//2.生索引,status = 1或者2 才生索引
			if util.IntAll(update["status"]) == 1 || util.IntAll(update["status"]) == 2 {
				data := make(map[string]interface{}, 0)
				data["name"] = name
				data["name_id"] = ret["name_id"]
				if ret["seo_id"] != nil {
					data["seo_id"] = ret["seo_id"]
				}
				data["id"] = ret["name_id"]
				data["buyer_name"] = name
				//data["province"] = ret["area"]
				//data["city"] = ret["city"]
				areaCode := util.ObjToString(ret["area_code"])
				data["province"] = areaMap[areaCode].Province
				cityCode := util.ObjToString(ret["city_code"])
				data["city"] = areaMap[cityCode].City

				data["buyerclass"] = ret["buyerclass"]
				if ret["createtime"] != nil {
					if createtime, ok := ret["createtime"].(time.Time); ok {
						data["createtime"] = createtime.Unix()
						if ret["updatetime"] != nil {
							if updatetime, ok := ret["updatetime"].(time.Time); ok {
								data["updatetime"] = updatetime.Unix()
							}
						} else {
							data["updatetime"] = createtime.Unix()
						}
					}
				}

				sql := fmt.Sprintf(`select count(id) from dws_f_ent_contact where name_id = '%v'`, ret["name_id"])
				counts := Mysql.CountBySql(sql)
				if counts > 0 {
					data["is_contact"] = true
				} else {
					data["is_contact"] = false
				}
				arrEs := make([]map[string]interface{}, 0) //最终生索引数据
				arrEs = append(arrEs, data)
				err = Es.InsertOrUpdate(config.Conf.DB.Es.IndexBuyer, arrEs)
				if err != nil {
					log.Info("buyerOnce", zap.Any("InsertOrUpdate err", err))
				}
				// 华为云新集群，存储标讯、项目、凭安数据
				if config.Conf.DB.Es.Addr3 != "" {
					err = Es3.InsertOrUpdate(config.Conf.DB.Es.IndexBuyer, arrEs)
					if err != nil {
						log.Info("buyerOnce", zap.Any("Addr3 InsertOrUpdate err", err))
					}
				}
			}
		}

		rows.Close()

		if err := rows.Err(); err != nil {
			log.Info("buyerOnce", zap.Any("err", err))
		}

	}

	log.Info("buyerOnce", zap.Int("结束，总数是：", total), zap.Int("realCount", realCount))
}

// dealSpecialNames 拿到凭安 特企数据 以及爬虫采购单位数据
func initSpecialNames() {
	sess := MgoS.GetMgoConn()
	defer MgoS.DestoryMongoConn(sess)

	query := sess.DB(MgoS.DbName).C("special_enterprise").Find(nil).Iter()
	count := 0
	log.Info("dealSpecialNames", zap.String("special_enterprise", "special_enterprise"))

	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			} else {
				specialNames[tmp["company_name"].(string)] = true
			}
		}
	}

	//2.
	time.Sleep(time.Second)
	log.Info("dealSpecialNames", zap.String("special_foundation", "special_foundation"))
	query2 := sess.DB("mixdata").C("special_foundation").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query2.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			} else {
				specialNames[tmp["company_name"].(string)] = true
			}
		}
	}

	//3.
	time.Sleep(time.Second)
	log.Info("dealSpecialNames", zap.String("special_gov_unit", "special_gov_unit"))
	query3 := sess.DB("mixdata").C("special_gov_unit").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query3.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			} else {
				specialNames[tmp["company_name"].(string)] = true
			}
		}
	}

	//4.
	time.Sleep(time.Second)
	log.Info("dealSpecialNames", zap.String("special_hongkong_company", "special_hongkong_company"))
	query4 := sess.DB("mixdata").C("special_hongkong_company").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query4.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			} else {
				specialNames[tmp["company_name"].(string)] = true
			}
		}
	}

	//5.
	time.Sleep(time.Second)
	log.Info("dealSpecialNames", zap.String("special_hongkong_company_history", "special_hongkong_company_history"))
	query5 := sess.DB("mixdata").C("special_hongkong_company_history").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query5.Next(tmp); count++ {
		if tmp["use_name"] != nil && tmp["use_name"] != "" {
			if specialNames[tmp["use_name"].(string)] {
				continue
			}
			specialNames[tmp["use_name"].(string)] = true
		}
	}
	//6.
	time.Sleep(time.Second)
	log.Info("dealSpecialNames", zap.String("special_law_office", "special_law_office"))
	query6 := sess.DB("mixdata").C("special_law_office").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query6.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			}
			specialNames[tmp["company_name"].(string)] = true
		}
	}

	//7.

	log.Info("dealSpecialNames", zap.String("special_social_organ", "special_social_organ"))
	query7 := sess.DB("mixdata").C("special_social_organ").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query7.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			}
			specialNames[tmp["company_name"].(string)] = true
		}
	}

	log.Info("dealSpecialNames", zap.String("special_trade_union", "special_trade_union"))
	//8.
	query8 := sess.DB("mixdata").C("special_trade_union").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query8.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			}
			specialNames[tmp["company_name"].(string)] = true
		}
	}
	//9. 爬虫采集的采购单位
	query9 := sess.DB("mixdata").C("special_buyer_name").Find(nil).Iter()
	count = 0
	for tmp := make(map[string]interface{}); query9.Next(tmp); count++ {
		if tmp["company_name"] != nil && tmp["company_name"] != "" {
			if specialNames[tmp["company_name"].(string)] {
				continue
			}
			specialNames[tmp["company_name"].(string)] = true
		}
	}

}

// buyerAll 采购单位全量数据
func buyerAll() {
	arrEs := make([]map[string]interface{}, 0) //最终生索引数据
	winerEsLock := &sync.Mutex{}
	//if len(specialNames) < 1 {
	//	initSpecialNames()
	//}

	countSql := fmt.Sprintf(`SELECT	count(id)	FROM dws_f_ent_tags `)
	dataCounts := Mysql.CountBySql(countSql)
	if dataCounts > 0 {
		log.Info("buyerAll", zap.Any("采购单位全量：", dataCounts))
	} else {
		log.Info("buyerAll", zap.String("采购单位全量是0", "没有更新数据"))
		return
	}

	finalId := 0
	lastSql := fmt.Sprintf(`SELECT id  FROM	dws_f_ent_tags  ORDER BY id DESC  LIMIT 1`)
	lastInfo := Mysql.SelectBySql(lastSql)
	if len(*lastInfo) > 0 {
		finalId = util.IntAll((*lastInfo)[0]["id"])
	} else {
		log.Info("buyerAll", zap.String("获取最大ID失败", "没有数据"))
		return
	}

	log.Info("buyerAll", zap.Int("finalId", finalId))
	buyerPool := make(chan bool, 10) //控制线程数
	wg := &sync.WaitGroup{}

	lastid, total := 0, 0
	realCount := 0

	for {
		query := fmt.Sprintf(`
 SELECT
               b.name, 
			   b.seo_id,
               t.id,
               t.status,
               t.name_id, 
			   b.company_id,
               t.createtime,
               t.updatetime,
               c.area, 
               c.city, 
               class.name AS buyerclass 
              
           FROM 
               dws_f_ent_tags AS t 
               LEFT JOIN code_buyerclass AS class ON class.code = t.labelvalues
               LEFT JOIN dws_f_ent_baseinfo AS b ON b.name_id = t.name_id
               LEFT JOIN code_area AS c ON b.city_code = c.code 

			WHERE  t.id > %d and t.status > 0
           ORDER BY t.id ASC
          LIMIT %d;
      `, lastid, 1000)

		ctx := context.Background()
		rows, err := Mysql.DB.QueryContext(ctx, query)
		if err != nil {
			log.Info("buyerAll", zap.Any("QueryContext err", err))
		}

		if finalId == lastid {
			log.Info("buyerAll over", zap.Any("total", total), zap.Any("lastid", lastid))
			break
		}

		columns, err := rows.Columns()
		if err != nil {
			log.Info("buyerAll", zap.Any("rows.Columns", err))
		}

		for rows.Next() {
			scanArgs := make([]interface{}, len(columns))
			values := make([]interface{}, len(columns))
			ret := make(map[string]interface{})

			for k := range values {
				scanArgs[k] = &values[k]
			}

			err = rows.Scan(scanArgs...)
			if err != nil {
				log.Info("buyerAll", zap.Any("rows.Scan", err))
				break
			}

			for i, col := range values {
				if v, ok := col.([]uint8); ok {
					ret[columns[i]] = string(v)
				} else {
					ret[columns[i]] = col
				}
			}

			lastid = util.IntAll(ret["id"])
			total++
			if total%2000 == 0 {
				log.Info("buyerAll", zap.Int("current total", total), zap.Int("lastid", lastid))
			}

			buyerPool <- true
			wg.Add(1)

			go func(tmp map[string]interface{}) {
				defer func() {
					<-buyerPool
					wg.Done()
				}()
				name := util.ObjToString(tmp["name"])
				//MySQL 更新
				//update := map[string]interface{}{}
				//company_id ！= ‘’ 或者在凭安 特殊企业或者在爬虫采购单位里
				//if util.ObjToString(tmp["company_id"]) != "" || specialNames[name] {
				//	realCount++
				//	update["status"] = 1
				//} else if ruleBuyer(name) { //不符合条件，排除
				//	update["status"] = -1
				//} else { //默认2，认为可信
				//	realCount++
				//	update["status"] = 2
				//}
				////1.更新MySQL
				//where := map[string]interface{}{
				//	"name_id": tmp["name_id"],
				//}
				//if len(update) > 0 {
				//	Mysql.Update("dws_f_ent_tags", where, update)
				//}

				//2.生索引,status = 1或者2 才生索引
				if util.IntAll(tmp["status"]) == 1 || util.IntAll(tmp["status"]) == 2 {
					data := make(map[string]interface{}, 0)
					data["name"] = name
					data["name_id"] = tmp["name_id"]
					if ret["seo_id"] != nil {
						data["seo_id"] = tmp["seo_id"]
					}
					data["id"] = tmp["name_id"]
					data["buyer_name"] = name
					data["province"] = tmp["area"]
					data["city"] = tmp["city"]
					data["buyerclass"] = tmp["buyerclass"]
					if ret["createtime"] != nil {
						if createtime, ok := tmp["createtime"].(time.Time); ok {
							data["createtime"] = createtime.Unix()
							if ret["updatetime"] != nil {
								if updatetime, ok := tmp["updatetime"].(time.Time); ok {
									data["updatetime"] = updatetime.Unix()
								}
							} else {
								data["updatetime"] = createtime.Unix()
							}
						}
					}

					sql := fmt.Sprintf(`select count(id) from dws_f_ent_contact where name_id = '%v'`, tmp["name_id"])
					counts := Mysql.CountBySql(sql)
					if counts > 0 {
						data["is_contact"] = true
					} else {
						data["is_contact"] = false
					}

					//写入es
					winerEsLock.Lock()
					arrEs = append(arrEs, data)
					if len(arrEs) >= EsBulkSize {
						tmps := arrEs
						Es.BulkSave(config.Conf.DB.Es.IndexBuyer, tmps)
						arrEs = []map[string]interface{}{}
					}
					winerEsLock.Unlock()
				}

			}(ret)

		}

		if err := rows.Err(); err != nil {
			log.Info("buyerAll", zap.Any("rows.Err()", err))
		}

		err = rows.Close()
		if err != nil {
			log.Info("buyerAll", zap.Any("rows.Close() err", err))
		}

		wg.Wait()
	}

	if len(arrEs) > 0 {
		tmps := arrEs
		Es.BulkSave(config.Conf.DB.Es.IndexBuyer, tmps)
		arrEs = []map[string]interface{}{}
	}

	log.Info("buyerAll", zap.Int("结束，总数是：", total), zap.Int("realCount", realCount))

}

type RegionInfo struct {
	Province string
	City     string
	District string
}

func loadCodeRegionMap(db *sql.DB) (map[string]RegionInfo, error) {
	query := `SELECT code, area, city, district FROM code_area`
	rows, err := db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	codeMap := make(map[string]RegionInfo)
	for rows.Next() {
		var code, area, city, district sql.NullString
		if err := rows.Scan(&code, &area, &city, &district); err != nil {
			return nil, err
		}
		codeStr := code.String

		codeMap[codeStr] = RegionInfo{
			Province: area.String,
			City:     city.String,
			District: district.String,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return codeMap, nil
}