package crontask import ( "encoding/json" "net/url" "path" "strings" "time" "wechat-api/ent" "wechat-api/ent/batchmsg" "wechat-api/ent/contact" "wechat-api/ent/custom_types" "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 } } } // 这里是待插入到 msg 表的数据 msgs := make([]*ent.MsgCreate, 0) // 这里是把 batch.Msg 转换为 json 数组 msgArray := make([]custom_types.Action, 0) err := json.Unmarshal([]byte(batch.Msg), &msgArray) l.Logger.Infof("msgArray length= %v, err:%v", len(msgArray), err) if err != nil { // json 解析失败 msgArray = make([]custom_types.Action, 0) } for _, user := range userlist { // 这里改动主要是 batch_msg 目前支持批量添加图文,导致 batch_msg 的 msg 字段为 json // msg 里包括文字和图片,msgtype=1 为文字, msgtype=2 为图片 // 每一条文字或者图片 都是一条单独的消息 if len(msgArray) > 0 { // 这里是新格式(msg内容为json),需要遍历数组 for _, msgItem := range msgArray { msgRow := l.svcCtx.DB.Msg.Create(). SetNotNilFromwxid(&batch.Fromwxid). SetNotNilToid(&user.Wxid). SetMsgtype(int32(msgItem.Type)). SetNotNilMsg(&msgItem.Content). SetStatus(0). SetNotNilBatchNo(&batch.BatchNo) msgs = append(msgs, msgRow) } } } if len(msgs) > 0 { // 加事务,批量操作一条 batch_msg 和 一堆 msg 信息 tx, err := l.svcCtx.DB.Tx(l.ctx) if err != nil { l.Logger.Errorf("start db transaction err: %v", err) continue } _, err = tx.BatchMsg.UpdateOneID(batch.ID).Where(batchmsg.StatusNEQ(1)).SetStatus(1).Save(l.ctx) if err != nil { _ = tx.Rollback() l.Logger.Errorf("batchmsg update err: %v", err) continue } _, err = tx.Msg.CreateBulk(msgs...).Save(l.ctx) if err != nil { _ = tx.Rollback() l.Logger.Errorf("msg CreateBulk err: %v", err) continue } _ = tx.Commit() } 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 } } // 获取当前批次的所有待发送消息 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 { // 这里之前只有文字消息(既 msgtype=1) 目前增加了图片 所以增加了msgtype=2 // 所以增加了一个判断,判断发送的内容类型,如果是文字就调用SendTextMsg,如果是图片就调用SendPicMsg if msg.Msgtype == 1 { err = hookClient.SendTextMsg(msg.Toid, msg.Msg) } else if msg.Msgtype == 2 { diyfilename := getFileName(msg.Msg) err = hookClient.SendPicMsg(msg.Toid, msg.Msg, diyfilename) } // 每次发完暂停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) } } // 根据URL获取图片名 func getFileName(photoUrl string) string { u, err := url.Parse(photoUrl) if err != nil { return "" } return path.Base(u.Path) }