|
|
@@ -1,18 +1,22 @@
|
|
|
package job
|
|
|
|
|
|
import (
|
|
|
+ "fmt"
|
|
|
"net/http"
|
|
|
+ "os"
|
|
|
. "public"
|
|
|
. "push/config"
|
|
|
putil "push/util"
|
|
|
"qfw/util"
|
|
|
+ "qfw/util/mail"
|
|
|
"qfw/util/mongodb"
|
|
|
- "sort"
|
|
|
+ "qfw/util/redis"
|
|
|
+ "strconv"
|
|
|
+ "strings"
|
|
|
"sync"
|
|
|
"time"
|
|
|
|
|
|
"github.com/donnie4w/go-logger/logger"
|
|
|
- mgo "gopkg.in/mgo.v2"
|
|
|
"gopkg.in/mgo.v2/bson"
|
|
|
)
|
|
|
|
|
|
@@ -20,276 +24,33 @@ var (
|
|
|
MoveFields = []string{"s_m_openid", "a_m_openid", "phone", "usertype", "jpushid", "opushid", "words", "ratemode", "wxpush", "apppush", "mailpush", "pchelperpush", "timestamp", "subscribe", "applystatus", "appphonetype", "email", "size", "modifydate", "mergeorder", "nickname", "firstpushtime"}
|
|
|
)
|
|
|
|
|
|
-type Move struct {
|
|
|
- Info map[string]interface{}
|
|
|
- Ids []interface{}
|
|
|
+func init() {
|
|
|
+ if Config.MinutePushSize > 0 {
|
|
|
+ putil.LimitMaxOneMinutePush(&Jobs.Push.minutePushPool, Config.MinutePushSize)
|
|
|
+ }
|
|
|
+ if Config.FastigiumMinutePushSize > 0 {
|
|
|
+ putil.LimitMaxOneMinutePush(&Jobs.Push.fastigiumMinutePushPool, Config.FastigiumMinutePushSize)
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
type pushJob struct {
|
|
|
- taskType int
|
|
|
- pool chan bool
|
|
|
- wait *sync.WaitGroup
|
|
|
- lock *sync.Mutex
|
|
|
- moveLock *sync.Mutex
|
|
|
- moveWait *sync.WaitGroup
|
|
|
- movePool chan bool
|
|
|
- mergeLock *sync.Mutex
|
|
|
- mergeWait *sync.WaitGroup
|
|
|
- mergePool chan bool
|
|
|
- lastId string
|
|
|
- users *[]map[string]interface{}
|
|
|
+ taskType int
|
|
|
+ pool chan bool
|
|
|
+ wait *sync.WaitGroup
|
|
|
+ lock *sync.Mutex
|
|
|
+ lastId string
|
|
|
+ users *[]map[string]interface{}
|
|
|
+ vipTempSave bool
|
|
|
+ minutePushPool chan bool
|
|
|
+ fastigiumMinutePushPool chan bool
|
|
|
}
|
|
|
|
|
|
-//taskType 1--实时推送 2--实时推送+一天三次的8点推送 3--一天三次推送 4--九点推送
|
|
|
-func (p *pushJob) Execute(taskType int, isMoveDatas bool) {
|
|
|
+//taskType 1--一天三次推送 2--九点推送
|
|
|
+func (p *pushJob) Execute(taskType int) {
|
|
|
+ defer util.Catch()
|
|
|
p.lock.Lock()
|
|
|
defer p.lock.Unlock()
|
|
|
p.taskType = taskType
|
|
|
- logger.Info("开始推送任务。。。", p.taskType)
|
|
|
- if isMoveDatas {
|
|
|
- p.Move()
|
|
|
- }
|
|
|
- p.Push()
|
|
|
-}
|
|
|
-func (p *pushJob) Move() {
|
|
|
- defer util.Catch()
|
|
|
- logger.Info("推送任务", p.taskType, "开始迁移数据。。。")
|
|
|
- nowUnix := time.Now().Unix()
|
|
|
- sess := mongodb.GetMgoConn()
|
|
|
- defer mongodb.DestoryMongoConn(sess)
|
|
|
- it := sess.DB(DbName).C("pushspace_temp").Find(map[string]interface{}{
|
|
|
- "timestamp": map[string]interface{}{
|
|
|
- "$lt": nowUnix,
|
|
|
- },
|
|
|
- }).Sort("userid").Iter()
|
|
|
- moves := map[string]*Move{}
|
|
|
- logger.Info("推送任务", p.taskType, "开始遍历pushspace_temp")
|
|
|
- index, number, length := 0, 0, 0
|
|
|
- for data := make(map[string]interface{}); it.Next(&data); {
|
|
|
- p.movePool <- true
|
|
|
- p.moveWait.Add(1)
|
|
|
- index++
|
|
|
- go func(temp map[string]interface{}) {
|
|
|
- defer func() {
|
|
|
- <-p.movePool
|
|
|
- p.moveWait.Done()
|
|
|
- p.moveLock.Unlock()
|
|
|
- }()
|
|
|
- userId := util.ObjToString(temp["userid"])
|
|
|
- p.moveLock.Lock()
|
|
|
- move := moves[userId]
|
|
|
- if move != nil {
|
|
|
- list, _ := move.Info["list"].(SortList)
|
|
|
- idMap := map[string]bool{}
|
|
|
- for _, v := range list {
|
|
|
- idMap[util.ObjToString((*v.Info)["_id"])] = true
|
|
|
- }
|
|
|
- newList := putil.ToSortList(temp["list"])
|
|
|
- for _, v := range newList {
|
|
|
- if idMap[util.ObjToString((*v.Info)["_id"])] {
|
|
|
- continue
|
|
|
- }
|
|
|
- list = append(list, v)
|
|
|
- }
|
|
|
- sort.Sort(list)
|
|
|
- if len(list) > SysConfig.MaxPushSize {
|
|
|
- list = list[:SysConfig.MaxPushSize]
|
|
|
- }
|
|
|
- temp["list"] = list
|
|
|
- move.Info = temp
|
|
|
- move.Ids = append(move.Ids, temp["_id"])
|
|
|
- } else {
|
|
|
- temp["list"] = putil.ToSortList(temp["list"])
|
|
|
- move = &Move{
|
|
|
- Info: temp,
|
|
|
- Ids: []interface{}{temp["_id"]},
|
|
|
- }
|
|
|
- }
|
|
|
- moves[userId] = move
|
|
|
- length++
|
|
|
- if length == SysConfig.MoveBatch {
|
|
|
- p.Merge(&number, nowUnix, moves)
|
|
|
- length = 0
|
|
|
- moves = map[string]*Move{}
|
|
|
- }
|
|
|
- }(data)
|
|
|
- data = make(map[string]interface{})
|
|
|
- if index%500 == 0 {
|
|
|
- logger.Info("推送任务", p.taskType, "pushspace_temp加载到内存：", index)
|
|
|
- }
|
|
|
- }
|
|
|
- p.moveWait.Wait()
|
|
|
- if length > 0 {
|
|
|
- p.Merge(&number, nowUnix, moves)
|
|
|
- length = 0
|
|
|
- moves = map[string]*Move{}
|
|
|
- }
|
|
|
- logger.Info("推送任务", p.taskType, "迁移数据结束。。。", index)
|
|
|
-}
|
|
|
-func (p *pushJob) Merge(number *int, nowUnix int64, moves map[string]*Move) {
|
|
|
- *number++
|
|
|
- logger.Info("推送任务", p.taskType, "第", *number, "次开始合并数据")
|
|
|
- index := 0
|
|
|
- saveArray := []map[string]interface{}{}
|
|
|
- saveArray_delete := []interface{}{}
|
|
|
- updateArray_query := []map[string]interface{}{}
|
|
|
- updateArray_set := []map[string]interface{}{}
|
|
|
- updateArray_delete := []interface{}{}
|
|
|
- for u, m := range moves {
|
|
|
- p.mergePool <- true
|
|
|
- p.mergeWait.Add(1)
|
|
|
- go func(userId string, move *Move) {
|
|
|
- defer func() {
|
|
|
- <-p.mergePool
|
|
|
- p.mergeWait.Done()
|
|
|
- }()
|
|
|
- sess := mongodb.GetMgoConn()
|
|
|
- defer mongodb.DestoryMongoConn(sess)
|
|
|
- var data map[string]interface{}
|
|
|
- sess.DB(DbName).C("pushspace").Find(map[string]interface{}{"userid": userId}).Select(map[string]interface{}{"list": 1, "templist": 1}).One(&data)
|
|
|
- if data == nil { //批量新增
|
|
|
- p.mergeLock.Lock()
|
|
|
- saveArray = append(saveArray, move.Info)
|
|
|
- saveArray_delete = append(saveArray_delete, move.Ids...)
|
|
|
- if len(saveArray) == BulkSize {
|
|
|
- p.SaveBulk(sess, &saveArray, &saveArray_delete)
|
|
|
- }
|
|
|
- p.mergeLock.Unlock()
|
|
|
- } else { //批量更新
|
|
|
- setMap := map[string]interface{}{}
|
|
|
- for _, field := range MoveFields {
|
|
|
- if move.Info[field] == nil {
|
|
|
- continue
|
|
|
- }
|
|
|
- setMap[field] = move.Info[field]
|
|
|
- }
|
|
|
- //
|
|
|
- newListOrig, _ := move.Info["list"].(SortList)
|
|
|
- if newListOrig == nil || len(newListOrig) == 0 {
|
|
|
- return
|
|
|
- }
|
|
|
- pushAll := make(map[string]interface{})
|
|
|
- for _, v := range []string{"", "temp"} {
|
|
|
- oldList := putil.ToSortList(data[v+"list"])
|
|
|
- if v == "temp" && oldList == nil {
|
|
|
- continue
|
|
|
- }
|
|
|
- idMap := map[string]bool{}
|
|
|
- for _, vv := range oldList {
|
|
|
- idMap[util.ObjToString((*vv.Info)["_id"])] = true
|
|
|
- }
|
|
|
- newList := make(SortList, 0)
|
|
|
- //去重
|
|
|
- for _, vv := range newListOrig {
|
|
|
- if idMap[util.ObjToString((*vv.Info)["_id"])] {
|
|
|
- continue
|
|
|
- }
|
|
|
- newList = append(newList, vv)
|
|
|
- }
|
|
|
- pLength := len(newList)
|
|
|
- if pLength == 0 {
|
|
|
- continue
|
|
|
- }
|
|
|
- rLength := len(oldList)
|
|
|
- if rLength+pLength > SysConfig.MaxPushSize {
|
|
|
- newList = append(newList, oldList...)
|
|
|
- sort.Sort(newList)
|
|
|
- setMap[v+"list"] = newList[:SysConfig.MaxPushSize]
|
|
|
- setMap[v+"size"] = SysConfig.MaxPushSize
|
|
|
- } else { //追加
|
|
|
- setMap[v+"size"] = rLength + pLength
|
|
|
- pushAll[v+"list"] = newList
|
|
|
- }
|
|
|
- }
|
|
|
- upSet := map[string]interface{}{
|
|
|
- "$set": setMap,
|
|
|
- }
|
|
|
- if len(pushAll) > 0 {
|
|
|
- upSet["$pushAll"] = pushAll
|
|
|
- }
|
|
|
- p.mergeLock.Lock()
|
|
|
- updateArray_delete = append(updateArray_delete, move.Ids...)
|
|
|
- updateArray_query = append(updateArray_query, map[string]interface{}{"_id": data["_id"]})
|
|
|
- updateArray_set = append(updateArray_set, upSet)
|
|
|
- if len(updateArray_query) == BulkSize {
|
|
|
- p.UpdateBulk(sess, &updateArray_query, &updateArray_set, &updateArray_delete)
|
|
|
- }
|
|
|
- p.mergeLock.Unlock()
|
|
|
- }
|
|
|
- }(u, m)
|
|
|
- index++
|
|
|
- if index%500 == 0 {
|
|
|
- logger.Info("推送任务", p.taskType, "第", *number, "次合并数据：", index)
|
|
|
- }
|
|
|
- }
|
|
|
- p.mergeWait.Wait()
|
|
|
- sess := mongodb.GetMgoConn()
|
|
|
- defer mongodb.DestoryMongoConn(sess)
|
|
|
- if len(saveArray) > 0 {
|
|
|
- p.SaveBulk(sess, &saveArray, &saveArray_delete)
|
|
|
- }
|
|
|
- if len(updateArray_query) > 0 {
|
|
|
- p.UpdateBulk(sess, &updateArray_query, &updateArray_set, &updateArray_delete)
|
|
|
- }
|
|
|
- logger.Info("推送任务", p.taskType, "第", *number, "次合并数据结束。。。", index)
|
|
|
-}
|
|
|
-func (p *pushJob) SaveBulk(sess *mgo.Session, saves *[]map[string]interface{}, deletes *[]interface{}) {
|
|
|
- coll := sess.DB(DbName).C("pushspace")
|
|
|
- bulk := coll.Bulk()
|
|
|
- for _, v := range *saves {
|
|
|
- bulk.Insert(v)
|
|
|
- }
|
|
|
- _, err := bulk.Run()
|
|
|
- if nil != err {
|
|
|
- logger.Info("BulkError", err)
|
|
|
- } else {
|
|
|
- p.DelBulk(sess, deletes)
|
|
|
- }
|
|
|
- *saves = []map[string]interface{}{}
|
|
|
-}
|
|
|
-func (p *pushJob) UpdateBulk(sess *mgo.Session, array_q, array_s *[]map[string]interface{}, array_d *[]interface{}) {
|
|
|
- coll := sess.DB(DbName).C("pushspace")
|
|
|
- bulk := coll.Bulk()
|
|
|
- for k, v := range *array_q {
|
|
|
- bulk.Update(v, (*array_s)[k])
|
|
|
- }
|
|
|
- _, err := bulk.Run()
|
|
|
- if nil != err {
|
|
|
- logger.Info("UpdateBulkError", err)
|
|
|
- } else {
|
|
|
- p.DelBulk(sess, array_d)
|
|
|
- }
|
|
|
- *array_q = []map[string]interface{}{}
|
|
|
- *array_s = []map[string]interface{}{}
|
|
|
-}
|
|
|
-func (p *pushJob) DelBulk(sess *mgo.Session, array *[]interface{}) {
|
|
|
- coll := sess.DB(DbName).C("pushspace_temp")
|
|
|
- count := 0
|
|
|
- bulk := coll.Bulk()
|
|
|
- for _, v := range *array {
|
|
|
- count++
|
|
|
- bulk.Remove(map[string]interface{}{
|
|
|
- "_id": v,
|
|
|
- })
|
|
|
- if count == BulkSize {
|
|
|
- _, err := bulk.Run()
|
|
|
- if nil != err {
|
|
|
- logger.Info("DelBulkError", err)
|
|
|
- }
|
|
|
- count = 0
|
|
|
- bulk = coll.Bulk()
|
|
|
- }
|
|
|
- }
|
|
|
- if count > 0 {
|
|
|
- _, err := bulk.Run()
|
|
|
- if nil != err {
|
|
|
- logger.Info("DelBulkError", err)
|
|
|
- }
|
|
|
- }
|
|
|
- *array = []interface{}{}
|
|
|
-}
|
|
|
-func (p *pushJob) Push() {
|
|
|
- defer util.Catch()
|
|
|
logger.Info("推送任务", p.taskType, "开始推送。。。")
|
|
|
batch_index := 0
|
|
|
for {
|
|
|
@@ -300,10 +61,10 @@ func (p *pushJob) Push() {
|
|
|
select {
|
|
|
case <-time.After(5 * time.Minute):
|
|
|
isTake = false
|
|
|
- logger.Error("推送任务", p.taskType, "推送放入通道超时，", temp["userid"], len(DoPush.minutePushPool), len(DoPush.fastigiumMinutePushPool))
|
|
|
+ logger.Error("推送任务", p.taskType, "推送放入通道超时，", temp["userid"], len(p.minutePushPool), len(p.fastigiumMinutePushPool))
|
|
|
go func() {
|
|
|
- if SysConfig.TimeoutWarn != "" {
|
|
|
- if _, err := http.Get(SysConfig.TimeoutWarn); err != nil {
|
|
|
+ if Config.TimeoutWarn != "" {
|
|
|
+ if _, err := http.Get(Config.TimeoutWarn); err != nil {
|
|
|
logger.Error("发送告警邮件错误", err)
|
|
|
}
|
|
|
}
|
|
|
@@ -344,56 +105,17 @@ func (p *pushJob) Push() {
|
|
|
}
|
|
|
logger.Info("推送任务", p.taskType, "开始推送用户", "userType", u.UserType, "userId", u.Id, "s_m_openid", u.S_m_openid, "a_m_openid", u.A_m_openid, "phone", u.Phone, "subscribe", u.Subscribe, "applystatus", u.ApplyStatus, "jpushid", u.Jpushid, "opushid", u.Opushid, "phoneType", u.AppPhoneType, "rateMode", u.RateMode, "email", u.Email)
|
|
|
wxPush, appPush, mailPush := 0, 0, 0
|
|
|
- if p.taskType == 1 {
|
|
|
- if u.WxPush == 1 {
|
|
|
- if u.ApplyStatus == 1 {
|
|
|
- wxPush = -1
|
|
|
- } else {
|
|
|
- wxPush = 1
|
|
|
- }
|
|
|
- }
|
|
|
- if u.AppPush == 1 {
|
|
|
- appPush = 1
|
|
|
- }
|
|
|
- if u.MailPush == 1 {
|
|
|
- if u.UserType == 0 && u.ApplyStatus == 1 {
|
|
|
- mailPush = -1
|
|
|
- } else if u.UserType == 5 && u.ApplyStatus == 1 && u.AppPush == 0 {
|
|
|
- mailPush = -1
|
|
|
- } else {
|
|
|
- mailPush = 1
|
|
|
- }
|
|
|
- }
|
|
|
- } else if p.taskType == 2 || p.taskType == 4 {
|
|
|
- if u.WxPush == 1 {
|
|
|
- wxPush = 1
|
|
|
- }
|
|
|
- if u.AppPush == 1 {
|
|
|
- appPush = 1
|
|
|
- }
|
|
|
- if u.MailPush == 1 {
|
|
|
- mailPush = 1
|
|
|
- }
|
|
|
- } else if p.taskType == 3 {
|
|
|
- if u.WxPush == 1 && u.ApplyStatus == 1 {
|
|
|
- wxPush = 1
|
|
|
- }
|
|
|
- if u.MailPush == 1 {
|
|
|
- mailPush = 1
|
|
|
- }
|
|
|
+ if u.WxPush == 1 {
|
|
|
+ wxPush = 1
|
|
|
}
|
|
|
- list := putil.ToSortList(v["list"])
|
|
|
- tempList := putil.ToSortList(v["templist"])
|
|
|
- t_wxPush, t_mailPush := util.IntAll(v["tempwxpush"]), util.IntAll(v["tempmailpush"])
|
|
|
- if tempList != nil {
|
|
|
- if wxPush == 1 && t_wxPush == 0 {
|
|
|
- wxPush = 0
|
|
|
- }
|
|
|
- if mailPush == 1 && t_mailPush == 0 {
|
|
|
- mailPush = 0
|
|
|
- }
|
|
|
+ if u.AppPush == 1 {
|
|
|
+ appPush = 1
|
|
|
}
|
|
|
- logger.Info("推送任务", p.taskType, "用户接收方式", "userId", u.Id, "wxPush", wxPush, "appPush", appPush, "mailPush", mailPush, "pchelperPush", u.PchelperPush, "t_wxPush", t_wxPush, "t_mailPush", t_mailPush)
|
|
|
+ if u.MailPush == 1 {
|
|
|
+ mailPush = 1
|
|
|
+ }
|
|
|
+ list := putil.ToSortList(v["list"])
|
|
|
+ logger.Info("推送任务", p.taskType, "用户接收方式", "userId", u.Id, "wxPush", wxPush, "appPush", appPush, "mailPush", mailPush, "pchelperPush", u.PchelperPush)
|
|
|
if wxPush != 1 && appPush != 1 && mailPush != 1 {
|
|
|
return
|
|
|
}
|
|
|
@@ -413,7 +135,7 @@ func (p *pushJob) Push() {
|
|
|
mailPush = 0
|
|
|
}
|
|
|
}
|
|
|
- isSaveSuccess, wxStatus, appStatus, mailStatus := DoPush.Execute(p.taskType, wxPush, appPush, mailPush, u, list, tempList)
|
|
|
+ isSaveSuccess, wxStatus, appStatus, mailStatus := p.Push(p.taskType, wxPush, appPush, mailPush, u, list)
|
|
|
if isSaveSuccess {
|
|
|
if u.FirstPushTime == 0 {
|
|
|
go mongodb.Update("user", map[string]interface{}{
|
|
|
@@ -430,28 +152,9 @@ func (p *pushJob) Push() {
|
|
|
//判断是否要删除数据
|
|
|
sess := mongodb.GetMgoConn()
|
|
|
defer mongodb.DestoryMongoConn(sess)
|
|
|
- if wxPush == -1 || mailPush == -1 {
|
|
|
- //如果该用户还有微信或者邮箱推送，把list字段的值挪到templist
|
|
|
- update := map[string]interface{}{}
|
|
|
- set := map[string]interface{}{
|
|
|
- "tempwxpush": wxPush,
|
|
|
- "tempmailpush": mailPush,
|
|
|
- }
|
|
|
- if tempList == nil {
|
|
|
- update["$rename"] = map[string]interface{}{"list": "templist"}
|
|
|
- } else {
|
|
|
- update["$unset"] = map[string]interface{}{"list": ""}
|
|
|
- }
|
|
|
- update["$set"] = set
|
|
|
- err := sess.DB(DbName).C("pushspace").UpdateId(v["_id"], update)
|
|
|
- if err != nil {
|
|
|
- logger.Error("推送任务", p.taskType, "update error", err)
|
|
|
- }
|
|
|
- } else {
|
|
|
- err := sess.DB(DbName).C("pushspace").RemoveId(v["_id"])
|
|
|
- if err != nil {
|
|
|
- logger.Error("推送任务", p.taskType, "remove error", err)
|
|
|
- }
|
|
|
+ err := sess.DB(DbName).C("pushspace").RemoveId(v["_id"])
|
|
|
+ if err != nil {
|
|
|
+ logger.Error("推送任务", p.taskType, "remove error", err)
|
|
|
}
|
|
|
if wxStatus == -1 || appStatus == -1 || mailStatus == -1 {
|
|
|
f_count, err := sess.DB(DbName).C("pushspace").FindId(v["_id"]).Count()
|
|
|
@@ -493,7 +196,7 @@ func (p *pushJob) Push() {
|
|
|
}
|
|
|
}(temp, isTake)
|
|
|
}
|
|
|
- if batch_size < SysConfig.PushBatch {
|
|
|
+ if batch_size < Config.PushBatch {
|
|
|
break
|
|
|
}
|
|
|
}
|
|
|
@@ -509,16 +212,11 @@ func (p *pushJob) OncePushBatch(batch_index int) int {
|
|
|
defer mongodb.DestoryMongoConn(sess)
|
|
|
var query map[string]interface{}
|
|
|
//根据任务类型，查找ratemode
|
|
|
- if p.taskType == 1 || p.taskType == 2 {
|
|
|
+ if p.taskType == 1 {
|
|
|
query = map[string]interface{}{
|
|
|
"ratemode": 1,
|
|
|
}
|
|
|
- } else if p.taskType == 3 {
|
|
|
- query = map[string]interface{}{
|
|
|
- "ratemode": 1,
|
|
|
- "applystatus": 1,
|
|
|
- }
|
|
|
- } else if p.taskType == 4 {
|
|
|
+ } else if p.taskType == 2 {
|
|
|
query = map[string]interface{}{
|
|
|
"ratemode": 2,
|
|
|
}
|
|
|
@@ -526,9 +224,9 @@ func (p *pushJob) OncePushBatch(batch_index int) int {
|
|
|
logger.Error("taskType error", p.taskType)
|
|
|
return i
|
|
|
}
|
|
|
- if len(SysConfig.TestIds) > 0 {
|
|
|
+ if len(Config.TestIds) > 0 {
|
|
|
query["userid"] = map[string]interface{}{
|
|
|
- "$in": SysConfig.TestIds,
|
|
|
+ "$in": Config.TestIds,
|
|
|
}
|
|
|
}
|
|
|
if p.lastId != "" {
|
|
|
@@ -543,10 +241,320 @@ func (p *pushJob) OncePushBatch(batch_index int) int {
|
|
|
p.lastId = util.BsonIdToSId(temp["_id"])
|
|
|
*p.users = append(*p.users, temp)
|
|
|
temp = make(map[string]interface{})
|
|
|
- if i == SysConfig.PushBatch {
|
|
|
+ if i == Config.PushBatch {
|
|
|
break
|
|
|
}
|
|
|
}
|
|
|
logger.Info("推送任务", p.taskType, "第", batch_index, "批用户加载结束", p.lastId)
|
|
|
return i
|
|
|
}
|
|
|
+
|
|
|
+func (p *pushJob) Push(taskType int, wxPush, appPush, mailPush int, u *UserInfo, list SortList) (isSaveSuccess bool, wxStatus, appStatus, mailStatus int) {
|
|
|
+ if wxPush == 1 || appPush == 1 || mailPush == 1 || u.PchelperPush == 1 {
|
|
|
+ if list != nil {
|
|
|
+ isSaveSuccess, wxStatus, appStatus, mailStatus = p.DoPush(taskType, true, wxPush, appPush, mailPush, u, &list)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return isSaveSuccess, wxStatus, appStatus, mailStatus
|
|
|
+}
|
|
|
+func (p *pushJob) DoPush(taskType int, isSave bool, wxPush, appPush, mailPush int, k *UserInfo, sl *SortList) (isSaveSuccess bool, wxStatus, appStatus, mailStatus int) {
|
|
|
+ defer util.Catch()
|
|
|
+ mailContent := ""
|
|
|
+ jpushtitle := ""
|
|
|
+ lastInfoDate := int64(0)
|
|
|
+ TitleArray := []string{}
|
|
|
+ infos := []*MatchInfo{}
|
|
|
+ infosLength := 0
|
|
|
+ publishTitle := map[string]bool{}
|
|
|
+ dateymd := util.NowFormat(util.Date_yyyyMMdd)
|
|
|
+ dayCountKey := DayCountKey(dateymd, k.Id)
|
|
|
+ onceCountKey := OnceCountKey(dateymd, k.Id)
|
|
|
+ dayCount := redis.GetInt("pushcache_2_a", dayCountKey)
|
|
|
+ isVipUser := IsVipUser(k.VipStatus)
|
|
|
+ now := time.Now()
|
|
|
+ for _, ks := range *sl {
|
|
|
+ k2 := *ks.Info
|
|
|
+ title := strings.Replace(k2["title"].(string), "\n", "", -1)
|
|
|
+ title = Re.ReplaceAllString(title, "$1")
|
|
|
+ area := util.ObjToString(k2["area"])
|
|
|
+ if area == "A" {
|
|
|
+ area = "全国"
|
|
|
+ }
|
|
|
+ newTitle := fmt.Sprintf("[%s]%s", area, title)
|
|
|
+ if publishTitle[newTitle] {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ publishTitle[newTitle] = true
|
|
|
+ infosLength++
|
|
|
+ infos = append(infos, ks)
|
|
|
+ TitleArray = append(TitleArray, newTitle)
|
|
|
+ if infosLength == 1 {
|
|
|
+ jpushtitle = title
|
|
|
+ lastInfoDate = util.Int64All(k2["publishtime"])
|
|
|
+ }
|
|
|
+ //增加行业的处理
|
|
|
+ industry := ""
|
|
|
+ industryclass := "industry"
|
|
|
+ if k2["s_subscopeclass"] != nil {
|
|
|
+ k2sub := strings.Split(util.ObjToString(k2["s_subscopeclass"]), ",")
|
|
|
+ if len(k2sub) > 0 {
|
|
|
+ industry = k2sub[0]
|
|
|
+ if industry != "" {
|
|
|
+ ss := strings.Split(industry, "_")
|
|
|
+ if len(ss) > 1 {
|
|
|
+ industry = ss[0]
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if mailPush == 1 { //关于邮件的处理
|
|
|
+ mailSid := util.CommonEncodeArticle("mailprivate", util.ObjToString(k2["_id"]))
|
|
|
+ url := fmt.Sprintf("%s/article/mailprivate/%s.html", Config.JianyuDomain, mailSid)
|
|
|
+ classArea := "area"
|
|
|
+ classType := "type"
|
|
|
+ infotype := util.ObjToString(k2["subtype"])
|
|
|
+ if infotype == "" {
|
|
|
+ infotype = util.ObjToString(k2["toptype"])
|
|
|
+ }
|
|
|
+ if infotype == "" {
|
|
|
+ infotype = util.ObjToString(k2["type"])
|
|
|
+ if infotype == "tender" {
|
|
|
+ infotype = "招标"
|
|
|
+ } else if infotype == "bid" {
|
|
|
+ infotype = "中标"
|
|
|
+ }
|
|
|
+ }
|
|
|
+ dates := util.LongToDate(k2["publishtime"], false)
|
|
|
+ //标题替换
|
|
|
+ otitle := title
|
|
|
+ for _, kw := range k.Keys {
|
|
|
+ kws := strings.Split(kw, "+")
|
|
|
+ n := 0
|
|
|
+ otitle2 := otitle
|
|
|
+ for _, kwn := range kws {
|
|
|
+ ot := strings.Replace(otitle2, kwn, "<span class='keys'>"+kwn+"</span>", 1)
|
|
|
+ if ot != otitle {
|
|
|
+ n++
|
|
|
+ otitle2 = ot
|
|
|
+ } else {
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if n == len(kws) {
|
|
|
+ otitle = otitle2
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if industry == "" {
|
|
|
+ industryclass = ""
|
|
|
+ }
|
|
|
+ mailContent += fmt.Sprintf(Config.Mail_content, infosLength, url, otitle, classArea, area, classType, infotype, industryclass, industry, dates)
|
|
|
+ }
|
|
|
+ if isVipUser {
|
|
|
+ if dayCount >= Config.VipOneDayMaxPushSize {
|
|
|
+ break
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ //限制最大信息条数
|
|
|
+ if infosLength >= Config.MaxPushSize {
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ dayCount++
|
|
|
+ }
|
|
|
+ if infosLength == 0 {
|
|
|
+ logger.Info("推送任务", taskType, "没有要推送的数据！", k.Id)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ redis.Put("pushcache_2_a", dayCountKey, dayCount, 86400)
|
|
|
+ redis.Put("pushcache_2_a", onceCountKey, infosLength, 86400)
|
|
|
+ //限制一分钟最大的推送数量
|
|
|
+ if p.fastigiumMinutePushPool != nil {
|
|
|
+ if hour := now.Hour(); hour >= FastigiumStart && hour <= FastigiumEnd {
|
|
|
+ <-p.fastigiumMinutePushPool //高峰期
|
|
|
+ }
|
|
|
+ } else if p.minutePushPool != nil {
|
|
|
+ <-p.minutePushPool //正常期
|
|
|
+ }
|
|
|
+ pushDate := ""
|
|
|
+ if taskType != 0 && isSave {
|
|
|
+ //推送记录id
|
|
|
+ pushDate = putil.SaveSendInfo(k, infos)
|
|
|
+ if pushDate == "" {
|
|
|
+ logger.Info("推送任务", taskType, "保存出错", k.Id)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ logger.Info("推送任务", taskType, "保存成功", pushDate, k.Id)
|
|
|
+ isSaveSuccess = true
|
|
|
+ }
|
|
|
+ if isVipUser && (now.Day() == 28 || now.Weekday().String() == "Friday") && (k.RateMode == 3 || k.RateMode == 4) {
|
|
|
+
|
|
|
+ }
|
|
|
+ logger.Info("推送任务", taskType, "开始进行终端推送", k.Id)
|
|
|
+ if isSaveSuccess {
|
|
|
+ //pc端助手推送
|
|
|
+ if k.S_m_openid != "" {
|
|
|
+ logger.Info("推送任务", taskType, "开始助手推送", k.Id, "s_m_openid", k.S_m_openid)
|
|
|
+ isPushOk := putil.SendPcHelper(map[string]interface{}{"clientCode": k.S_m_openid})
|
|
|
+ logger.Info("推送任务", taskType, "助手推送结束", isPushOk, k.Id, "s_m_openid", k.S_m_openid)
|
|
|
+ }
|
|
|
+ if k.Phone != "" {
|
|
|
+ logger.Info("推送任务", taskType, "开始助手推送", k.Id, "phone", k.Phone)
|
|
|
+ isPushOk := putil.SendPcHelper(map[string]interface{}{"clientCode": k.Phone})
|
|
|
+ logger.Info("推送任务", taskType, "助手推送结束", isPushOk, k.Id, "phone", k.Phone)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if wxPush == 1 {
|
|
|
+ logger.Info("推送任务", taskType, "开始微信推送", k.ApplyStatus, k.Id)
|
|
|
+ isPushOk := true
|
|
|
+ if k.ApplyStatus == 1 {
|
|
|
+ TmpTip := ""
|
|
|
+ minute := now.Unix() - lastInfoDate
|
|
|
+ if minute > -1 && minute < 61 {
|
|
|
+ TmpTip = fmt.Sprintf("%d秒前发布的", minute)
|
|
|
+ } else {
|
|
|
+ minute = minute / 60
|
|
|
+ if minute < 121 {
|
|
|
+ if minute < 1 {
|
|
|
+ minute = 1
|
|
|
+ }
|
|
|
+ TmpTip = fmt.Sprintf("%d分钟前发布的", minute)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ Tip1 := util.If(TmpTip == "", "", TmpTip+":\n").(string)
|
|
|
+ LastTip := ""
|
|
|
+ if infosLength > 1 {
|
|
|
+ LastTip = fmt.Sprintf("...(共%d条)", infosLength)
|
|
|
+ }
|
|
|
+ LastTipLen := len([]rune(LastTip))
|
|
|
+ wxTitleKeys := strings.Join(k.Keys, ";")
|
|
|
+ if len([]rune(wxTitleKeys)) > 8 {
|
|
|
+ wxTitleKeys = string([]rune(wxTitleKeys)[:8]) + "..."
|
|
|
+ }
|
|
|
+ wxtitle := fmt.Sprintf(Config.WxTitle, wxTitleKeys)
|
|
|
+ TitleLen := len([]rune(wxtitle))
|
|
|
+ reLen := 200 - TitleLen - 10 - WxGroupLen - len([]rune(Tip1))
|
|
|
+ WXTitle := ""
|
|
|
+ bshow := false
|
|
|
+ for n := 1; n < len(TitleArray)+1; n++ {
|
|
|
+ curTitle := TitleArray[n-1]
|
|
|
+ tmptitle := WXTitle + fmt.Sprintf("%d %s\n", n, curTitle)
|
|
|
+ ch := reLen - len([]rune(tmptitle))
|
|
|
+ if ch < LastTipLen { //加上后大于后辍，则没有完全显示
|
|
|
+ if ch == 0 && n == len(TitleArray) {
|
|
|
+ WXTitle = tmptitle
|
|
|
+ bshow = true
|
|
|
+ } else {
|
|
|
+ ch_1 := reLen - len([]rune(WXTitle)) - LastTipLen
|
|
|
+ if ch_1 > 8 {
|
|
|
+ curLen := len([]rune(curTitle))
|
|
|
+ if ch_1 > curLen {
|
|
|
+ ch_1 = curLen
|
|
|
+ }
|
|
|
+ WXTitle += fmt.Sprintf("%d %s\n", n, string([]rune(curTitle)[:ch_1-3]))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } else if ch == LastTipLen {
|
|
|
+ WXTitle = tmptitle
|
|
|
+ if n == len(TitleArray) {
|
|
|
+ bshow = true
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ WXTitle = tmptitle
|
|
|
+ if n == len(TitleArray) {
|
|
|
+ bshow = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if bshow {
|
|
|
+ LastTip = ""
|
|
|
+ }
|
|
|
+ //推送微信
|
|
|
+ isPushOk = putil.SendWeixin(k, Tip1+WXTitle+LastTip, wxtitle, pushDate)
|
|
|
+ if isPushOk {
|
|
|
+ wxStatus = 1
|
|
|
+ } else {
|
|
|
+ wxStatus = -1
|
|
|
+ }
|
|
|
+ }
|
|
|
+ logger.Info("推送任务", taskType, "微信推送结束", k.ApplyStatus, isPushOk, k.Id)
|
|
|
+ }
|
|
|
+ if appPush == 1 {
|
|
|
+ logger.Info("推送任务", taskType, "开始app推送", k.Id)
|
|
|
+ descriptAppend := ""
|
|
|
+ if infosLength > 1 {
|
|
|
+ descriptAppend = fmt.Sprintf("\n...(共%d条)", infosLength)
|
|
|
+ jpushtitle = fmt.Sprintf("1. %s", jpushtitle)
|
|
|
+ }
|
|
|
+ go mongodb.Update("user", map[string]interface{}{
|
|
|
+ "_id": bson.ObjectIdHex(k.Id),
|
|
|
+ }, map[string]interface{}{
|
|
|
+ "$inc": map[string]interface{}{
|
|
|
+ "i_apppushunread": 1,
|
|
|
+ },
|
|
|
+ }, false, false)
|
|
|
+ isPushOk := putil.SendApp(map[string]interface{}{
|
|
|
+ "phoneType": k.AppPhoneType,
|
|
|
+ "descript": jpushtitle,
|
|
|
+ "descriptAppend": descriptAppend,
|
|
|
+ "type": "bid",
|
|
|
+ "userId": k.Id,
|
|
|
+ "url": "/jyapp/free/sess/" + Se.EncodeString(k.Id+",_id,"+strconv.Itoa(int(now.Unix()))+",historypush"),
|
|
|
+ "otherPushId": k.Opushid,
|
|
|
+ "jgPushId": k.Jpushid, //极光-推送id
|
|
|
+ })
|
|
|
+ if isPushOk {
|
|
|
+ appStatus = 1
|
|
|
+ } else {
|
|
|
+ appStatus = -1
|
|
|
+ }
|
|
|
+ logger.Info("推送任务", taskType, "app推送结束", isPushOk, k.Id)
|
|
|
+ }
|
|
|
+ //发送邮件
|
|
|
+ if mailPush == 1 {
|
|
|
+ logger.Info("推送任务", taskType, "开始邮箱推送", k.Id)
|
|
|
+ html := fmt.Sprintf(Config.Mail_html, strings.Replace(strings.Join(k.Keys, ";"), "+", " ", -1), mailContent)
|
|
|
+ isPushOk := p.SendMail(k.Email, Config.Mail_title, html, nil)
|
|
|
+ if isPushOk {
|
|
|
+ mailStatus = 1
|
|
|
+ } else {
|
|
|
+ mailStatus = -1
|
|
|
+ }
|
|
|
+ logger.Info("推送任务", taskType, "邮箱推送结束", isPushOk, k.Id)
|
|
|
+ }
|
|
|
+ return
|
|
|
+}
|
|
|
+
|
|
|
+//推送邮件(含附件)
|
|
|
+func (p *pushJob) SendMail(email, subject, html string, fmdatas []map[string]interface{}) bool {
|
|
|
+ if !Config.IsPushMail || len(Gmails) == 0 {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ if Config.MailSleep > 0 {
|
|
|
+ time.Sleep(time.Duration(Config.MailSleep) * time.Millisecond)
|
|
|
+ }
|
|
|
+ defer util.Catch()
|
|
|
+ //生成附件
|
|
|
+ var fnamepath, rename string
|
|
|
+ // if len(fmdatas) > 0 { //开启导出
|
|
|
+ // fnamepath, rename = putil.GetBidInfoXlsx(fmdatas)
|
|
|
+ // }
|
|
|
+ status := false
|
|
|
+ index := len(email) % 2
|
|
|
+ if index >= len(Gmails) {
|
|
|
+ index = 0
|
|
|
+ }
|
|
|
+ gmail := Gmails[index]
|
|
|
+ for i := 0; i < len(Gmails); i++ {
|
|
|
+ status = mail.GSendMail("剑鱼标讯", email, "", "", subject, html, fnamepath, rename, gmail)
|
|
|
+ if status {
|
|
|
+ break
|
|
|
+ } else {
|
|
|
+ gmail = Gmails[index^1]
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if fnamepath != "" {
|
|
|
+ os.Remove(fnamepath)
|
|
|
+ }
|
|
|
+ return status
|
|
|
+}
|
