data_processing
/
spider_editor


			
				
					
						
						
							123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202
							package timetask

import (
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"net/smtp"
	qu "qfw/util"
	mgdb "qfw/util/mongodb"
	mgu "qfw/util/mongodbutil"
	sp "spiderutil"
	"strconv"
	"strings"
	"time"
	"util"

	"github.com/cron"

	"gopkg.in/mgo.v2/bson"
	. "gopkg.in/mgo.v2/bson"
)

var Mail map[string]interface{}

func TimeTask() {
	defer qu.Catch()
	c := cron.New()
	c.Start()
	c.AddFunc("0 20 9 ? * MON-FRI", CheckCreateTask)
	c.AddFunc("0 0 */1 ? * *", CheckLuaMove)
	c.AddFunc("0 30 23 * * *", UpdateSiteInfo) //定时更新站点信息
	c.AddFunc("0 0 23 * * *", UpdateCodeHeart) //定时更新爬虫心跳信息
}

//监测爬虫由历史转增量时未成功的
func CheckLuaMove() {
	defer qu.Catch()
	qu.Debug("开始检测爬虫节点移动...")
	query := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$gte": time.Now().Add(-(time.Hour * 1)).Unix(),
			"$lte": time.Now().Unix(),
		},
		"ok": false,
	}
	qu.Debug("query:", query)
	list := *mgdb.Find("luamovelog", query, nil, nil, false, -1, -1)
	text := ""
	if len(list) > 0 {
		for _, l := range list {
			stype := qu.ObjToString(l["type"])
			code := qu.ObjToString(l["code"])
			text += code + "：" + stype + "；"
		}
	}
	if text != "" {
		for i := 1; i <= 3; i++ {
			res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", sp.Config.JkMail["api"], sp.Config.JkMail["to"], "lua-move-fail", text))
			if err == nil {
				res.Body.Close()
				read, err := ioutil.ReadAll(res.Body)
				qu.Debug("邮件发送：", string(read), err)
				break
			}
		}
	}
}

//检测创建任务失败的爬虫
func CheckCreateTask() {
	defer qu.Catch()
	qu.Debug("开始检测任务创建...")
	query := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$gte": GetTime(0),
		},
	}
	codes := []string{}
	list := *mgdb.Find("luacreatetaskerr", query, nil, nil, false, -1, -1)
	if len(list) > 0 {
		for _, l := range list {
			code := qu.ObjToString(l["code"])
			codes = append(codes, code)
		}
	}
	if len(codes) > 0 {
		for i := 1; i <= 3; i++ {
			res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", sp.Config.JkMail["api"], sp.Config.JkMail["to"], "lua-createtask-err", "爬虫："+strings.Join(codes, "；")))
			if err == nil {
				res.Body.Close()
				read, err := ioutil.ReadAll(res.Body)
				qu.Debug("邮件发送：", string(read), err)
				break
			}
		}
	}
}

func SendToMail() {
	mailInfo := *(qu.ObjToMap(Mail["smtp"]))
	host := mailInfo["host"].(string)
	from := mailInfo["from"].(string)
	pwd := mailInfo["password"].(string)
	subject := mailInfo["subject"].(string)

	hour := time.Now().Hour()
	if hour == 8 {
		//定时查询数据库 查询需要发邮件的人和相关信息
		timeStr := time.Now().Format("2006-01-02")
		the_time, _ := time.ParseInLocation("2006-01-02", timeStr, time.Local)
		time_zero := the_time.Unix()         //当日凌晨的时间戳
		time_twentyFour := time_zero + 86399 //当日24时的时间戳
		//聚合查询数据
		//mgdb.InitMongodbPool(5, "192.168.3.207:27080", "editor")
		sess := mgdb.GetMgoConn()
		defer mgdb.DestoryMongoConn(sess)
		var res []M
		sess.DB("editor").C("task").Pipe([]M{M{"$match": M{"l_complete": M{"$gte": time_zero, "$lte": time_twentyFour}, "i_state": M{"$gte": 1, "$lte": 2}}},
			M{"$group": M{"_id": "$s_modifyid", "count": M{"$sum": 1}}}}).All(&res)
		//遍历数据进行发邮件
		for _, v := range res {
			_id, ok := v["_id"].(string)
			if ok {
				query := bson.M{
					"_id": bson.ObjectIdHex(_id),
				}
				user := *mgdb.FindOne("user", query)
				if user["s_email"] == nil {
					continue
				}
				num := strconv.Itoa(v["count"].(int))
				body := `<html><body><h3>你有` + num + `条任务需要今天完成</h3></body></html>`
				hp := strings.Split(host, ":")
				auth := smtp.PlainAuth("", from, pwd, hp[0])
				content_type := "Content-Type: text/html; charset=UTF-8"
				msg := []byte("To: " + user["s_email"].(string) + "\r\nFrom: " + from + "\r\nSubject: " + subject + "\r\n" + content_type + "\r\n\r\n" + body)
				send_to := strings.Split(user["s_email"].(string), ";")
				err := smtp.SendMail(host, auth, from, send_to, msg)
				if err == nil {
					log.Println(user["s_email"].(string), "  sendMail   success")
				} else {
					log.Println(user["s_email"].(string), "  sendMail   fail")
				}
			}
		}
		time.Sleep(1 * time.Hour)
	}
	//time.AfterFunc(30*time.Minute, func() { SendToMail(to, num) })
	time.AfterFunc(30*time.Minute, SendToMail)
}

//获取第day天凌晨的时间戳
func GetTime(day int) int64 {
	defer qu.Catch()
	nowTime := time.Now().AddDate(0, 0, day)
	timeStr := qu.FormatDate(&nowTime, qu.Date_Short_Layout)
	t, _ := time.ParseInLocation(qu.Date_Short_Layout, timeStr, time.Local)
	return t.Unix()
}

func UpdateSiteInfo() {
	defer qu.Catch()
	qu.Debug("定时更新站点信息开始...")
	sites, _ := util.MgoE.Find(sp.Config.SiteColl, ``, ``, `{"site":1}`, false, -1, -1)
	for _, s := range *sites {
		site := qu.ObjToString(s["site"])
		domain, status, event, platform, area, city, district, _ := util.GetLuasInfoBySite(site)
		set := map[string]interface{}{
			"$set": map[string]interface{}{
				"platform":      platform,
				"event":         event,
				"spider_status": status,
				"domain":        domain,
				"area":          area,
				"city":          city,
				"district":      district,
				"updatetime":    time.Now().Unix(),
			},
		}
		util.MgoE.UpdateById(sp.Config.SiteColl, s["_id"], set)
	}
	qu.Debug("定时更新站点信息完成...")

}

func UpdateCodeHeart() {
	qu.Debug("定时更新爬虫心跳信息...")
	defer qu.Catch()
	query := map[string]interface{}{
		"state": map[string]interface{}{
			"$gt": 6,
		},
	}
	list, _ := util.MgoE.Find("luaconfig", query, nil, map[string]interface{}{"code": 1}, false, -1, -1)
	for _, l := range *list {
		mgu.Update("spider_heart", "spider", "spider", map[string]interface{}{"code": l["code"]}, map[string]interface{}{
			"$set": map[string]interface{}{"del": true},
		}, false, false)
	}
	qu.Debug("定时更新爬虫心跳信息完成...")
}