all_platform/job/job.go

package job

import (
	"all_platform/conn"
	"all_platform/dbpool"
	"bytes"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/robfig/cron/v3"
)

func Job() {
	job := cron.New(cron.WithSeconds())

	job.Schedule(cron.Every(5*time.Minute), &myJob{
		t: time.Now(),
	})
	job.Schedule(cron.Every(30*time.Minute), &myJob1{
		t: time.Now(),
	})
	job.Start()
	select {}
}

// 查询ip地区，并写入数据库
type myJob struct {
	i int
	t time.Time
}

type myJob1 struct {
	i int
	t time.Time
}

// 更新ip地址用户地区
func (j *myJob) Run() {
	j.i++
	now := time.Now()
	sub := now.Sub(j.t)
	j.t = now
	fmt.Printf("ip地址更新: %d, duraction : %d ms \n", j.i, sub.Milliseconds())
	//查询sql 获取用户信息
	rows, err := dbpool.Query("SELECT  DT_SRCIPADDR  FROM  `dt_zgdz`.`dt_attack_event` WHERE DT_CUNTRY IS NULL LIMIT 1")
	if err != nil {
		log.Println(err)
	}
	defer rows.Close()

	// var typecho_userss []dbpool.Typecho_users
	var ipStr sql.NullString
	for rows.Next() {
		if err := rows.Scan(&ipStr); err != nil {
			log.Println(err)
		}
		// typecho_userss = append(typecho_userss, users)
	}
	if err := rows.Err(); err != nil {
		log.Fatal(err)
	}
	if ipStr.Valid {
		//请求接口获取ip地区
		header := map[string]string{
			"Content-Type": "application/json",
		}
		url := "http://ip-api.com/json/" + ipStr.String + "?lang=zh-CN"
		iplistStr := conn.DT_GET(url, header, nil)
		var data map[string]interface{}
		err = json.Unmarshal([]byte(iplistStr), &data)
		if err != nil {
			log.Fatal(err)
		}
		//开始更新数据
		result, err := dbpool.ExecuteSQL("UPDATE `dt_zgdz`.`dt_attack_event` SET `DT_CUNTRY`=? where  DT_SRCIPADDR = ?", data["country"].(string)+"/"+data["city"].(string), ipStr.String)
		if err != nil {
			log.Println(err)
		}
		// 获取插入操作的最后插入ID和受影响的行数
		lastInsertId, err := result.LastInsertId()
		if err != nil {
			log.Fatal(err)
		}
		rowsAffected, err := result.RowsAffected()
		if err != nil {
			log.Fatal(err)
		}
		log.Printf("Last Insert ID: %d, Rows Affected: %d\n", lastInsertId, rowsAffected)
	} else {
		log.Println("没有数据")
	}
}

// 每天更新一次星座数据
func (j1 *myJob1) Run() {
	j1.i++
	now := time.Now()
	sub := now.Sub(j1.t)
	j1.t = now
	fmt.Printf("星座信息更新: %d, duraction : %d ms \n", j1.i, sub.Milliseconds())
	//更具星座列表获取星座数据
	//查询sql 获取用户信息
	rows, err := dbpool.Query("SELECT DT_ITEMID FROM `all_platform`.`dt_constellation` ")
	if err != nil {
		log.Println(err)
	}
	defer rows.Close()

	var typecho_userss []string
	var itemIDStr sql.NullString
	for rows.Next() {
		if err := rows.Scan(&itemIDStr); err != nil {
			log.Println(err)
		}
		typecho_userss = append(typecho_userss, itemIDStr.String)
	}
	if err := rows.Err(); err != nil {
		log.Fatal(err)
	}
	for _, v := range typecho_userss {
		//查询sql 获取用户信息
		rows, err := dbpool.Query("SELECT DT_DATE FROM `all_platform`.`dt_constellation_info` WHERE DT_DATE=? LIMIT 1 ;", time.Now().Format("2006-01-02"))
		if err != nil {
			log.Println(err)
		}
		defer rows.Close()

		// var typecho_userss []string
		var itemIDStr sql.NullString
		for rows.Next() {
			if err := rows.Scan(&itemIDStr); err != nil {
				log.Println(err)
			}
			// typecho_userss = append(typecho_userss, itemIDStr.String)
		}
		if err := rows.Err(); err != nil {
			log.Fatal(err)
		}
		//不等于今天，插入数据，等于今天，不插入数据
		if itemIDStr.String != time.Now().Format("2006-01-02") {
			//请求接口获取ip地区
			header := map[string]string{
				"Content-Type": "application/json",
			}
			jsonStr := `{
		"cardDataReq": [
			{
				"serviceID": "e61264eb17f94d518dc143344304ef29",
				"beginTime": "` + time.Now().Format("2006-01-02") + `T00:00+0800",
				"endTime": "` + time.Now().Format("2006-01-02") + `T23:59+0800",
				"parameters": {
					"itemID": "` + v + `",
					"type": [
						"today",
						"tomorrow",
						"week",
						"month"
					]
				}
			}
		],
		"reqAllInfo": 1
    }`
			listStr := conn.DT_POST("https://api-calendar-drcn.music.dbankcloud.cn/calendar-sub-service/v1/carddata/simple/querybyids", header, bytes.NewBuffer([]byte(jsonStr)))
			var data map[string]interface{}
			err = json.Unmarshal([]byte(listStr), &data)
			if err != nil {
				log.Fatal(err)
			}
			var dataStr map[string]interface{}
			if err := json.Unmarshal([]byte(data["datas"].([]interface{})[0].(map[string]interface{})["originalData"].([]interface{})[0].(map[string]interface{})["data"].(string)), &dataStr); err != nil {
				log.Fatal(err)
			}
			// 将 map 转换为 JSON 格式的字符串
			week, err := json.Marshal(dataStr["week"])
			month, err := json.Marshal(dataStr["month"])
			today, err := json.Marshal(dataStr["today"])
			tomorrow, err := json.Marshal(dataStr["tomorrow"])
			if err != nil {
				fmt.Println("Error marshaling JSON:", err)
				return
			}
			if dataStr["date"].(string) != "" {
				//开始更新数据
				result, err := dbpool.ExecuteSQL("INSERT INTO `all_platform`.`dt_constellation_info` (`DT_DATE`, `DT_SIGNNAME`, `DT_WEEK`, `DT_MONTH`, `DT_TODAY`, `DT_TOMORROW`) VALUES (?,?,?,?,?,?);", dataStr["date"], dataStr["signName"], week, month, today, tomorrow)
				if err != nil {
					log.Println(err)
				}
				// 获取插入操作的最后插入ID和受影响的行数
				lastInsertId, err := result.LastInsertId()
				if err != nil {
					log.Fatal(err)
				}
				rowsAffected, err := result.RowsAffected()
				if err != nil {
					log.Fatal(err)
				}
				log.Printf("Last Insert ID: %d, Rows Affected: %d\n", lastInsertId, rowsAffected)
			} else {
				log.Println("没有数据")
			}
		} else {
			log.Println("已有今天的数据，不在插入数据")
		}
	}
}