data_validation/isrepeat/main.go

package main

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

var (
	Mgo *mgo.MongodbSim
)

func init() {
	Mgo = &mgo.MongodbSim{
		MongodbAddr: "127.0.0.1:27086", // 127.0.0.1:27084
		Size:        5,
		DbName:      "jyqykhfw",
	}
	Mgo.InitPool()
}

func ShowTable() {
	fmt.Println("================================")
	fmt.Println("小工具-数据处理")
	fmt.Println("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 p string
			fmt.Scan(&p)
			if p != "" {
				task1(p)
			}
		} else if flag == 2 {
			fmt.Println("导入推送表数据.")
			task2()
		} else if flag == 3 {
			fmt.Println("请输入文档地址，进行数据判重.")
			var p string
			fmt.Scan(&p)
			if p != "" {
				task3()
			}
		} else if flag == 4 {
			fmt.Println("请输入表名，进行数据判重.")
			var coll, stime, etime, sType string
			fmt.Scan(&coll)
			if coll != "" {
				fmt.Println("请输入开始时间， 格式：2006-01-02")
				fmt.Scan(&stime)
				if stime != "" {
					fmt.Println("请输入结束时间， 格式：2006-01-02")
					fmt.Scan(&etime)
					if etime != "" {
						fmt.Println("请输入排序方式，1正序、-1倒序")
						fmt.Scan(&sType)
						if sType != "" {
							task4(coll, stime, etime, sType)
						}
					} else {
						fmt.Println("请输入结束时间， 格式：2006-01-02")
					}
				} else {
					fmt.Println("请输入开始时间， 格式：2006-01-02")
				}
			}
		}
	}
}

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

func task1(path string) {
	if path != "" {

		Mgo.Update("f_sourceinfo_chinaunicom_zb_data", nil, bson.M{"$unset": bson.M{"tag": ""}}, false, true)

		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()
				id = util.SE.DecodeString(id)
				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
				}
				// 临时
				update["tag"] = "临时"
				lastid = id
				qu.Debug(update)
				Mgo.UpdateById("f_sourceinfo_chinaunicom_zb_data", id, bson.M{"$set": update})
			}
		}
		qu.Debug(fmt.Sprintf("更新数据成功，更新： %d条", count))
		os.Exit(0)
	}
}

func task2() {
	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)

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

	q := bson.M{"tag": "临时"}
	query := sess.DB(Mgo.DbName).C("f_sourceinfo_chinaunicom_zb_data").Find(q).Select(nil).Iter()
	count := 0
	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
			delete(info, "field_source")
			delete(info, "regions_log")
			Mgo.Save("tmp_usermail", info)

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

func task3() {
	var path string
	flag.StringVar(&path, "f", "", "文件路径")
	flag.Parse()
	if path != "" {
		file, err := xlsx.OpenFile(path)
		if err != nil {
			panic(err)
		}
		sheet := file.Sheets[0]
		cellFieldName := map[int]string{}
		for rn, row := range sheet.Rows {
			if rn == 0 {
				for index, cell := range row.Cells {
					//if cell.Value == "招标人" || cell.Value == "中标人" || cell.Value == "中标金额" {
					//	qu.Debug(cell.Value, index)
					//}
					if v := FieldsMap[cell.Value]; v != "" {
						cellFieldName[index] = v
					}
				}
			} else {
				q := bson.M{}
				for i, f := range cellFieldName {
					if val := row.Cells[i].Value; val != "" && (f == "s_winner" || f == "buyer") {
						q[f] = val
					}
					if val := row.Cells[i].Value; val != "" && f == "bidamount" {
						if qu.Float64All(val) != 0 {
							q[f] = qu.Float64All(val)
						}
					}
				}
				info, _ := Mgo.FindOne("zglt_history", q)
				if len(*info) > 0 {
					row.Cells[12].SetValue(-1)
				} else {
					row.Cells[12].SetValue(1)
				}
			}
		}
		err = file.Save(path)
	} else {
		flag.PrintDefaults()
	}
}