timed_tasks/escheck/main.go

package main

import (
	util "app.yhyue.com/data_processing/common_utils"
	"app.yhyue.com/data_processing/common_utils/elastic"
	"app.yhyue.com/data_processing/common_utils/mongodb"
	"encoding/json"
	"fmt"
	es "github.com/olivere/elastic/v7"
	"github.com/robfig/cron/v3"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"
)

//定时任务，去统计bidding索引，排查问题、预警

type T struct {
	Cron    string
	Name    string
	Min     int
	Max     int
	Type    string
	Tjscope string
	Mgo     string
}

//es数据最报警
var (
	config                      map[string]interface{}
	to                          string
	api                         string
	esAddr, esAddr2, esAddr3    string
	esIndex, esIndex2, esIndex3 string
	username, password          string
	username3, password3        string
	Ts                          = []*T{}
	esQ                         = `{"query": {"range": {"id": {"gte": "%s","lt": "%s"}}}}`
	esQ1                        = `{"query": {"bool": {"must": [{"range": {"id": {"gte": "%s","lt": "%s"}}},{"terms": {"bidding.site": ["元博网（采购与招标网）","中国招标与采购网"]}}]}}}`
)

func init() {
	util.ReadConfig(&config)
	jkmail := config["jkmail"].(map[string]interface{})
	to, _ = jkmail["to"].(string)
	api, _ = jkmail["api"].(string)
	//esAddr, _ = config["esAddr"].(string)
	esIndex, _ = config["esIndex"].(string)
	esAddr2, _ = config["esAddr2"].(string)
	esIndex2, _ = config["esIndex2"].(string) //
	if _, ok := config["esAddr3"]; ok {
		esAddr3, _ = config["esAddr3"].(string)
		esIndex3, _ = config["esIndex3"].(string) //
		username3, _ = config["username3"].(string)
		password3, _ = config["password3"].(string)
	}
	username, _ = config["username"].(string)
	password, _ = config["password"].(string)
	tasks, _ := config["task"].([]interface{})
	for _, t := range tasks {
		bs, _ := json.Marshal(t)
		var v *T
		json.Unmarshal(bs, &v)
		if v != nil {
			Ts = append(Ts, v)
		}
	}
}

func main() {
	log.Println("start..")
	if len(Ts) > 0 {
		local, _ := time.LoadLocation("Asia/Shanghai")
		c := cron.New(cron.WithLocation(local), cron.WithSeconds())
		for _, v := range Ts {
			c.AddFunc(v.Cron, v.task)
		}
		c.Start()
		defer c.Stop()
		select {}
	}
	log.Println("end..")

}

func (t *T) task() {
	//初始化语句
	qt := strings.Split(t.Tjscope, ",")
	if len(qt) != 2 {
		return
	}
	st, et := int64(0), int64(0)
	now := time.Now()
	switch qt[1] {
	case "h":
		et = now.Unix()
		st = et + util.Int64All(qt[0])*3600
	case "d":
		st = util.GetDayStartSecond(util.IntAll(qt[0]))
		et = util.GetDayStartSecond(0)
	}
	st1 := fmt.Sprintf("%x0000000000000000", st)
	et1 := fmt.Sprintf("%x0000000000000000", et)

	eq := fmt.Sprintf(esQ, st1, et1)
	fmt.Println("eq", eq)
	eq1 := fmt.Sprintf(esQ1, st1, et1)
	fmt.Println("eq1", eq1)

	rangeQuery := es.NewRangeQuery("id").Gte(st1).Lt(et1)
	termsQuery := es.NewTermsQuery("site", "元博网（采购与招标网）", "中国招标与采购网")
	countQuery := es.NewBoolQuery().
		Must(rangeQuery)

	count1Query := es.NewBoolQuery().Must(rangeQuery).Filter(termsQuery)

	es2 := elastic.Elastic{S_esurl: esAddr2, I_size: 2, Username: username, Password: password}
	es2.InitElasticSize()

	es3 := elastic.Elastic{S_esurl: esAddr3, I_size: 5, Username: username3, Password: password3}
	es3.InitElasticSize()

	count := int(es2.Count(esIndex2, esIndex2, countQuery))    //公司 es集群 数量统计
	count1 := int(es2.Count(esIndex, esIndex, count1Query))    //竞品网站 数量统计
	countNew := int(es3.Count(esIndex3, esIndex3, countQuery)) //华为云 新集群
	fmt.Println("count", count)
	fmt.Println("count1", count1)
	fmt.Println("countNew", countNew)

	switch t.Type {
	case "alert":
		if count < t.Min || count > t.Max || countNew < t.Min || countNew > t.Max {
			report := fmt.Sprintf("告警%s,最小%d,最大%d,统计结果：%d", t.Name, t.Min, t.Max, count)
			t.SendMail(report)
		}
	case "report":
		report := fmt.Sprintf("统计报告%s，【统计结果】,es库数量：%d", t.Name, count)
		if len(t.Mgo) > 5 {
			fs := strings.Split(t.Mgo, "|")
			fmgo := mongodb.NewMgoWithUser(fs[0], fs[3], fs[1], fs[2], 1)
			id1 := mongodb.StringTOBsonId(st1)
			id2 := mongodb.StringTOBsonId(et1)

			mq := bson.M{"_id": bson.M{"$gte": id1, "$lt": id2}} //一天时间内的id段
			fd := bson.M{"extracttype": 1, "sensitive": 1, "dataging": 1, "site": 1, "infoformat": 1, "comeintime": 1, "pici": 1, "publishtime": 1, "competehref": 1, "attach_text": 1}
			sess := fmgo.GetMgoConn()
			defer fmgo.DestoryMongoConn(sess)
			/**
			count   一天内，es 中 数据总量
			count1  一天内，es 中 竞品总量
			count2  一天内，mgo 总入库量
			count3 一天内，mgo 有效数据 总数
			count4  一天内，mgo 中 竞品数据总量
			count5  一天内，mgo 有效数据中，竞品的数量
			countNew 一天内，es3新集群 中 数据总量
			timeCount pici-comeintime 时差在12小时内的数据
			*/
			count2, count3 := 0, 0               //
			count4, count5 := 0, 0               //竟品
			es_comeintime_totaltime := int64(0)  //comeintime 和 生索引 pici 时间 差值的总和
			es_publishtime_totaltime := int64(0) //publishtime 和 生索引 pici 时间 差值的总和
			es_comeintime_avgtime := int64(0)    //comeintime 和 生索引 pici 时间 差值的平均值
			es_publishtime_avgtime := int64(0)   //publishtime 和 生索引 pici 时间 差值的平均值

			file_totaltime := int64(0)
			no_file_totaltime := int64(0)
			file_avgltime := int64(0)
			no_file_avgltime := int64(0)
			timeCount := 0 //12小时  统计时间差的数据量
			fileCount := 0
			noFileCount := 0

			//统计pici -comeintime 时间差,1、3、5、10、15、30、30+ 分钟
			var pc_diff1 int64
			var pc_diff3 int64
			var pc_diff5 int64
			var pc_diff10 int64
			var pc_diff15 int64
			var pc_diff30 int64
			var pc_diff301 int64

			query := sess.DB(fs[3]).C(fs[4]).Find(mq).Select(fd).Iter()
			for tmp := make(map[string]interface{}); query.Next(tmp); count2++ {
				if util.ObjToString(tmp["site"]) == "元博网（采购与招标网）" || util.ObjToString(tmp["site"]) == "中国招标与采购网" {
					count4++
				}
				if util.IntAll(tmp["extracttype"]) != -1 && util.ObjToString(tmp["sensitive"]) != "测试" && util.IntAll(tmp["dataging"]) != 1 && util.Float64All(tmp["infoformat"]) != 3 {
					count3++
					comeintime := util.Int64All(tmp["comeintime"])
					publishtime := util.Int64All(tmp["publishtime"])
					pici := util.Int64All(tmp["pici"])

					tp := time.Unix(publishtime, 0)
					isZero := false
					if (tp.Hour() == 0 && tp.Minute() == 0) && tp.Second() == 0 {
						isZero = true
					}

					if (comeintime-publishtime) < 12*60*60 && !isZero {
						if pici > 0 {
							diff1 := pici - comeintime
							diff2 := pici - publishtime
							if diff1 < 0 {
								fmt.Println("diff1", diff1, tmp["_id"])
							} else if diff1 <= 60 {
								pc_diff1++
							} else if diff1 <= 3*60 {
								pc_diff3++
							} else if diff1 <= 5*60 {
								pc_diff5++
							} else if diff1 <= 10*60 {
								pc_diff10++
							} else if diff1 <= 15*60 {
								pc_diff15++
							} else if diff1 <= 30*60 {
								pc_diff30++
							} else {
								pc_diff301++
							}

							if diff2 < 0 {
								fmt.Println("diff2", diff2, tmp["_id"])
							}
							es_comeintime_totaltime += diff1
							es_publishtime_totaltime += diff2
							timeCount++

							if _, ok := tmp["attach_text"]; ok {
								curtime := tmp["_id"].(primitive.ObjectID).Timestamp().Unix()
								diff3 := curtime - comeintime
								if diff3 >= 0 {
									file_totaltime += diff3
								}
								fileCount++
							} else {
								curtime := tmp["_id"].(primitive.ObjectID).Timestamp().Unix()
								diff4 := curtime - comeintime
								if diff4 >= 0 {
									no_file_totaltime += diff4
								}
								noFileCount++
							}
						}
					}

					if util.ObjToString(tmp["site"]) == "元博网（采购与招标网）" || util.ObjToString(tmp["site"]) == "中国招标与采购网" {
						count5++
					}
				}
			}

			if timeCount > 0 {
				es_comeintime_avgtime = es_comeintime_totaltime / int64(timeCount)
				es_publishtime_avgtime = es_publishtime_totaltime / int64(timeCount)
			}
			if fileCount > 0 {
				file_avgltime = file_totaltime / int64(fileCount)
			}
			if noFileCount > 0 {
				no_file_avgltime = no_file_totaltime / int64(noFileCount)
			}

			report += ",mgo统计:" + fmt.Sprint(count3) + ",差值:" + fmt.Sprint(count3-count) + ",mgo总入库量:" + fmt.Sprint(count2)
			report += "<br>" + "【竟品统计结果】:" + strconv.Itoa(count1) + ",mgo统计:" + fmt.Sprint(count5) + ",差值:" + fmt.Sprint(count5-count1) + ",mgo总入库量" + fmt.Sprint(count4) + "<br>【新集群统计结果】es数量：" + strconv.Itoa(countNew) + ",mgo统计:" + fmt.Sprint(count3) + ",差值:" + fmt.Sprint(count3-countNew)
			//存入数据库
			yesterday := now.AddDate(0, 0, -1)
			insert := map[string]interface{}{
				"es_count":                 count,    //
				"es3_count":                countNew, //
				"mgo_count":                count3,
				"es_mgo_diff":              count3 - count,
				"mgo_total":                count2,
				"competitor_es_count":      count1, //竞品网站es 数量
				"competitor_mgo_count":     count5,
				"competitor_diff":          count5 - count1,
				"competitor_mgo_total":     count4,
				"date":                     yesterday.Format("2006-01-02"),
				"es_comeintime_totaltime":  es_comeintime_totaltime,
				"es_publishtime_totaltime": es_publishtime_totaltime,
				"es_comeintime_avgtime":    es_comeintime_avgtime,
				"es_publishtime_avgtime":   es_publishtime_avgtime,
				"file_avgltime":            file_avgltime,
				"no_file_avgltime":         no_file_avgltime,
				"file_totaltime":           file_totaltime,
				"no_file_totaltime":        no_file_totaltime,
				"file_count":               fileCount,
				"no_file_count":            noFileCount,
				"pc_diff1":                 pc_diff1,
				"pc_diff3":                 pc_diff3,
				"pc_diff5":                 pc_diff5,
				"pc_diff10":                pc_diff10,
				"pc_diff15":                pc_diff15,
				"pc_diff30":                pc_diff30,
				"pc_diff301":               pc_diff301,
				"timeCount":                timeCount,
			}
			fmgo.Save("bidding_ribao", insert)

		}
		t.SendMail(report)
		//t.SendMail("【竟品统计结果】")
		//fmt.Println(report)

	}
	log.Println("task over:", t.Name, eq, count)
}

func (t *T) SendMail(report string) {
	url := fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, to, t.Name, report)
	fmt.Println("url=>", url)
	res, err := http.Get(url)
	if err != nil {
		fmt.Println("SendMail err ", err)
	} else {
		fmt.Println("SendMail res ", res)
	}
}