wechat-api/crontask/send_msg.go

package crontask

import (
	"strings"
	"time"
	"wechat-api/ent"
	"wechat-api/ent/batchmsg"
	"wechat-api/ent/contact"
	"wechat-api/ent/label"
	"wechat-api/ent/labelrelationship"
	"wechat-api/ent/msg"
	"wechat-api/ent/wx"
	"wechat-api/hook"
)

func (l *CronTask) sendMsg() {
	// 获取 BatchMsg 表中 start_time 小于当前时间并且 status 为 0 或 1 的数据
	batchlist, err := l.svcCtx.DB.BatchMsg.Query().Where(batchmsg.StartTimeLT(time.Now()), batchmsg.StatusIn(0, 1)).All(l.ctx)
	if err != nil {
		l.Logger.Errorf("batchlist err: %v", err)
		return
	}

	for _, batch := range batchlist {
		// 记录当前批次开始处理
		l.Logger.Info("batch start: ", batch.BatchNo)
		// 如果 批次 status 为 0，则先产生待发送消息
		if batch.Status == 0 {
			userlist := make([]*ent.Contact, 0)
			if batch.Tag == "all" {
				// 获取 contact 表中  wx_wxid 等于 req.Fromwxid 的 type 为1或2的数据
				userlist, err = l.svcCtx.DB.Contact.Query().Where(contact.WxWxid(batch.Fromwxid), contact.TypeIn(1, 2)).All(l.ctx)
				if err != nil {
					l.Logger.Errorf("userlist err: %v", err)
					continue
				}
			} else {

				tags := strings.Split(batch.Tag, ",")

				// 获取 label 表中 name 为 tags的记录
				labids, err := l.svcCtx.DB.Label.Query().Where(label.NameIn(tags...)).IDs(l.ctx)
				if err != nil {
					l.Logger.Errorf("labids err: %v", err)
					continue
				}
				// 获取 label_relationship 表中，label_id 等于 labids 的 contact_id
				labelrelationships, err := l.svcCtx.DB.LabelRelationship.Query().Where(labelrelationship.LabelIDIn(labids...)).All(l.ctx)
				if err != nil {
					l.Logger.Errorf("labelrelationships err: %v", err)
					continue
				}
				contact_ids := make([]uint64, 0)
				for _, labelrelationship := range labelrelationships {
					contact_ids = append(contact_ids, labelrelationship.ContactID)
				}
				if len(contact_ids) > 0 {
					// 获取 contact 表中 wx_wxid 等于 req.Fromwxid 并且 id 等于 contact_ids 并且 type 为1或2 的数据
					userlist, err = l.svcCtx.DB.Contact.Query().Where(contact.WxWxid(batch.Fromwxid), contact.IDIn(contact_ids...), contact.TypeIn(1, 2)).All(l.ctx)
					if err != nil {
						l.Logger.Errorf("userlist err: %v", err)
						continue
					}
				}
			}

			msgs := make([]*ent.MsgCreate, 0)

			for _, user := range userlist {

				msg := l.svcCtx.DB.Msg.Create().
					SetNotNilFromwxid(&batch.Fromwxid).
					SetNotNilToid(&user.Wxid).
					SetMsgtype(1).
					SetNotNilMsg(&batch.Msg).
					SetStatus(0).
					SetNotNilBatchNo(&batch.BatchNo)

				msgs = append(msgs, msg)
			}

			if len(msgs) > 0 {
				_, err = l.svcCtx.DB.Msg.CreateBulk(msgs...).Save(l.ctx)
				if err != nil {
					l.Logger.Errorf("msg CreateBulk err: %v", err)
					continue
				}
			} else {
				// 如果没有消息，直接更新批次状态为已发送
				_, err = l.svcCtx.DB.BatchMsg.UpdateOneID(batch.ID).
					SetStatus(2).
					SetTotal(0).
					SetSuccess(0).
					SetFail(0).
					Save(l.ctx)
				if err != nil {
					l.Logger.Errorf("batchmsg update err: %v", err)
				}
				continue
			}

			_, err = l.svcCtx.DB.BatchMsg.UpdateOneID(batch.ID).Where(batchmsg.StatusNEQ(1)).SetStatus(1).Save(l.ctx)
			if err != nil {
				l.Logger.Errorf("batchmsg update err: %v", err)
				continue
			}
		}

		// 获取当前批次的所有待发送消息
		msglist, err := l.svcCtx.DB.Msg.Query().Where(msg.BatchNoEQ(batch.BatchNo), msg.StatusEQ(0)).All(l.ctx)
		if err != nil {
			l.Logger.Errorf("msglist err: %v", err)
			continue
		}

		wxInfo, err := l.svcCtx.DB.Wx.Query().Where(wx.Wxid(batch.Fromwxid)).Only(l.ctx)
		if err != nil {
			l.Logger.Errorf("wxInfo err: %v", err)
			continue
		}
		serverInfo, err := l.svcCtx.DB.Server.Get(l.ctx, wxInfo.ServerID)
		if err != nil {
			l.Logger.Errorf("serverInfo err: %v", err)
			continue
		}

		hookClient := hook.NewHook(serverInfo.PrivateIP, serverInfo.AdminPort, wxInfo.Port)

		//循环发送消息
		for _, msg := range msglist {
			err = hookClient.SendTextMsg(msg.Toid, msg.Msg)
			// 每次发完暂停1秒
			time.Sleep(time.Second)
			if err != nil {
				l.Logger.Errorf("send msg err: %v", err)
				_, err = l.svcCtx.DB.Msg.UpdateOneID(msg.ID).SetStatus(2).Save(l.ctx)
				if err != nil {
					l.Logger.Errorf("msg update err: %v", err)
					continue
				}
				continue
			}

			_, err = l.svcCtx.DB.Msg.UpdateOneID(msg.ID).SetStatus(1).Save(l.ctx)
			if err != nil {
				l.Logger.Errorf("msg update err: %v", err)
				continue
			}
		}

		// 获取当前批次的所有发送的消息总数
		total, _ := l.svcCtx.DB.Msg.Query().Where(msg.BatchNoEQ(batch.BatchNo)).Count(l.ctx)

		// 获取当前批次的所有发送成功的消息总数
		success, _ := l.svcCtx.DB.Msg.Query().Where(msg.BatchNoEQ(batch.BatchNo), msg.StatusEQ(1)).Count(l.ctx)

		// 获取当前批次的所有发送失败的消息总数
		fail, _ := l.svcCtx.DB.Msg.Query().Where(msg.BatchNoEQ(batch.BatchNo), msg.StatusEQ(2)).Count(l.ctx)

		// 更新批次状态为已发送，同时更新发送总数、发送成功数量、失败数量、结束时间
		_, err = l.svcCtx.DB.BatchMsg.UpdateOneID(batch.ID).
			SetStatus(2).
			SetTotal(int32(total)).
			SetSuccess(int32(success)).
			SetFail(int32(fail)).
			SetStopTime(time.Now()).
			Save(l.ctx)
		if err != nil {
			l.Logger.Errorf("batchmsg update err: %v", err)
			continue
		}
		l.Logger.Info("batch stop: ", batch.BatchNo)

	}

}