当前位置：首页 > article >正文

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

article 2025/10/9 7:02:02

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

前言

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

配置需要：

一台公网可访问的服务器，用来和企业微信通信验证（仅做验证用，用后即可删除）
一个公司的域名，如公司官网为www.abc.com，则需要xxx.abc.com（仅做验证用，用后即可删除）
一个固定公网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 syssys.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.documentElementname_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].dataif msg_type == "text":  # 文本消息name = name_xml[0].childNodes[0].data  # 发送者idmsg = 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].datapic_url = pic_xml[0].childNodes[0].dataf.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"    # 文件URLclass 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 Falsemessage = {"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 Falsedef _format_description(self, desc):formatted = desc.replace('\n', '<br>')return formatted[:1500] + "..." if len(desc) > 1500 else formatteddef 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 Nonedata = response.json()if data.get("errcode") == 0:return data.get("access_token")print(f"获取Token失败：{data.get('errmsg')}")return Noneexcept Exception as e:print(f"获取Token异常：{str(e)}")return Nonedef _handle_response(self, response):try:if response.status_code != 200:print(f"请求失败，状态码：{response.status_code}")return Falseresult = response.json()if result.get('errcode') == 0:print("操作成功")return Trueprint(f"操作失败：[{result.get('errcode')}] {result.get('errmsg')}")return Falseexcept json.JSONDecodeError:print("响应解析失败")return Falseif __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)

效果图
在这里插入图片描述

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

前言

操作步骤

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

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

3. 配置企业可信IP

4. 配置接收消息服务器

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

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

7. 接口开发

相关文章：