调用官方 API 最让人头疼也最容易踩坑的环节有两个:第一是怎么证明“我是我”(接口鉴权);第二是为什么辛辛苦苦写的代码发出去总是报错(错误码与异常处理)。今天咱们把企业微信接口调用的底层逻辑一次性彻底说清楚。
核心技术拆解
  1. 全局凭证 Access Token:企业微信几乎所有的 API(除了群机器人 Webhook)都需要在请求地址里带上一个 access_token。这个 Token 的有效期通常是 2 小时,绝对不能每次请求都去重新申请,否则会瞬间触发频率限制(Rate Limit)。正确的做法是:全局缓存,提前刷新

  2. 错误码机制(errcode):企业微信的所有接口无论成功与否,都会返回一个 JSON,其中必定包含 errcodeerrmsg。只要 errcode !== 0,就代表出错了。必须针对常见的错误码(如 40014:不合法的 access_token,45009:接口调用超过限制)编写专门的重试或降级处理逻辑。

下面我们用 Python 编写一个兼顾 Token 缓存机制与异常错误处理的图文消息发送封装类。

完整代码实现(Python)

use-strict

import requests
import time
import json
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class WeChatClient:
    def __init__(self, corpid, corpsecret):
        self.corpid = corpid
        self.corpsecret = corpsecret
        self.access_token = None
        self.token_expires_at = 0

    def get_access_token(self):
        """
        获取 Access Token,内置内存缓存机制,避免频繁请求
        """
        current_time = time.time()
        # 如果 Token 还有效(提前 200 秒过期刷新),直接返回缓存值
        if self.access_token and current_time < self.token_expires_at:
            return self.access_token

        url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corpid}&corpsecret={self.corpsecret}"
        try:
            response = requests.get(url, timeout=5)
            res_data = response.json()
            
            if res_data.get("errcode") == 0:
                self.access_token = res_data.get("access_token")
                # 凭证有效期一般为 7200 秒,这里设定在 7000 秒后强制过期刷新
                expires_in = res_data.get("expires_in", 7200)
                self.token_expires_at = current_time + expires_in - 200
                logging.info("成功刷新并获取到新的 Access Token")
                return self.access_token
            else:
                logging.error(f"获取 Access Token 失败: {res_data}")
                return None
        except Exception as e:
            logging.error(f"获取 Access Token 网络请求异常: {e}")
            return None

    def send_app_message(self, user_id, title, description, url):
        """
        向指定企业员工发送图文类型的应用消息
        """
        token = self.get_access_token()
        if not token:
            logging.error("无法发送消息,因为未获取到有效的 Access Token。")
            return False

        api_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
        
        payload = {
            "touser": user_id,
            "msgtype": "news",
            "agentid": 1000002, # 替换成你自建应用的 AgentID
            "news": {
                "articles": [
                    {
                        "title": title,
                        "description": description,
                        "url": url,
                        "picurl": "https://example.com/logo.png"
                    }
                ]
            }
        }

        try:
            response = requests.post(api_url, data=json.dumps(payload), timeout=5)
            result = response.json()
            
            errcode = result.get("errcode")
            if errcode == 0:
                logging.info(f"成功向用户 {user_id} 发送图文消息!")
                return True
            elif errcode == 40014:
                # 专属处理:Token 过期或失效,清空缓存后递归重试一次
                logging.warning("Access Token 已失效,正在尝试重新获取并重试...")
                self.access_token = None
                self.token_expires_at = 0
                return self.send_app_message(user_id, title, description, url)
            else:
                logging.error(f"发送应用消息出错,错误码: {errcode}, 原因: {result.get('errmsg')}")
                return False
                
        except Exception as e:
            logging.error(f"发送应用消息时发生网络异常: {e}")
            return False

# --- 实战演练 ---
if __name__ == "__main__":
    client = WeChatClient("YOUR_CORPID", "YOUR_CORPSECRET")
    # client.send_app_message("ZhangSan", "系统运维周报发布", "点击查看本周服务器运行健康报告", "https://example.com/report")

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐