jy-data-extract/qyxy/qyxy_change/task.go

package main

import (
	"fmt"
	"github.com/cron"
	"go.mongodb.org/mongo-driver/bson"
	"log"
	"qfw/util"
	"regexp"
	"strings"
	"sync"
	"time"
)

var MgoSaveCache = make(chan map[string]interface{}, 5000)
var SP = make(chan bool, 5)
var updatePool = make(chan []map[string]interface{}, 5000)
var updateSp = make(chan bool, 5)

func TimeTask() {
	c := cron.New()
	//cronstr := "0 0 " + fmt.Sprint(TaskTime) + " * * ?"			//每天TaskTime跑一次
	cronstrPa := "0 0 15 ? * WED" //凭安增量数据每周二跑一次
	_ = c.AddFunc(cronstrPa, func() { GetData() })
	c.Start()
}

func GetData() {
	defer util.Catch()
	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)

	pool := make(chan bool, 10)
	wg := &sync.WaitGroup{}

	q := bson.M{"_id": bson.M{"$gt": lastId}}
	//q := bson.M{"_id": 368454367}
	var zid int
	it := sess.DB("mixdata").C("company_change").Find(q).Select(nil).Iter()
	count := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		if count%20000 == 0 {
			log.Println("current:", count)
		}
		zid = util.IntAll(tmp["_id"])
		pool <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-pool
				wg.Done()
			}()
			currentTime := time.Now().Unix()
			query := bson.M{"company_id": tmp["company_id"]}
			info, b := MgoMix.FindOneByField(CollSave, query, bson.M{"changes": 1})
			if b && len(*info) > 0 {
				if util.ObjToString(tmp["_operation_type"]) == "insert" {
					update := make(map[string]interface{})
					item := make(map[string]interface{})
					item["change_field"] = tmp["change_field"]
					item["content_before"] = tmp["content_before"]
					item["content_after"] = tmp["content_after"]
					item["change_date"] = tmp["change_date"]
					setMark(item) //change_name_new
					//update["changes"] = changes
					update["update_time"] = currentTime
					saveInfo := []map[string]interface{}{
						{"company_id": tmp["company_id"]},
						{"$set": update, "$push": map[string]interface{}{"changes": item}},
					}
					updatePool <- saveInfo
				}
			} else {
				query := bson.M{"_id": tmp["company_id"]}
				qyxy, b1 := MgoMix.FindOne("qyxy_std", query)
				if b1 && len(*qyxy) > 0 {
					save := make(map[string]interface{})
					var changes []map[string]interface{}
					item := make(map[string]interface{})
					item["change_field"] = tmp["change_field"]
					item["content_before"] = tmp["content_before"]
					item["content_after"] = tmp["content_after"]
					item["change_date"] = tmp["change_date"]
					setMark(item) //change_name_new
					changes = append(changes, item)

					save["company_name"] = (*qyxy)["company_name"]
					save["company_id"] = (*qyxy)["_id"]
					save["changes"] = changes
					save["create_time"] = currentTime
					save["update_time"] = currentTime
					//save["_id"] = primitive.NewObjectID()
					saveInfo := []map[string]interface{}{
						{"company_id": tmp["company_id"]},
						{"$set": save},
					}
					updatePool <- saveInfo
				}
			}
		}(tmp)
	}
	util.Debug("over---", count, zid)
}

func setMark(tmp map[string]interface{}) {
	for _, v := range ChangeMap {
		str := util.ObjToString(tmp["change_field"])
		regArr := v["change_key_reg"].([]string)
		for _, v1 := range regArr {
			matched, _ := regexp.MatchString(v1, str)
			if matched {
				tmp["change_name_new"] = v["change_name"]
				return
			}
		}
	}
}

func clearRepeat(list []interface{}) []interface{} {
	var tmp []interface{}
	if len(list) > 1 {
		for k, v := range list {
			if k < len(list)-1 {
				if fmt.Sprint(list[k]) != fmt.Sprint(list[k+1]) {
					tmp = append(tmp, v)
				}
			} else {
				tmp = append(tmp, v)
			}
		}
		return tmp
	} else {
		return list
	}
}

// TaskAll 存量数据
func TaskAll() {
	defer util.Catch()
	sess := MgoMix.GetMgoConn()
	defer MgoMix.DestoryMongoConn(sess)

	pool := make(chan bool, 10)
	wg := &sync.WaitGroup{}
	field := bson.M{"company_name": 1}
	it := sess.DB("mixdata").C(CollQy).Find(nil).Select(field).Iter()
	count := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		if count%20000 == 0 {
			log.Println("current:", count, tmp["_id"])
		}
		if strings.Contains(util.ObjToString(tmp["company_type"]), "个体") {
			continue
		}
		pool <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-pool
				wg.Done()
			}()
			tmp["company_id"] = tmp["_id"]
			delete(tmp, "_id")
			query := bson.M{"company_id": tmp["company_id"]}
			info, b := Mgo.Find("company_change", query, nil, nil, false, -1, -1)
			if b && len(*info) > 0 {
				disposeFuc(*info, tmp)
			}
		}(tmp)
	}
}

func taskInfo(id string) {
	update := make(map[string]interface{})
	q := map[string]interface{}{"company_id": id}
	info, _ := Mgo.Find("company_change", q, nil, nil, false, -1, -1)
	if len(*info) > 0 {
		var changes []map[string]interface{}
		currentTime := time.Now().Unix()
		for _, v := range *info {
			item := make(map[string]interface{})
			item["change_field"] = v["change_field"]
			item["content_before"] = v["content_before"]
			item["content_after"] = v["content_after"]
			item["change_date"] = v["change_date"]
			setMark(item) //change_name_new
			changes = append(changes, item)
		}
		update["changes"] = changes
		update["update_time"] = currentTime
	}
	util.Debug(update)
	MgoMix.Update("qyxy_change", q, map[string]interface{}{"$set": update}, false, false)
}

func disposeFuc(maps []map[string]interface{}, tmp map[string]interface{}) {
	var changes []map[string]interface{}
	currentTime := time.Now().Unix()
	for _, v := range maps {
		item := make(map[string]interface{})
		item["change_field"] = v["change_field"]
		item["content_before"] = v["content_before"]
		item["content_after"] = v["content_after"]
		item["change_date"] = v["change_date"]
		setMark(item) //change_name_new
		changes = append(changes, item)
	}
	tmp["changes"] = changes
	tmp["create_time"] = currentTime
	tmp["update_time"] = currentTime

	MgoSaveCache <- tmp
}

func SaveData() {
	log.Println("Mgo Save...")
	arru := make([]map[string]interface{}, 200)
	indexu := 0
	for {
		select {
		case v := <-MgoSaveCache:
			arru[indexu] = v
			indexu++
			if indexu == 200 {
				SP <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-SP
					}()
					MgoMix.SaveBulk(CollSave, arru...)
				}(arru)
				arru = make([]map[string]interface{}, 200)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				SP <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-SP
					}()
					MgoMix.SaveBulk(CollSave, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, 200)
				indexu = 0
			}
		}
	}
}

func updateMethod() {
	arru := make([][]map[string]interface{}, 200)
	indexu := 0
	for {
		select {
		case v := <-updatePool:
			arru[indexu] = v
			indexu++
			if indexu == 200 {
				updateSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateSp
					}()

					MgoMix.UpSertBulk(CollSave, arru...)
				}(arru)
				arru = make([][]map[string]interface{}, 200)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				updateSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateSp
					}()
					MgoMix.UpSertBulk(CollSave, arru...)
				}(arru[:indexu])
				arru = make([][]map[string]interface{}, 200)
				indexu = 0
			}
		}
	}
}