企业微信自建应用：Python 实现消息收发功能

企业微信自建应用：Python 实现消息收发功能 | 极客日志

import json
import urllib.parse
import requests

# 企业 ID
corpid = 'your_corp_id'
# 应用 ID
agentid = 'your_agent_id'
# 应用 Secret
corpsecret = 'your_secret'
# 接收消息的用户
touser = '@all'

# 企业微信 API 的基础 URL
base = 'https://qyapi.weixin.qq.com'

# 构造函数 (获取 access-token 的 API URL)
access_token_api = urllib.parse.urljoin(base, '/cgi-bin/gettoken')

# 定义请求参数
params = {'corpid': corpid, 'corpsecret': corpsecret}

# 发送 GET 请求获取 access-token
response = requests.get(url=access_token_api, params=params).json()
access_token = response['access_token']

# 构建发送消息的完整 URL
message_send_api = urllib.parse.urljoin(base, f'/cgi-bin/message/send?access_token={access_token}')

# 定义要发送的消息数据（文本格式）
data = {
    'touser': touser,
    'msgtype': 'text',
    'agentid': agentid,
    'text': {'content': '测试数据：hello world!'}
}

# 发送 POST 请求以发送消息
response = requests.post(url=message_send_api, data=json.dumps(data)).json()

if response['errcode'] == 0:
    print('发送成功')
else:
    print(response)

pip install requests

pip install pycryptodome

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import PlainTextResponse, Response
from WXBizMsgCrypt3 import WXBizMsgCrypt
import xmltodict
import logging
import time
import xml.etree.ElementTree as ET

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

# 企业微信配置
TOKEN = 'your_token'
ENCODING_AES_KEY = 'your_encoding_aes_key'
CORP_ID = 'your_corp_id'

try:
    wxcpt = WXBizMsgCrypt(TOKEN, ENCODING_AES_KEY, CORP_ID)
except Exception as e:
    logger.error(f"初始化 WXBizMsgCrypt 失败：{str(e)}")
    raise

@app.get("/callback")
async def verify_url(msg_signature: str, timestamp: str, nonce: str, echostr: str):
    """验证 URL 有效性"""
    try:
        logger.info(f"收到验证请求：msg_signature={msg_signature}, timestamp={timestamp}, nonce={nonce}")
        ret, sEchoStr = wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
        if ret == 0:
            logger.info("URL 验证成功")
            return PlainTextResponse(content=sEchoStr)
        else:
            logger.error(f"URL 验证失败，错误码：{ret}")
            raise HTTPException(status_code=400, detail="验证失败")
    except Exception as e:
        logger.error(f"验证过程发生错误：{str(e)}")
        raise HTTPException(status_code=500, detail="服务器内部错误")

@app.post("/callback")
async def receive_message(request: Request):
    """接收并处理企业微信消息"""
    try:
        body = await request.body()
        msg_signature = request.query_params.get("msg_signature")
        timestamp = request.query_params.get("timestamp")
        nonce = request.query_params.get("nonce")

        if not all([msg_signature, timestamp, nonce]):
            raise HTTPException(status_code=400, detail="缺少必要的参数")

        logger.info(f"收到消息推送：msg_signature={msg_signature}, timestamp={timestamp}, nonce={nonce}")

        # 解密消息
        ret, sMsg = wxcpt.DecryptMsg(body, msg_signature, timestamp, nonce)
        if ret != 0:
            logger.error(f"消息解密失败，错误码：{ret}")
            raise HTTPException(status_code=400, detail="消息解密失败")

        # 解析 XML 消息
        xml_dict = xmltodict.parse(sMsg)
        logger.info(f"解密后的消息内容：{xml_dict}")

        xml_content = xml_dict['xml']
        to_user_name = xml_content.get('ToUserName')
        from_user_name = xml_content.get('FromUserName')
        create_time = xml_content.get('CreateTime')
        msg_type = xml_content.get('MsgType')
        content = xml_content.get('Content')
        msg_id = xml_content.get('MsgId')
        agent_id = xml_content.get('AgentID')

        logger.info(f"收到消息：{content}")

        # 构造回复消息
        reply_content = content.replace('吗', '').replace('?', '!').replace('？', '！')
        current_time = str(int(time.time()))
        reply_msg = f"""<xml>
<ToUserName><![CDATA[{from_user_name}]]></ToUserName>
<FromUserName><![CDATA[{to_user_name}]]></FromUserName>
<CreateTime>{current_time}</CreateTime>
<MsgType><![CDATA[text]]></MsgType>
<Content><![CDATA[{reply_content}]]></Content>
<MsgId>{msg_id}</MsgId>
<AgentID>{agent_id}</AgentID>
</xml>"""

        # 加密回复消息
        ret, encrypted_msg = wxcpt.EncryptMsg(reply_msg, nonce, current_time)
        if ret != 0:
            logger.error(f"消息加密失败，错误码：{ret}")
            raise HTTPException(status_code=500, detail="消息加密失败")

        logger.info("成功构造并加密回复消息")
        return Response(content=encrypted_msg, media_type="application/xml")
    except Exception as e:
        logger.error(f"处理消息时发生错误：{str(e)}")
        raise HTTPException(status_code=500, detail="服务器内部错误")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000)

企业微信自建应用：Python 实现消息收发功能

企业微信自建应用：Python 实现消息收发功能

第一步：配置企业微信应用

第二步：测试发送消息

第三步：集成加解密库

第四步：开发消息收发逻辑

第五步：配置回调地址

更多推荐文章

相关免费在线工具

企业微信自建应用：Python 实现消息收发功能

企业微信自建应用：Python 实现消息收发功能

第一步：配置企业微信应用

第二步：测试发送消息

第三步：集成加解密库

第四步：开发消息收发逻辑

第五步：配置回调地址

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具