master

分支 (1)

管理

管理

master

watcher
/
main.go

package main

import (
	"github.com/labstack/echo"
	"net/http"
	"github.com/labstack/echo/middleware"
	"github.com/asaskevich/govalidator"
	"os"
	"time"
	"github.com/labstack/gommon/log"
	"github.com/influxdata/influxdb/client"
	"net/url"
	"fmt"
	"strconv"
	"flag"
	"github.com/PedroGao/CmsGo/pkg/ip"
	"io/ioutil"
	"github.com/json-iterator/go"
)

const (
	welcome = `
		<!DOCTYPE html>
		<html lang="zh-cn">

		<head>
		  <meta charset="UTF-8">
		  <meta name="viewport" content="width=device-width, initial-scale=1.0">
		  <meta http-equiv="X-UA-Compatible" content="ie=edge">
		  <title>Watcher</title>
		  <style>
		    .body {
		      text-align: center;
		      margin: 0 auto;
		      width: 1000px;
		      border: 1px solid #003300;
		    }
		  </style>
		</head>

		<body>
		  <div class="body">
		    <h1>Hello From Watcher</h1>
		    <small>Powered By Pedro🤣</small>
		  </div>
		</body>

		</html>
	`
	TAOBAO_IP_REQUEST_URL = "http://ip.taobao.com/service/getIpInfo.php?ip=%s"
)

// ip_flow struct
type (
	IpFlow struct {
		// 传入
		Devkey  string `json:"devkey" valid:"required,devkey"`
		Puid    int    `json:"puid" valid:"required,int"`
		Ip      string `json:"ip" valid:"required,ip"`
		ApiSign string `json:"api_sign" valid:"required"`
		Appkey  string `json:"appkey" valid:"required"`
		// 查询所得
		Country string `json:"country"`
		Region  string `json:"region"` // 省份
		City    string `json:"city"`
		Isp     string `json:"isp"`
	}

	NotFoundIp struct {
		NotComplete *IpFlow   `json:"not_complete"`
		Time        time.Time `json:"time"`
	}

	EValidator struct {
	}
)

func (cv *EValidator) Validate(i interface{}) error {
	if ok, err := govalidator.ValidateStruct(i); ok {
		return nil
	} else {
		return err
	}
}

func addCustomValidators() {
	// Add your own struct validation tags
	govalidator.TagMap["devkey"] = govalidator.Validator(func(str string) bool {
		return str == "8uy6da" || str == "ku7hax"
	})
}

var (
	region *ip.Ip2Region
	// todo 把当时的时间记录下来，然后丢入channel里面
	unusedIps chan *NotFoundIp
	//receivedIps chan string
	ipInfos chan *IpFlow
	e       *echo.Echo
	conn    *client.Client
	dbPath  string
)

func init() {
	flag.StringVar(&dbPath, "db_path", "pkg/data/ip2region.db", "the ip db path")
}

func initRegion(path string) {
	_, err := os.Stat(path)
	if os.IsNotExist(err) {
		panic("not found db " + path)
	}
	if region, err = ip.New(path); err != nil {
		panic("open ip db failed")
	}
	// 寻找地域失败的ip 100 的缓冲区
	unusedIps = make(chan *NotFoundIp, 100)

	// 接收到ip 1000的缓冲区
	//receivedIps = make(chan string, 1000)

	//查询得到的ip区域信息 1000的缓冲区
	ipInfos = make(chan *IpFlow, 1000)
}

func main() {
	flag.Parse()
	// db init
	initRegion(dbPath)
	// server init
	e = echo.New()
	// not debug
	e.Debug = false
	e.Use(middleware.Recover())
	e.Use(middleware.Logger())
	// add validator for echo
	addCustomValidators()
	e.Validator = &EValidator{}

	// init influx client
	host, err := url.Parse(fmt.Sprintf("http://%s:%d", "localhost", 8086))
	if err != nil {
		panic(err)
	}
	conf := client.Config{
		URL:      *host,
		Username: "root",
		Password: "root",
	}
	conn, err = client.NewClient(conf)
	if err != nil {
		panic(err)
	}

	// route
	e.GET("/", func(c echo.Context) error {
		return c.HTML(http.StatusOK, welcome)
	})

	// 写信息
	go func() {
		for {
			select {
			case info := <-ipInfos:
				e.Logger.Infof("get an ip info : %v", info)
				go writePoints(info, time.Now())
				// 若一个小时未收到任何信息输出一条日志
			case <-time.After(time.Hour * 1):
				e.Logger.Info("there is no ip infos received in an hour")
			}
		}
	}()

	// 若有信息查询不到
	go func() {
		for {
			select {
			case unusedInfo := <-unusedIps:
				e.Logger.Infof("get an unused ip info : %v", unusedInfo)
				go queryAndWrite(unusedInfo)
				// 若一天未收到任何信息输出一条日志
			case <-time.After(time.Hour * 24):
				e.Logger.Info("there is no unused ip infos received in a day")
			}
		}
	}()

	e.POST("/ip", ipHandler)

	e.Logger.SetLevel(log.INFO)

	e.Logger.Fatal(e.Start(":3000"))
}

func writePoints(info *IpFlow, time time.Time) {
	point := client.Point{
		Measurement: "ip_flow",
		Tags: map[string]string{
			"puid":     strconv.Itoa(info.Puid),
			"devkey":   info.Devkey,
			"api_sign": info.ApiSign,
			"ip":       info.Ip,
			"appkey":   info.Appkey,
		},
		Fields: map[string]interface{}{
			"country": info.Country,
			"region":  info.Region,
			"city":    info.City,
			"isp":     info.Isp,
		},
		Time: time,
	}
	bps := client.BatchPoints{
		Points:   []client.Point{point},
		Database: "pkg",
		//RetentionPolicy: "default",
	}
	if _, err := conn.Write(bps); err != nil {
		e.Logger.Infof("write an ip info to influxdb err: %v", err)
	}
}

func queryAndWrite(unusedInfo *NotFoundIp) {
	if ok := QueryForRegion(unusedInfo.NotComplete.Ip, unusedInfo.NotComplete); ok {
		writePoints(unusedInfo.NotComplete, unusedInfo.Time)
	}
}

func QueryForRegion(ip string, NotComplete *IpFlow) bool {
	url := fmt.Sprintf(TAOBAO_IP_REQUEST_URL, ip)
	resp, err := http.Get(url)
	if err != nil {
		e.Logger.Infof("get taobao ip api err: %v", err)
	}
	bytes, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		e.Logger.Infof("get taobao ip api err: %v", err)
	}
	code := jsoniter.Get(bytes, "code").ToInt()
	// 其中code的值的含义为，0：成功，1：失败。
	if code == 0 {
		data := jsoniter.Get(bytes, "data")
		country := data.Get("country").ToString()
		region := data.Get("region").ToString()
		city := data.Get("city").ToString()
		isp := data.Get("isp").ToString()
		NotComplete.Country = country
		NotComplete.Region = region
		NotComplete.City = city
		NotComplete.Isp = isp
		return true
	}
	return false
}

func ipHandler(c echo.Context) error {
	var (
		err error
	)
	ip := new(IpFlow)

	if err = c.Bind(ip); err != nil {
		c.Error(echo.NewHTTPError(400, err.Error()))
		return nil
	}
	if err = c.Validate(ip); err != nil {
		c.Error(echo.NewHTTPError(400, err.Error()))
		return nil
	}

	// 将ip查询放入协程中
	go func() {
		// 如果没有找到ip 将ip写入到channel中
		if info, err := region.BtreeSearch(ip.Ip); err != nil {
			// 传入的信息及传入时间写到channel中，供后续调用
			unusedIps <- &NotFoundIp{
				NotComplete: ip,
				Time:        time.Now(),
			}
		} else {
			// 写入 ip信息
			ip.Country = info.Country
			ip.Region = info.Region
			ip.City = info.City
			ip.Isp = info.ISP
			// 写入 channel中
			ipInfos <- ip
		}
	}()
	return c.JSON(http.StatusOK, map[string]string{"message": "ok"})
}