data_field_dispose/monitor/main.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"github.com/robfig/cron"
	"github.com/spf13/cobra"
	"go.mongodb.org/mongo-driver/bson"
	"go.uber.org/zap"
	"io/ioutil"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
	"monitor/config"
	"net/http"
	"time"
)

var (
	WebUrl  = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=45962efc-ca87-4996-9ffa-08bf6608ab7a"
	WebUrl1 = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=1594c58d-c279-4fdc-8d44-c201160e3731"

	WarningStr  = "数据采集bidding_file表数据，已累计%d条数据未处理"
	WarningStr1 = "标的物等字段识别提醒，bidding表数据已积累%d条未处理"
	WarningStr2 = "项目合并表数据，超过一个1小时(查询时间点: %d)未有合并新数据"
)

func init() {
	config.Init("./common.toml")
	InitLog()
}

func main() {
	rootCmd := &cobra.Command{Use: "my cmd"}
	rootCmd.AddCommand(biddingFile())
	if err := rootCmd.Execute(); err != nil {
		fmt.Println("rootCmd.Execute failed", err.Error())
	}
}

func biddingFile() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "bidding_warn",
		Short: "Start statistics bidding_file data",
		Run: func(cmd *cobra.Command, args []string) {
			InitMgo()

			crn := cron.New()
			_ = crn.AddFunc("@hourly", func() {
				taskFile()
			})

			_ = crn.AddFunc("0 */30 * * * ?", func() {
				taskPy()
			})

			_ = crn.AddFunc("@hourly", func() {
				taskProject()
			})
			crn.Start()

			c := make(chan bool, 1)
			<-c
		},
	}
	return cmdClient
}

func taskFile() {
	count := MgoB.Count("bidding_file", bson.M{"moveok": bson.M{"$exists": false}})
	if count > config.Conf.Serve.FileWarn {
		SendMsg(fmt.Sprintf(WarningStr, count))
	} else {
		log.Info("bidding_file", zap.Int("count", count))
	}
}

func taskPy() {
	info, _ := MgoB.Find("bidding_processing_ids", bson.M{"dataprocess": bson.M{"$in": []int{1, 2}}}, nil, bson.M{"count": 1, "dataprocess": 1}, false, -1, -1)
	count := 0
	for _, m := range *info {
		count += util.IntAll(m["count"])
	}
	if count > config.Conf.Serve.PyWarn {
		SendMsg(fmt.Sprintf(WarningStr1, count))
	} else {
		log.Info("bidding_py", zap.Int("count", count))
	}
}

func taskProject() {
	now := time.Now().Unix() - 1*60*60
	c := MgoP.Count(config.Conf.DB.MongoP.Coll, bson.M{"pici": bson.M{"$gte": now}})
	if c <= 0 {
		SendMsg1(fmt.Sprintf(WarningStr2, now))
	} else {
		log.Info("projectset", zap.Int("count", c))
	}
}

func SendMsg(content string) {
	client := &http.Client{}
	data := map[string]interface{}{"msgtype": "text", "text": map[string]interface{}{
		"content": content, "mentioned_mobile_list": []string{"13373929153", "15090279371", "15639297172"},
	}}
	bytesData, _ := json.Marshal(data)
	req, _ := http.NewRequest("POST", WebUrl, bytes.NewReader(bytesData))
	resp, _ := client.Do(req)
	body, _ := ioutil.ReadAll(resp.Body)
	log.Info("SendMsg", zap.String("resp", string(body)))
}

func SendMsg1(content string) {
	client := &http.Client{}
	data := map[string]interface{}{"msgtype": "text", "text": map[string]interface{}{
		"content": content, "mentioned_mobile_list": []string{"15090279371", "15639297172"},
	}}
	bytesData, _ := json.Marshal(data)
	req, _ := http.NewRequest("POST", WebUrl1, bytes.NewReader(bytesData))
	resp, _ := client.Do(req)
	body, _ := ioutil.ReadAll(resp.Body)
	log.Info("SendMsg", zap.String("resp", string(body)))
}