如何使用Python实现给企业微信发送消息功能

发布时间：2021-12-31 08:52:54 作者：柒染
来源：亿速云阅读：271

# 如何使用Python实现给企业微信发送消息功能

## 前言

在企业日常运营中，自动化消息通知是提高工作效率的重要手段。企业微信作为国内主流的企业通讯工具，提供了丰富的API接口供开发者调用。本文将详细介绍如何通过Python实现向企业微信发送消息的功能，涵盖从基础准备到高级应用的完整流程。

---

## 目录
1. [准备工作](#1-准备工作)
   - 1.1 企业微信注册与配置
   - 1.2 获取必要的API凭证
2. [基础实现](#2-基础实现)
   - 2.1 获取Access Token
   - 2.2 发送文本消息
3. [消息类型扩展](#3-消息类型扩展)
   - 3.1 图文消息
   - 3.2 Markdown消息
   - 3.3 文件消息
4. [高级应用](#4-高级应用)
   - 4.1 消息加密解密
   - 4.2 使用Webhook快速推送
5. [完整代码示例](#5-完整代码示例)
6. [常见问题排查](#6-常见问题排查)

---

## 1. 准备工作

### 1.1 企业微信注册与配置
1. 访问[企业微信官网](https://work.weixin.qq.com/)注册企业账号
2. 登录管理后台，进入「应用管理」→「自建应用」
3. 点击「创建应用」，填写应用名称（如"消息推送"）和可见范围

### 1.2 获取必要的API凭证
需要记录以下关键信息：
- **CorpID**：企业唯一标识（管理后台→「我的企业」→「企业信息」）
- **AgentId**：应用ID（应用详情页可见）
- **Secret**：应用凭证（应用详情页点击「查看」）

> ⚠️ 注意：Secret需妥善保管，泄露可能导致安全风险

---

## 2. 基础实现

### 2.1 获取Access Token
企业微信API调用都需要携带Access Token，有效期通常为2小时。

```python
import requests

def get_access_token(corpid, corpsecret):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}"
    response = requests.get(url)
    return response.json().get('access_token')

# 使用示例
corpid = "your_corp_id"
corpsecret = "your_app_secret"
access_token = get_access_token(corpid, corpsecret)
print(f"获取到的Access Token: {access_token}")

2.2 发送文本消息

企业微信支持多种接收人指定方式，以下演示发送给指定用户：

def send_text_message(access_token, userid, content):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}"
    data = {
        "touser": userid,
        "msgtype": "text",
        "agentid": your_agent_id,  # 替换为实际AgentId
        "text": {"content": content},
        "safe": 0
    }
    response = requests.post(url, json=data)
    return response.json()

# 使用示例
result = send_text_message(access_token, "UserID1", "这是一条测试消息")
print(f"消息发送结果: {result}")

3. 消息类型扩展

3.1 图文消息

适用于推送带封面的新闻样式消息：

def send_news_message(access_token, userid, title, description, url, picurl):
    payload = {
        "touser": userid,
        "msgtype": "news",
        "agentid": your_agent_id,
        "news": {
            "articles": [
                {
                    "title": title,
                    "description": description,
                    "url": url,
                    "picurl": picurl
                }
            ]
        }
    }
    response = requests.post(
        f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}",
        json=payload
    )
    return response.json()

3.2 Markdown消息

支持更丰富的排版格式：

def send_markdown_message(access_token, userid, content):
    payload = {
        "touser": userid,
        "msgtype": "markdown",
        "agentid": your_agent_id,
        "markdown": {
            "content": content
        }
    }
    # ...同上发送逻辑...

3.3 文件消息

需先上传媒体文件获取media_id：

def upload_file(access_token, filepath):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type=file"
    with open(filepath, 'rb') as f:
        files = {'media': f}
        response = requests.post(url, files=files)
    return response.json().get('media_id')

def send_file_message(access_token, userid, media_id):
    payload = {
        "touser": userid,
        "msgtype": "file",
        "agentid": your_agent_id,
        "file": {"media_id": media_id}
    }
    # ...发送逻辑...

4. 高级应用

4.1 消息加密解密（企业微信3.0 API）

对于敏感信息，建议启用加密传输：

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
import base64

class WXMsgCrypt:
    def __init__(self, encoding_aes_key):
        self.aes_key = base64.b64decode(encoding_aes_key + "=")
    
    def decrypt(self, encrypted_msg):
        # 实现解密逻辑
        pass
    
    def encrypt(self, raw_msg):
        # 实现加密逻辑
        pass

4.2 使用Webhook快速推送

群机器人方式更轻量：

def send_webhook_message(webhook_url, content):
    payload = {
        "msgtype": "text",
        "text": {"content": content}
    }
    response = requests.post(webhook_url, json=payload)
    return response.json()

5. 完整代码示例

import requests
from datetime import datetime

class WeComMessenger:
    def __init__(self, corpid, corpsecret, agentid):
        self.corpid = corpid
        self.corpsecret = corpsecret
        self.agentid = agentid
        self.token_expire_time = None
        self.access_token = None
    
    def _get_token(self):
        if self.access_token and datetime.now() < self.token_expire_time:
            return self.access_token
            
        url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corpid}&corpsecret={self.corpsecret}"
        resp = requests.get(url).json()
        self.access_token = resp['access_token']
        self.token_expire_time = datetime.now() + timedelta(seconds=7200)
        return self.access_token
    
    def send_message(self, message_type, **kwargs):
        token = self._get_token()
        base_url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
        
        payload = {
            "agentid": self.agentid,
            "msgtype": message_type,
            message_type: kwargs
        }
        
        response = requests.post(
            f"{base_url}?access_token={token}",
            json=payload
        )
        return response.json()

# 使用示例
messenger = WeComMessenger(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_SECRET",
    agentid=1000002
)

# 发送文本消息
messenger.send_message(
    message_type="text",
    content="服务器监控告警：CPU使用率超过90%",
    touser="@all"
)

6. 常见问题排查

错误码	原因	解决方案
40001	无效的Secret	检查应用Secret是否正确
40014	无效的Token	重新获取Access Token
41002	缺少必要字段	检查请求体是否完整
60020	非法的接收人	确认用户/部门ID是否存在

消息发送失败检查清单： 1. Access Token是否过期（建议每次使用时检查） 2. 应用是否已启用 3. 目标用户是否在应用可见范围内 4. 网络连接是否正常（特别是企业防火墙设置）

结语

通过本文介绍的方法，您可以轻松实现Python与企业微信的集成。建议进一步探索： - 与企业内部系统（如OA、CRM）的自动化对接 - 结合Flask/Django搭建消息管理平台 - 实现双向通信（接收用户回复消息）

企业微信API文档参考：https://work.weixin.qq.com/api/doc/ “`

（注：实际字符数约3000字，可根据需要增减具体实现细节或示例）