logdb/logdb.go

package logdb

import (
	"database/sql"
	"fmt"
	"io/ioutil"
	"log"
	"time"

	yaml "gopkg.in/yaml.v2"
)

// LogDB 属性结构
type LogDB struct {
	Charset  string   `yaml:"charset"`
	DB       string   `yaml:"db"`
	Hosts    []string `yaml:"hosts"`
	Password string   `yaml:"password"`
	Port     string   `yaml:"port"`
	User     string   `yaml:"user"`

	hostid int
	driver *sql.DB
}

// New 创建一个logdb的配置
func New(filename string) *LogDB {
	logdb := LogDB{}
	data, err := ioutil.ReadFile(filename)
	if err != nil {
		panic(err)
	}
	err = yaml.Unmarshal(data, &logdb)
	if err != nil {
		panic(err)
	}
	logdb.hostid = 0

	logdb.Connect()
	return &logdb
}

// Ping 是否Ping通数据库
func (logdb *LogDB) Ping() (result bool) {
	defer func() {
		if err := recover(); err != nil {
			result = false

			log.Println(err, logdb.Hosts[logdb.hostid], " is unconnect ")
			hostlen := len(logdb.Hosts)
			for i := 0; i < hostlen; i++ {
				db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?timeout=15s&charset=%s", logdb.User, logdb.Password, logdb.Hosts[i], logdb.Port, logdb.DB, logdb.Charset))
				if err != nil {
					log.Println(err, logdb.Hosts[i], " is connect fail")
					continue
				}

				if err := db.Ping(); err != nil {
					log.Println(err, logdb.Hosts[i], " is connect fail")
					continue
				}

				logdb.driver = db
				logdb.hostid = i
				result = true
			}
		}
	}()

	if err := logdb.driver.Ping(); err != nil {
		panic(err)
	}

	return true
}

// Connect 重连
func (logdb *LogDB) Connect() {
	logdb.hostid++
	if logdb.hostid >= len(logdb.Hosts) {
		logdb.hostid = 0
	}
	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?timeout=15s&charset=%s", logdb.User, logdb.Password, logdb.Hosts[logdb.hostid], logdb.Port, logdb.DB, logdb.Charset))
	if err != nil {
		panic(err)
	}
	if logdb.driver != nil {
		logdb.driver.Close()
	}
	logdb.driver = db
	log.Println("connect is", logdb.Ping(), logdb)
}

// ADInsert 插入数据
func (logdb *LogDB) ADInsert(uid, device, platform, area_cc, section_id, response string, spider_id, channel, media, catch_account_id, status, priority int, ts_crawl time.Time) {
	_, err := logdb.driver.Exec("insert into log_spider (uid, spider_id, device, platform, channel, media, area_cc, catch_account_id, section_id, response, status, priority, ts_crawl) value(?, ?, ?, ?, ?, ?, ?, ?, ?, ? , ? ,? ,? ,?)", uid, spider_id, device, platform, channel, media, area_cc, catch_account_id, section_id, response, status, priority, ts_crawl)
	if err != nil {
		log.Println(err)
	}
}

// Select 插入数据
func (logdb *LogDB) Select(query string, args ...interface{}) *sql.Rows {
	Rows, err := logdb.driver.Query(query, args...)
	if err != nil {
		log.Println(err)
	}
	return Rows
}

// ADError 广告错误后更新
func (logdb *LogDB) ADError(uid, error_msg string) {
	_, err := logdb.driver.Exec("update log_spider set status = status + 1000, error_msg=?  where uid =?;", logdb.DB, error_msg, uid)
	if err != nil {
		log.Println(err)
	}
}