wechat-api/crontask/send_wx.go

package crontask

import (
	"context"
	"encoding/json"
	"regexp"
	"strings"
	"time"
	"wechat-api/ent"
	"wechat-api/ent/contact"
	"wechat-api/ent/messagerecords"
	"wechat-api/ent/server"
	"wechat-api/ent/wx"
	"wechat-api/hook"
	"wechat-api/internal/types/payload"
)

func (l *CronTask) sendWx() {
	ctx := context.Background()
	config :=
		`{
    "ip": "172.18.41.219",
    "number": 10,
    "wx_list": [
        {
            "port": "30001",
            "wxid": "wxid_77au928zeb2p12",
            "nickname": "百事通"
        },
        {
            "port": "30007",
            "wxid": "wxid_edc0mvp188ms22",
            "nickname": "爱博闻"
        },
        {
            "port": "30040",
            "wxid": "wxid_wupstwe851nb12",
            "nickname": "轻马小助手"
        }
    ]
}`

	var p payload.SendWxPayload
	if err := json.Unmarshal([]byte(config), &p); err != nil {
		l.Logger.Errorf("failed to unmarshal the payload :%v", err)
		return
	}

	// 更新最旧的 p.Number 条 status 值为 1 或 4 的 message_records 记录的 status 为 2
	messages, err := l.svcCtx.DB.MessageRecords.Query().
		Where(messagerecords.StatusIn(1, 4)).
		Order(ent.Asc(messagerecords.FieldCreatedAt)).
		Limit(p.Number).
		All(ctx)
	if err != nil {
		l.Logger.Errorf("get messageRecords list failed %v", err)
		return
	}

	startTime := time.Now()

	type SendTextMsgReq struct {
		Wxid string `json:"wxid"`
		Msg  string `json:"msg"`
	}

	type SendFileMsgReq struct {
		Wxid        string `json:"wxid"`
		Filepath    string `json:"filepath"`
		Diyfilename string `json:"diyfilename"`
	}

	// 从 Redis 中获取服务器信息(ip, port)
	getServerInfo := func(wxid string) (string, error) {
		key := "crontask_wx_server_info"
		val, _ := l.svcCtx.Rds.HGet(ctx, key, wxid).Result()
		//if err != nil {
		//	return "", err
		//}

		if val == "" {
			wx, err := l.svcCtx.DB.Wx.Query().Where(wx.WxidEQ(wxid)).First(l.ctx)
			if err != nil {
				l.Logger.Errorf("get wx info failed wxid=%v err=%v", wxid, err)
				return "", err
			}
			if wx.ServerID != 0 {
				server, err := l.svcCtx.DB.Server.Query().Where(server.IDEQ(wx.ServerID)).First(l.ctx)
				if err != nil {
					l.Logger.Errorf("get server info failed wxid=%v err=%v", wxid, err)
					return "", err
				}
				val = server.PrivateIP + ":" + server.AdminPort + ":" + wx.Port
				l.svcCtx.Rds.HSet(ctx, key, wxid, val)
			}
		}

		return val, nil
	}

	// 从 Redis 中获取服务器信息(ip, port)
	getContactInfo := func(botwxid string, wxid string) (*ent.Contact, error) {
		contactInfo, _ := l.svcCtx.DB.Contact.Query().Where(contact.WxWxidEQ(botwxid), contact.WxidEQ(wxid)).First(l.ctx)

		return contactInfo, nil
	}

	for _, v := range messages {
		// 更新 status 值为 2（发送中）
		tx, _ := l.svcCtx.DB.Tx(l.ctx)
		if v.Content != "" {
			serverInfo, _ := getServerInfo(v.BotWxid)
			serverIp := ""
			adminPort := ""
			wxPort := ""
			if serverInfo != "" {
				infoArray := strings.Split(serverInfo, ":")
				serverIp, adminPort, wxPort = infoArray[0], infoArray[1], infoArray[2]
			}

			_, err = tx.MessageRecords.UpdateOneID(v.ID).SetStatus(2).Save(ctx)
			if err != nil {
				l.Logger.Errorf("update messageRecords failed id=%v err=%v", v.ID, err)
				continue
			}

			hookClient := hook.NewHook(serverIp, adminPort, wxPort)
			if v.ContentType == 1 {
				content := v.Content
				if containsPlaceholder(content) {
					contactInfo, _ := getContactInfo(v.BotWxid, v.ContactWxid)
					content = varReplace(content, contactInfo)
				}
				err = hookClient.SendTextMsg(v.ContactWxid, content, v.BotWxid)
			} else {
				re := regexp.MustCompile(`[^/]+$`)
				fileName := re.FindString(v.Content)
				err = hookClient.SendPicMsg(v.ContactWxid, v.Content, fileName, v.BotWxid)
			}

			if err != nil {
				_ = tx.Rollback()
				continue
			} else {
				_, err := tx.MessageRecords.UpdateOneID(v.ID).SetStatus(3).SetSendTime(time.Now()).Save(ctx)
				if err != nil {
					_ = tx.Rollback()
				} else {
					_ = tx.Commit()
				}
			}

			time.Sleep(time.Duration(60/p.Number) * time.Second)
		} else {
			_, err := tx.MessageRecords.UpdateOneID(v.ID).SetStatus(3).SetSendTime(time.Now()).Save(ctx)
			if err != nil {
				_ = tx.Rollback()
			} else {
				_ = tx.Commit()
			}
		}
	}

	finishTime := time.Now()
	l.Logger.Infof("This process cost %v", finishTime.Sub(startTime).String())
	return
}

func containsPlaceholder(s string) bool {
	pattern := `\$\{.*?\}`
	matched, _ := regexp.MatchString(pattern, s)
	return matched
}

func varReplace(s string, contactInfo *ent.Contact) string {
	nickname := ""
	if contactInfo != nil {
		nickname = contactInfo.Nickname
	}
	s = strings.Replace(s, "${nickname}", nickname, -1)
	return s
}