push/pushUnbinding/task/task.go

package task

import (
	util "app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/date"
	elastic "app.yhyue.com/moapp/jybase/es"
	"app.yhyue.com/moapp/jybase/logger"
	qrpc "app.yhyue.com/moapp/jybase/rpc"
	. "bp.jydev.jianyu360.cn/BaseService/pushpkg/p"
	"fmt"
	"pushUnbinding/config"
	"strconv"
	"strings"
	"sync"
	"time"
)

type TaskInfo struct{}

var (
	Count = 0
	Info  = make(map[string]interface{})
)

func (t *TaskInfo) Run() {
	logger.Info("开始TaskInfo")
	getBidInfo() //获取最新标讯信息
	if Count == 0 {
		logger.Info("未获取到最新标讯信息")
		return
	}
	taskInfo()
}

func getBidInfo() {
	query := `{"query": {"bool": {"must": [{"range": {"pici": {"gte": ` + fmt.Sprint(config.TaskConfig.Pici) + `,"lt": ` + fmt.Sprint(time.Now().Unix()) + `}}},{"bool": {"should": [{"terms": {"toptype": ["预告","招标","结果","其它"]}}],"minimum_should_match": 1}}]}},"_source": ["id","title","projectname","publishtime","buyer","area","pici"],"sort": [{"publishtime": "desc"}],"size": 1}`
	countQ := `{"query": {"bool": {"must": [{"range": {"pici": {"gte": ` + util.ObjToString(config.TaskConfig.Pici) + `,"lt": ` + util.ObjToString(time.Now().Unix()) + `}}},{"bool": {"should": [{"terms": {"toptype": ["预告","招标","结果","其它"]}}],"minimum_should_match": 1}}]}}}`
	logger.Info("countQ: ", countQ)
	c, info := elastic.GetWithCount("bidding", "", countQ, query)
	if c > 0 {
		Count = int(c)
		Info = (*info)[0]
	}
}

func taskInfo() {
	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	q := map[string]interface{}{
		"i_appid":    2,
		"s_m_openid": map[string]int{"$exists": 1},
		"i_ispush":   map[string]int{"$ne": 0},
		"s_phone":    map[string]int{"$exists": 0},
		"s_m_phone":  map[string]int{"$exists": 0}}
	f := map[string]interface{}{"s_m_openid": 1}
	it := config.MgoJy.GetMgoConn().DB(config.Config.MongoDb.JianYu.DbName).C("user").Find(q).Select(f).Iter()
	logger.Info("查询数据量：", config.MgoJy.Count("user", q))
	count := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); count++ {
		if count%2000 == 0 {
			logger.Info("current ---", count)
		}
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			sendWeixin(tmp)
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	Count = 0
	Info = make(map[string]interface{})
	logger.Info("轮次结束 ---", count)
}

// 微信远程调用，实现模板发送消息
func sendWeixin(user map[string]interface{}) {
	config.TaskConfig.Pici = util.Int64All(Info["pici"])
	util.WriteSysConfig("./task.json", &config.TaskConfig)
	pb := util.Int64All(Info["publishtime"])
	pname := util.ObjToString(Info["projectname"])
	buyer := util.ObjToString(Info["buyer"])
	tmplData := map[string]*qrpc.TmplItem{
		"thing1": { //
			Value: strings.ReplaceAll(pname, buyer, ""),
			Color: config.Config.WxTplMsg.KeyWord1.Color,
		},
		"thing6": {
			Value: buyer,
			Color: config.Config.WxTplMsg.KeyWord2.Color,
		},
		"character_string4": {
			Value: fmt.Sprint(Count),
			Color: config.Config.WxTplMsg.KeyWord3.Color,
		},
		"time3": {
			Value: date.FormatDateByInt64(&pb, date.Date_Full_Layout),
			Color: config.Config.WxTplMsg.KeyWord4.Color,
		},
	}
	wxTmplMsg := &qrpc.WxTmplMsg{
		OpenId:   util.ObjToString(user["s_m_openid"]),
		TplId:    config.Config.WxTplMsg.Id,
		TmplData: tmplData,
		Url:      config.Config.WxDomain + "/front/sess/" + Se.EncodeString(util.ObjToString(Info["id"])+",_id,"+strconv.Itoa(int(time.Now().Unix()))+",") + "__" + Se.EncodeString(config.Config.WxTplMsg.Url),
		//Url:      fmt.Sprintf(config.Config.WxDomain+config.Config.WxTplMsg.Url, encrypt.CommonEncodeArticle("content", util.ObjToString(Info["id"]))),
	}
	_, err := qrpc.WxSendTmplMsg(config.Config.WxRpcServer, wxTmplMsg)
	if err != nil {
		logger.Info("wx模版消息发送失败", err)
	}
}