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

企业微信Python SDK：高效群发消息实战

news 2026/8/8 23:46:43

1. 核心鉴权逻辑封装：Access Token 的管理

所有对QiWe 开放平台API 的调用都需要有效的Access Token。SDK 的首要任务是高效管理这个 Token 的生命周期。Token 获取和刷新应在 SDK 内部自动处理，对用户透明。

根据架构图，鉴权依赖于企业 ID (CorpID)和应用密钥 (Secret)。

Python SDK 鉴权实现示例

Python

import requests import time class QiWeApiClient: def __init__(self, corp_id: str, client_secret: str, api_base_url: str): self.corp_id = corp_id self.client_secret = client_secret self.api_base_url = api_base_url self._access_token = None self._token_expires_at = 0 def _get_access_token(self) -> str: """ 自动检查并刷新 Access Token """ # 预留 300 秒（5分钟）缓冲时间进行刷新 if time.time() < self._token_expires_at - 300 and self._access_token: return self._access_token print("Token expired or close to expiration. Refreshing token...") # 模拟调用 获取企业令牌 接口 auth_url = f"{self.api_base_url}/auth/token" # 实际生产环境应使用分布式锁防止并发刷新 try: response = requests.post(auth_url, json={ "corp_id": self.corp_id, "client_secret": self.client_secret }) response.raise_for_status() data = response.json() token = data.get("access_token") expires_in = data.get("expires_in", 7200) # 默认 7200s self._access_token = token self._token_expires_at = time.time() + expires_in print("Token refresh successful.") return token except requests.RequestException as e: print(f"Token refresh failed: {e}") raise def _send_request(self, method: str, endpoint: str, **kwargs) -> dict: """ 通用请求发送方法，自动注入 Token """ token = self._get_access_token() url = f"{self.api_base_url}{endpoint}" headers = kwargs.pop('headers', {}) headers['Authorization'] = f'Bearer {token}' try: response = requests.request(method, url, headers=headers, **kwargs) response.raise_for_status() return response.json() except requests.HTTPError as e: # 错误码处理：例如 400xx 业务错误码或 401/403 鉴权错误 print(f"API Error ({e.response.status_code}): {e.response.text}") raise

2. 外部群消息的主动发送封装

主动发送外部群消息属于消息/会话/群聊相关功能模块。SDK 应该封装一个简单的方法，屏蔽底层异步执行和消息体封装的复杂性。

Python SDK 消息发送实现示例

Python

def send_external_group_text_message(self, chat_id: str, content: str) -> dict: """ 向指定的外部群发送文本消息 :param chat_id: 目标外部群聊 ID :param content: 文本内容 :return: API 响应数据 """ # 1. 消息体封装与校验 payload = { "chat_id": chat_id, "msg_type": "text", "content": { "text": content } } # 2. 调用发送端点（假设端点为 /messages/send_external_chat） endpoint = "/messages/send_external_chat" try: # 注意：底层非官方接口执行可能涉及异步任务投递， # 此处返回的 data 可能包含一个任务 ID，而非最终发送结果。 # 最终结果需通过回调或轮询机制获取。 response_data = self._send_request( method="POST", endpoint=endpoint, json=payload ) print(f"Message sent for group {chat_id}. Response: {response_data.get('task_id')}") return response_data except Exception as e: print(f"Failed to send message to {chat_id}: {e}") raise # --- 使用示例 --- # client = QiWeApiClient( # corp_id="YOUR_CORP_ID", # client_secret="YOUR_SECRET", # api_base_url="https://api.qiweapi.com/v1" # ) # try: # # 假设外部群 ID 为 wc123456 # client.send_external_group_text_message( # chat_id="wc123456", # content="这是一条通过 QiWe 开放平台发送的自动化消息。" # ) # except Exception: # # 异常处理逻辑... # pass