data_validation/tools/main.go

package main

import (
	"fmt"
	"github.com/tealeg/xlsx"
	"go.mongodb.org/mongo-driver/bson"
	mgo "mongodb"
	"os"
	qu "qfw/util"
	"strings"
	"sync"
	"time"
)

var (
	Sysconfig  map[string]interface{}
	Mgo, MgoJy *mgo.MongodbSim
)

func init() {
	qu.ReadConfig(&Sysconfig)
	MgoJy = &mgo.MongodbSim{
		MongodbAddr: qu.ObjToString(Sysconfig["jy_addr"]),
		Size:        qu.IntAll(Sysconfig["size"]),
		DbName:      qu.ObjToString(Sysconfig["jy_db"]),
	}
	MgoJy.InitPool()
	Mgo = &mgo.MongodbSim{
		MongodbAddr: qu.ObjToString(Sysconfig["addr"]),
		Size:        qu.IntAll(Sysconfig["size"]),
		DbName:      qu.ObjToString(Sysconfig["db"]),
	}
	Mgo.InitPool()
}

func ShowTable() {
	fmt.Println("================================")
	fmt.Println("小工具-数据处理")
	fmt.Println("1、查找数据，打上标识(isOptimization=1)")
	fmt.Println("2、导出数据到文档(判重)")
	fmt.Println("3、导入数据")
	fmt.Println("4、推送数据")
	fmt.Println("0、EXIT")
	fmt.Println("================================")
}

func main() {
	for {
		ShowTable()
		var flag int
		fmt.Print("请输入你的操作:")
		fmt.Scan(&flag)
		if flag == 0 {
			fmt.Println("退出成功")
			os.Exit(0)
		} else if flag == 1 {
			fmt.Println("请输入表名...")
			var coll string
			fmt.Scan(&coll)
			task1(coll)
		} else if flag == 2 {
			fmt.Println("请输入表名...")
			var coll string
			fmt.Scan(&coll)
			task2(coll)
		} else if flag == 3 {
			fmt.Println("请输入表名...")
			var coll, path string
			fmt.Scan(&coll)
			if coll != "" {
				fmt.Println("请输入文档地址...")
				fmt.Scan(&path)
				if path != "" {
					task3(coll, path)
				} else {
					qu.Debug("请输入文档地址...")
				}
			} else {
				qu.Debug("请输入表名...")
			}
		} else if flag == 4 {
			fmt.Println("请输入表名...")
			var coll string
			fmt.Scan(&coll)
			if coll != "" {
				task4(coll)
			}
		}
	}
}

var FieldsMap = map[string]string{
	"是否优选":      "is_push",
	"是否有效标识":    "is_effective",
	"运营商中标标签":   "tagname",
	"主体公司":      "tagname2",
	"中标人":       "s_winner",
	"招标人":       "buyer",
	"中标金额":      "bidamount",
	"是否为多标多包数据": "multipackage",
}

func task1(coll string) {
	if coll != "" {
		sess := Mgo.GetMgoConn()
		defer Mgo.DestoryMongoConn(sess)

		ch := make(chan bool, 2)
		wg := &sync.WaitGroup{}

		//q := bson.M{"isOptimization": 1}
		query := sess.DB(Mgo.DbName).C(coll).Find(nil).Select(nil).Iter()
		count := 0
		for tmp := make(map[string]interface{}); query.Next(&tmp); count++ {
			if count%500 == 0 {
				qu.Debug("current ---", count)
			}
			ch <- true
			wg.Add(1)
			go func(tmp map[string]interface{}) {
				defer func() {
					<-ch
					wg.Done()
				}()
				info := tmp["v_baseinfo"].(map[string]interface{})
				if qu.IntAll(info["publishtime"]) < 1717171200 {
					return
				}
				if (qu.ObjToString(info["subtype"]) == "中标" || qu.ObjToString(info["subtype"]) == "成交" ||
					qu.ObjToString(info["subtype"]) == "合同") && qu.ObjToString(info["rulename"]) != "运营商中标_20230915" &&
					(qu.Float64All(info["bidamount"]) <= 1000 || qu.Float64All(info["bidamount"]) >= 50000000 || qu.ObjToString(info["s_winner"]) == "") {
					//qu.Debug(tmp["_id"])
					if strings.Contains(qu.ObjToString(info["rulename"]), "运营商中标") {
						Mgo.UpdateById(coll, mgo.BsonIdToSId(tmp["_id"]), bson.M{"$set": bson.M{"isOptimization": 1}})
					}
				}
			}(tmp)
			tmp = make(map[string]interface{})
		}
		wg.Wait()
		qu.Debug(fmt.Sprintf("数据更新成功完成"))
	} else {
		qu.Debug("表明不能为空")
	}
	os.Exit(0)
}

func task2(coll string) {
	if coll != "" {
		sess := Mgo.GetMgoConn()
		defer Mgo.DestoryMongoConn(sess)

		file, err := xlsx.OpenFile("./taskexportdata.xlsx")
		if err != nil {
			qu.Debug("Load Excel Model Error")
			return
		}
		sheet := file.Sheets[0]

		q := bson.M{"isOptimization": 1}
		query := sess.DB(Mgo.DbName).C(coll).Find(q).Select(nil).Iter()
		count := 0
		for tmp := make(map[string]interface{}); query.Next(&tmp); count++ {
			baseinfo := tmp["v_baseinfo"].(map[string]interface{})
			if baseinfo["package"] != nil {
				pkg := baseinfo["package"].(map[string]interface{})
				for _, p := range pkg {
					row := sheet.AddRow()
					p1 := p.(map[string]interface{})
					winner := []string{}
					bidamount := float64(0)
					if p1["winner_all"] != nil {
						if all := p1["winner_all"].([]interface{}); all != nil {
							if len(all) > 0 {
								for _, a := range all {
									a1 := a.(map[string]interface{})
									if qu.ObjToString(a1["winner"]) != "" {
										winner = append(winner, qu.ObjToString(a1["winner"]))
									}
									bidamount = qu.Float64All(a1["bidamount"])
								}
							}
						}
					}
					row.AddCell().SetValue(qu.ObjToString(tmp["id"]))
					row.AddCell().SetValue(strings.Join(winner, ","))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["buyer"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["projectcode"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["projectname"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["title"]))
					row.AddCell().SetValue(bidamount)
					row.AddCell().SetValue(qu.ObjToString(baseinfo["area"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["city"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["tagname"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["tagname2"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["href"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["jybxhref"]))
					row.AddCell().SetValue(qu.ObjToString(baseinfo["is_effective"]))
				}
			} else {
				row := sheet.AddRow()
				row.AddCell().SetValue(qu.ObjToString(tmp["id"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["s_winner"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["buyer"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["projectcode"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["projectname"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["title"]))
				row.AddCell().SetValue(baseinfo["bidamount"])
				row.AddCell().SetValue(qu.ObjToString(baseinfo["area"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["city"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["tagname"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["tagname2"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["href"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["jybxhref"]))
				row.AddCell().SetValue(qu.ObjToString(baseinfo["is_effective"]))
			}
		}
		qu.Debug(fmt.Sprintf("推送数据成功，推送成功： %d条", count))
		fname := fmt.Sprintf("./导出数据%d.xlsx", time.Now().Unix())
		qu.Debug("File Name：", fname)
		err = file.Save(fname)
		if err != nil {
			qu.Debug("Save Excel" + fname + "Error")
			return
		}
	} else {
		qu.Debug("表明不能为空")
	}
	os.Exit(0)
}

func task3(coll, path string) {
	if coll != "" && path != "" {
		file, err := xlsx.OpenFile(path)
		if err != nil {
			panic(err)
		}
		sheet := file.Sheets[0]
		count := 0
		idcolnum := -1
		cellFieldName := map[int]string{}
		lastid := ""
		c1 := 0 // 多包第n条数据
		for rn, row := range sheet.Rows {
			update := make(map[string]interface{})
			if rn == 0 {
				for index, cell := range row.Cells {
					if cell.Value == "唯一标识" || cell.Value == "标讯编码（infoID）" { //id所在列
						idcolnum = index
					}
					if v := FieldsMap[cell.Value]; v != "" {
						cellFieldName[index] = v
					}
				}
				if idcolnum == -1 {
					break
				}
				continue
			} else {
				id := row.Cells[idcolnum].String()
				for i, f := range cellFieldName {
					if val := row.Cells[i].Value; val != "" {
						if f == "is_push" {
							update[f] = qu.IntAll(val)
						} else if f == "multipackage" {
							update[fmt.Sprintf("v_baseinfo.%s", f)] = qu.IntAll(val)
						} else if f == "bidamount" {
							update[fmt.Sprintf("v_baseinfo.%s", f)] = qu.Float64All(val)
						} else {
							update[fmt.Sprintf("v_baseinfo.%s", f)] = val
						}
					}
				}
				if qu.IntAll(update["v_baseinfo.multipackage"]) == 1 {
					if c1 == 0 {
						count++
					}
					if c1 > 0 && lastid == id {
						info, _ := Mgo.FindById("f_sourceinfo_chinaunicom_zb_data", id, bson.M{"v_baseinfo.tagname": 1, "v_baseinfo.tagname2": 1, "v_baseinfo.s_winner": 1,
							"v_baseinfo.bidamount": 1, "v_baseinfo.package": 1})
						if len(*info) > 0 {
							baseinfo := (*info)["v_baseinfo"].(map[string]interface{})
							if baseinfo["package"] != nil {
								packageM := baseinfo["package"].(map[string]interface{})
								m := make(map[string]interface{})
								if update["v_baseinfo.bidamount"] != nil {
									m["bidamount"] = qu.Float64All(update["v_baseinfo.bidamount"])
								}
								if update["v_baseinfo.s_winner"] != nil {
									m["s_winner"] = qu.ObjToString(update["v_baseinfo.s_winner"])
								}
								packageM[fmt.Sprint(c1)] = map[string]interface{}{"winner_all": append([]interface{}{}, m)}
								update["v_baseinfo.package"] = packageM
							}
							if s := qu.ObjToString(baseinfo["tagname"]); s != "" {
								update["v_baseinfo.tagname"] = s + "," + qu.ObjToString(update["v_baseinfo.tagname"])
							}
							if s := qu.ObjToString(baseinfo["tagname2"]); s != "" {
								update["v_baseinfo.tagname2"] = s + "," + qu.ObjToString(update["v_baseinfo.tagname2"])
							}
							if s := qu.ObjToString(baseinfo["s_winner"]); s != "" {
								update["v_baseinfo.s_winner"] = s + "," + qu.ObjToString(update["v_baseinfo.s_winner"])
							}
							update["v_baseinfo.bidamount"] = qu.Float64All(update["v_baseinfo.bidamount"]) + qu.Float64All(baseinfo["bidamount"])
							c1++
						}
					} else {
						c1 = 0
						packageM := make(map[string]interface{})
						m := make(map[string]interface{})
						if update["v_baseinfo.bidamount"] != nil {
							m["bidamount"] = qu.Float64All(update["v_baseinfo.bidamount"])
						}
						if update["v_baseinfo.s_winner"] != nil {
							m["s_winner"] = qu.ObjToString(update["v_baseinfo.s_winner"])
						}
						packageM[fmt.Sprint(c1)] = map[string]interface{}{"winner_all": append([]interface{}{}, m)}
						update["v_baseinfo.package"] = packageM
						c1++
					}
				} else {
					count++
					c1 = 0
				}
				lastid = id
				Mgo.Update(coll, bson.M{"id": id}, bson.M{"$set": update}, false, false)
			}
		}
		qu.Debug(fmt.Sprintf("更新数据成功，更新： %d条", count))
	} else {
		qu.Debug(coll, path)
	}
	os.Exit(0)
}

func task4(coll string) {
	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)

	ch := make(chan bool, 5)
	wg := &sync.WaitGroup{}

	var q bson.M
	count := 0
	// 联通中标数据 特殊处理
	q = bson.M{"is_push": 1}
	query := sess.DB(Mgo.DbName).C(coll).Find(q).Select(nil).Iter()
	for tmp := make(map[string]interface{}); query.Next(&tmp); count++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			info := tmp["v_baseinfo"].(map[string]interface{})
			info["createtime"] = time.Now().Unix()
			if qu.ObjToString(info["id"]) == "" {
				info["id"] = tmp["id"]
			}
			info["isOptimization"] = 1
			info["ispanchong"] = 1
			Mgo.Save("tmp_usermail", info)
			Mgo.UpdateById(coll, tmp["_id"], bson.M{"$set": bson.M{"sendflag": true}})

		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	qu.Debug(fmt.Sprintf("推送数据成功，推送成功： %d条", count))
}