jy-data-extract/qyxy/save_mgo/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis"
	"go.mongodb.org/mongo-driver/bson"
	"mongodb"
	"qfw/util"
	"reflect"
	"strconv"
	"time"
)

var (
	MongoTool		*mongodb.MongodbSim
	rdb 			*redis.Client
)

func init() {
	MongoTool = &mongodb.MongodbSim{
		MongodbAddr: "172.17.145.163:27083,172.17.4.187:27082",
		Size:        10,
		DbName:      "mixdata",
		UserName: 	 "SJZY_RWESBid_Other",
		Password:	 "SJZY@O17t8herB3B",
	}
	MongoTool.InitPool()
}

func initRedis() (err error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second * 10)
	defer cancel()
	rdb = redis.NewClient(&redis.Options{
		Addr: "127.0.0.1:8379",
		PoolSize: 200,
		DB: 3,
	})
	_, err = rdb.Ping(ctx).Result()
	if err !=nil{
		fmt.Println("ping redis failed err:",err)
		return err
	}
	return nil
}

func main() {
	err := initRedis()
	if err !=nil{
		fmt.Println("init redis failed err :",err)
		return
	}
	ctx := context.Background()
	var cursor uint64
	var keys []string
	n := 0
	for {
		keys, cursor, err = rdb.Scan(ctx, cursor,"*",500).Result()
		if err !=nil{
			fmt.Println("scan keys failed err:",err)
			return
		}
		util.Debug("---keys---", len(keys))
		n += len(keys)
		for _,key := range keys{
			val, err := rdb.Get(ctx, key).Result()
			if err != nil{
				fmt.Println("get key values failed err:", err)
				return
			}
			val, _ = strconv.Unquote(val)						//  处理json字符串带转义符号
			maps := make(map[string]interface{})
			err1 := json.Unmarshal([]byte(val), &maps)
			if err1 != nil {
				util.Debug("-----map解析异常")
			}
			taskinfo(ctx, key, maps)
		}
		util.Debug("current---", n, cursor)
		if cursor == 0 {
			util.Debug("over---", n)
			break
		}
	}
}

func taskinfo(ctx context.Context, name string, tmp map[string]interface{})  {
	q := bson.M{"company_name": name}
	info, b := MongoTool.FindOneByField("qyxy_std", q, nil)
	if b && len(*info) > 0 {
		if types, ok := (*info)["bid_unittype"].([]interface{}); ok {
			t1 := util.ObjArrToStringArr(types)
			t2 := util.ObjArrToStringArr(tmp["bid_unittype"].([]interface{}))
			t2 = append(t2, t1...)
			tmp["bid_unittype"] = Duplicate(t2)
		}
		MongoTool.Update("qyxy_std", bson.M{"_id": (*info)["_id"]}, bson.M{"$set": tmp}, true, false)
	}else {
		tmp["company_name"] = name
		MongoTool.Save("qyxy_std_err", tmp)
		rdb.Del(ctx, name)
	}
}

func Duplicate(a interface{}) (ret []interface{}) {
	va := reflect.ValueOf(a)
	for i := 0; i < va.Len(); i++ {
		if i > 0 && reflect.DeepEqual(va.Index(i-1).Interface(), va.Index(i).Interface()) {
			continue
		}
		ret = append(ret, va.Index(i).Interface())
	}
	return ret
}