im wechat (#330)

* im wechat

* im wechat

Co-authored-by: alickliming <alickliming@didi.global.com>

etc/rdb.yml

@@ -73,7 +73,12 @@ sender:
     worker: 10
     api: http://127.0.0.1:2008/voice
   im:
-    # two choice: shell|api
+    # three choice: shell|api|wechat
     way: shell
     worker: 10
     api: http://127.0.0.1:2008/im
+
+wechat:
+  corp_id: "xxxxxxxxxxxxx"
+  agent_id: 1000000
+  secret: "xxxxxxxxxxxxxxxxx"

src/modules/rdb/config/yaml.go

@@ -17,6 +17,13 @@ type ConfigT struct {
 	Redis    redisSection             `yaml:"redis"`
 	Sender   map[string]senderSection `yaml:"sender"`
 	RabbitMQ rabbitmqSection          `yaml:"rabbitmq"`
+	WeChat   wechatSection            `yaml:"wechat"`
+}
+
+type wechatSection struct {
+	CorpID  string `yaml:"corp_id"`
+	AgentID int    `yaml:"agent_id"`
+	Secret  string `yaml:"secret"`
 }
 
 type ssoSection struct {

src/modules/rdb/corp/corp.go

@@ -0,0 +1,181 @@
+package corp
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"time"
+)
+
+// Err 微信返回错误
+type Err struct {
+	ErrCode int    `json:"errcode"`
+	ErrMsg  string `json:"errmsg"`
+}
+
+// AccessToken 微信企业号请求Token
+type AccessToken struct {
+	AccessToken string `json:"access_token"`
+	ExpiresIn   int    `json:"expires_in"`
+	Err
+	ExpiresInTime time.Time
+}
+
+// Client 微信企业号应用配置信息
+type Client struct {
+	CorpID      string
+	AgentID     int
+	AgentSecret string
+	Token       AccessToken
+}
+
+// Result 发送消息返回结果
+type Result struct {
+	Err
+	InvalidUser  string `json:"invaliduser"`
+	InvalidParty string `json:"infvalidparty"`
+	InvalidTag   string `json:"invalidtag"`
+}
+
+// Content 文本消息内容
+type Content struct {
+	Content string `json:"content"`
+}
+
+// Message 消息主体参数
+type Message struct {
+	ToUser  string  `json:"touser"`
+	ToParty string  `json:"toparty"`
+	ToTag   string  `json:"totag"`
+	MsgType string  `json:"msgtype"`
+	AgentID int     `json:"agentid"`
+	Text    Content `json:"text"`
+}
+
+// New 实例化微信企业号应用
+func New(corpID string, agentID int, AgentSecret string) *Client {
+	c := new(Client)
+	c.CorpID = corpID
+	c.AgentID = agentID
+	c.AgentSecret = AgentSecret
+	return c
+}
+
+// Send 发送信息
+func (c *Client) Send(msg Message) error {
+	if err := c.GetAccessToken(); err != nil {
+		return err
+	}
+
+	msg.AgentID = c.AgentID
+	url := "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + c.Token.AccessToken
+
+	resultByte, err := jsonPost(url, msg)
+	if err != nil {
+		return fmt.Errorf("invoke send api fail: %v", err)
+	}
+
+	result := Result{}
+	err = json.Unmarshal(resultByte, &result)
+	if err != nil {
+		return fmt.Errorf("parse send api response fail: %v", err)
+	}
+
+	if result.ErrCode != 0 {
+		err = fmt.Errorf("invoke send api return ErrCode = %d", result.ErrCode)
+	}
+
+	if result.InvalidUser != "" || result.InvalidParty != "" || result.InvalidTag != "" {
+		err = fmt.Errorf("invoke send api partial fail, invalid user: %s, invalid party: %s, invalid tag: %s", result.InvalidUser, result.InvalidParty, result.InvalidTag)
+	}
+
+	return err
+}
+
+// GetAccessToken 获取会话token
+func (c *Client) GetAccessToken() error {
+	var err error
+
+	if c.Token.AccessToken == "" || c.Token.ExpiresInTime.Before(time.Now()) {
+		c.Token, err = getAccessTokenFromWeixin(c.CorpID, c.AgentSecret)
+		if err != nil {
+			return fmt.Errorf("invoke getAccessTokenFromWeixin fail: %v", err)
+		}
+		c.Token.ExpiresInTime = time.Now().Add(time.Duration(c.Token.ExpiresIn-1000) * time.Second)
+	}
+
+	return err
+}
+
+// transport 全局复用，提升性能
+var transport = &http.Transport{
+	TLSClientConfig:    &tls.Config{InsecureSkipVerify: true},
+	DisableCompression: true,
+}
+
+// getAccessTokenFromWeixin 从微信服务器获取token
+func getAccessTokenFromWeixin(corpID, secret string) (accessToken AccessToken, err error) {
+	url := "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=" + corpID + "&corpsecret=" + secret
+
+	client := &http.Client{Transport: transport}
+	result, err := client.Get(url)
+	if err != nil {
+		return accessToken, fmt.Errorf("invoke api gettoken fail: %v", err)
+	}
+
+	if result.Body == nil {
+		return accessToken, fmt.Errorf("gettoken response body is nil")
+	}
+
+	defer result.Body.Close()
+
+	res, err := ioutil.ReadAll(result.Body)
+	if err != nil {
+		return accessToken, fmt.Errorf("read gettoken response body fail: %v", err)
+	}
+
+	err = json.Unmarshal(res, &accessToken)
+	if err != nil {
+		return accessToken, fmt.Errorf("parse gettoken response body fail: %v", err)
+	}
+
+	if accessToken.ExpiresIn == 0 || accessToken.AccessToken == "" {
+		err = fmt.Errorf("invoke api gettoken fail, ErrCode: %v, ErrMsg: %v", accessToken.ErrCode, accessToken.ErrMsg)
+		return accessToken, err
+	}
+
+	return accessToken, err
+}
+
+func jsonPost(url string, data interface{}) ([]byte, error) {
+	jsonBody, err := encodeJSON(data)
+	if err != nil {
+		return nil, err
+	}
+
+	r, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(jsonBody))
+	if err != nil {
+		return nil, err
+	}
+
+	if r.Body == nil {
+		return nil, fmt.Errorf("response body of %s is nil", url)
+	}
+
+	defer r.Body.Close()
+
+	return ioutil.ReadAll(r.Body)
+}
+
+func encodeJSON(v interface{}) ([]byte, error) {
+	var buf bytes.Buffer
+	encoder := json.NewEncoder(&buf)
+	encoder.SetEscapeHTML(false)
+	if err := encoder.Encode(v); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}

src/modules/rdb/cron/sender_im.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/didi/nightingale/src/common/dataobj"
 	"github.com/didi/nightingale/src/modules/rdb/config"
+	"github.com/didi/nightingale/src/modules/rdb/corp"
 	"github.com/didi/nightingale/src/modules/rdb/redisc"
 )
 
@@ -47,6 +48,8 @@ func sendIm(message *dataobj.Message) {
 		sendImByAPI(message)
 	case "shell":
 		sendImByShell(message)
+	case "wechat":
+		sendImByWeChat(message)
 	default:
 		logger.Errorf("not support %s to send im, im: %+v", config.Config.Sender["im"].Way, message)
 	}
@@ -68,3 +71,31 @@ func sendImByShell(message *dataobj.Message) {
 	output, err, isTimeout := sys.CmdRunT(time.Second*10, shell, strings.Join(message.Tos, ","), message.Content)
 	logger.Infof("SendImByShell, im:%+v, output:%s, error: %v, isTimeout: %v", message, output, err, isTimeout)
 }
+
+func sendImByWeChat(message *dataobj.Message) {
+	corpID := config.Config.WeChat.CorpID
+	agentID := config.Config.WeChat.AgentID
+	secret := config.Config.WeChat.Secret
+
+	cnt := len(message.Tos)
+	if cnt == 0 {
+		logger.Warningf("im send wechat fail, empty tos, message: %+v", message)
+		return
+	}
+
+	client := corp.New(corpID, agentID, secret)
+	var err error
+	for i := 0; i < cnt; i++ {
+		err = client.Send(corp.Message{
+			ToUser:  message.Tos[i],
+			MsgType: "text",
+			Text:    corp.Content{Content: message.Content},
+		})
+
+		if err != nil {
+			logger.Warningf("im wechat send to %s fail: %v", message.Tos[i], err)
+		} else {
+			logger.Infof("im wechat send to %s succ", message.Tos[i])
+		}
+	}
+}