企业微信自建应用来推送消息

前言

最近有个给特定部门推送消息的需求，所以配置一个应用专门用来推送消息。实现过程大致为：服务器生成每天的报告，通过调用API来发送消息。以前一直都是发邮件，整个邮箱里全是报告文件，看着很麻烦，所以想找个好点的方案给代替了。正好企业微信有发消息的API，配置步骤也正好做个记录。

配置需要：

1. 一台公网可访问的服务器，用来和企业微信通信验证（仅做验证用，用后即可删除）
2. 一个公司的域名，如公司官网为www.abc.com，则需要xxx.abc.com（仅做验证用，用后即可删除）
3. 一个固定公网IP，若无固定IP则可以使用DDNS获取一个域名，无公网IP的可以使用内网穿透（需长期使用，作为可信IP推送消息）

配置原理：

后台创建应用后还无法发送消息，需要配置一个或多个 可信IP ，配置好后就仅 可信IP 才能通过接口获取企业数据。而配置 可信IP 前需要先配置 可信域名 或 接收消息服务器URL ，本文只记录公司有固定公网IP如何配置 接收消息服务器URL 。配置好 可信IP 后就可以通过接口实现消息推送了。

若需要DDNS配置或内网穿透的配置可以评论或私信，我看到有人需要的话会再出一篇操作笔记。

操作步骤

1. 企业微信管理员登录后台，创建一个新应用

2. 进入新应用后配置可见范围

记录AgentId和Secret，后续会用到。

3. 配置企业可信IP

点击设置接收消息服务器URL，跳转到配置页面

4. 配置接收消息服务器

随机获取Token和Key后不要关闭这个页面，不要填URL一栏。需要先配置消息接收服务器进行验证后才能获取到URL

查看企业ID，后续会用到

查看用户ID，后续会用到，账号即userID

5. 公网服务器上配置验证脚本

拉取文件并解压

# 创建脚本目录
mkdir /wechat
cd /wechat

# github下载项目
wget https://github.com/sbzhu/weworkapi_python/archive/refs/heads/master.zip
 
# 解压
apt install unzip
unzip master.zip
 
# 改名
mv weworkapi_python-master weworkapi_python

需要先自行安装python3，不会可百度

创建python脚本，此脚本需放在脚本目录中，脚本中需要替换的地方按自己的数据进行替换。

vim /wechat/a.py

# -*- encoding:utf-8 -*-
from flask import abort, request
from flask import Flask
from xml.dom.minidom import parseString
import _thread
import time
import os
import sys

sys.path.append("/wechat/weworkapi_python/callback")  # 正确的模块导入路径
from WXBizMsgCrypt3 import WXBizMsgCrypt  # https://github.com/sbzhu/weworkapi_python 项目地址

app = Flask(__name__)


# 对应接受消息回调模式中的URL，如域名是'www.example.com' 那么在步骤4中填入的url就为"http://www.example.com/hook_path"
@app.route('/hook_path', methods=['GET', 'POST'])
def douban():
    if request.method == 'GET':
        echo_str = signature(request, 0)
        return (echo_str)
    elif request.method == 'POST':
        echo_str = signature2(request, 0)
        return (echo_str)


qy_api = [
    WXBizMsgCrypt("XXXToken替换XXX", "XXXKey替换XXX", "XX企业ID替换XX"),
]  # 对应接受消息回调模式中的token，EncodingAESKey 和 企业信息中的企业id


# 开启消息接受模式时验证接口连通性
def signature(request, i):
    msg_signature = request.args.get('msg_signature', '')
    timestamp = request.args.get('timestamp', '')
    nonce = request.args.get('nonce', '')
    echo_str = request.args.get('echostr', '')
    ret, sEchoStr = qy_api[i].VerifyURL(msg_signature, timestamp, nonce, echo_str)
    if (ret != 0):
        print("ERR: VerifyURL ret: " + str(ret))
        return ("failed")
    else:
        return (sEchoStr)


# 实际接受消息
def signature2(request, i):
    msg_signature = request.args.get('msg_signature', '')
    timestamp = request.args.get('timestamp', '')
    nonce = request.args.get('nonce', '')
    data = request.data.decode('utf-8')
    ret, sMsg = qy_api[i].DecryptMsg(data, msg_signature, timestamp, nonce)
    if (ret != 0):
        print("ERR: DecryptMsg ret: " + str(ret))
        return ("failed")
    else:
        with open("/var/log/qywx.log", 'a+') as f:  # 消息接收日志
            doc = parseString(sMsg)
            collection = doc.documentElement
            name_xml = collection.getElementsByTagName("FromUserName")
            msg_xml = collection.getElementsByTagName("Content")
            type_xml = collection.getElementsByTagName("MsgType")
            pic_xml = collection.getElementsByTagName("PicUrl")
            msg = ""
            name = ""
            msg_type = type_xml[0].childNodes[0].data
            if msg_type == "text":  # 文本消息
                name = name_xml[0].childNodes[0].data  # 发送者id
                msg = msg_xml[0].childNodes[0].data  # 发送的消息内容
                f.write(time.strftime('[%Y-%m-%d %H:%M:%S]') + "[ch%d] %s:%s\n" % (i, name, msg))
                _thread.start_new_thread(os.system, ("python3 command.py '%s' '%s' '%d' '%d'" % (name, msg, i, 0),))

            elif msg_type == "image":
                name = name_xml[0].childNodes[0].data
                pic_url = pic_xml[0].childNodes[0].data
                f.write(time.strftime('[%Y-%m-%d %H:%M:%S]') + "[ch%d] %s:图片消息\n" % (i, name))
                _thread.start_new_thread(os.system, ("python3 command.py '%s' '%s' '%d' '%d'" % (name, pic_url, i, 1),))

            f.close()

        return ("ok")


if __name__ == '__main__':
    app.run("0.0.0.0", 808)  # 自己设置端口

启动脚本

# 创建虚拟环境，，，也可以不在虚拟环境中运行，直接python3 a.py运行即可
apt install -y python3-venv
python3 -m venv /wechat/venv
source /wechat/venv/bin/activate

# 缺少模块就安装
pip install pycryptodome -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install flask -i https://pypi.tuna.tsinghua.edu.cn/simple

# 启动脚本
python3 a.py

启动成功

访问网站验证是否完成，如图显示failed即成功

如果无法访问可以看看云服务器的访问规则或防火墙是否放行端口。

6. 将该公网IP解析到域名

以前可以使用IP验证，现在需备案主体与当前企业主体相同或有关联关系的域名才能验证，自行通过域名解析服务商解析即可

例如：

​ 验证服务器的IP：123.123.123.123

​ 解析的域名为：aaa.abc.com

回到API配置页面，填写URL并保存即可

现在就可以配置可信IP了

配置完可信IP就可以调用接口发送信息了，然后验证服务器就没用了，API接收消息的配置也可以删掉。

7. 接口开发

API官方文档：概述 - 企业微信 API

开发者文档：开发前必读 - 文档 - 企业微信开发者中心

我需要消息推送的功能，则接入消息API，以下是我的测试代码

import requests
import json

# 企业微信配置常量
CORPID = "xxxxxxxxxx"   # 企业ID
CORPSECRET = "xxxxxxxxxxxxxxx"  # 应用的Secret，在应用页面查看
AGENTID = 100001    # 应用的AgentId，在应用页面查看
RECEIVER = "userID" # 用户ID，可在通讯录中查看

# 卡片消息变量
TITLE = "xxxx年xx月xx日测试报告"   # 卡片标题
DES_1 = "测试主要内容："   # 卡片文本内容
DES_2 = "测试人：xxx"   # 卡片文本内容
DES_3 = "对比结果：xxxx" # 卡片文本内容
DES_4 = "其他：xxxx"   # 卡片文本内容
URL = "file:///E:/Users/LenYu/Downloads/report.html"    # 文件URL

class MessageSender:
    def __init__(self):
        self.session = requests.Session()

    def send_text_card(self, title, description, url, btn_txt="详情"):
        try:
            token = self.get_access_token()
            if not token:
                print("获取Token失败")
                return False
            message = {
                "touser": RECEIVER,
                "msgtype": "textcard",
                "agentid": AGENTID,
                "textcard": {
                    "title": title[:126] + ".." if len(title) > 128 else title,
                    "description": self._format_description(description),
                    "url": url,
                    "btntxt": btn_txt[:4]
                }
            }

            response = self.session.post(
                "https://qyapi.weixin.qq.com/cgi-bin/message/send",
                params={"access_token": token},
                json=message
            )

            return self._handle_response(response)

        except Exception as e:
            print(f"发送文本卡片失败：{str(e)}")
            return False

    def _format_description(self, desc):
        formatted = desc.replace('\n', '<br>')
        return formatted[:1500] + "..." if len(desc) > 1500 else formatted

    def get_access_token(self):
        try:
            url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
            params = {"corpid": CORPID, "corpsecret": CORPSECRET}

            response = self.session.get(url, params=params)
            if not response.ok:
                print(f"Token请求失败：{response.status_code}")
                return None

            data = response.json()
            if data.get("errcode") == 0:
                return data.get("access_token")

            print(f"获取Token失败：{data.get('errmsg')}")
            return None

        except Exception as e:
            print(f"获取Token异常：{str(e)}")
            return None

    def _handle_response(self, response):
        try:
            if response.status_code != 200:
                print(f"请求失败，状态码：{response.status_code}")
                return False

            result = response.json()
            if result.get('errcode') == 0:
                print("操作成功")
                return True

            print(f"操作失败：[{result.get('errcode')}] {result.get('errmsg')}")
            return False

        except json.JSONDecodeError:
            print("响应解析失败")
            return False

if __name__ == "__main__":
    sender = MessageSender()

    card_info = {
        "title": TITLE,
        "description": f"{DES_1}\n- {DES_2}\n- {DES_3}\n- {DES_4}",
        "url": URL,
        "btn_txt": "查看报告"
    }

    sender.send_text_card(**card_info)

效果图