企业微信接入 WorkBuddy 的技术实现

WorkBuddy 作为一款智能工作助手，接入企业微信可显著提升企业内部协作效率。以下从配置流程、API 调用、消息处理等角度，结合代码示例详细说明技术实现方案。

企业微信应用配置

在企业微信管理后台创建自建应用，获取以下关键信息：

• CorpID：企业唯一标识
• AgentID：应用ID
• Secret：应用密钥

# 示例：配置参数存储
WORKBUDDY_CONFIG = {
    "corp_id": "wwxxxxxx",
    "agent_id": 1000002,
    "secret": "xxxxxxxxxxxxxxxx"
}

获取 Access Token

调用企业微信API需携带有效的Access Token，通过GET请求获取并缓存：

import requests
import time

def get_access_token(corp_id, secret):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corp_id}&corpsecret={secret}"
    response = requests.get(url).json()
    if response["errcode"] == 0:
        return response["access_token"]
    else:
        raise Exception(f"Token获取失败: {response}")

# 缓存Token示例
token_cache = {
    "token": None,
    "expire_time": 0
}

def ensure_valid_token():
    if time.time() < token_cache["expire_time"]:
        return token_cache["token"]
    token = get_access_token(WORKBUDDY_CONFIG["corp_id"], WORKBUDDY_CONFIG["secret"])
    token_cache.update({"token": token, "expire_time": time.time() + 7000})  # 有效期7200秒
    return token

消息接收与解析

企业微信通过回调推送用户消息，需配置可信域名并验证签名。以下为消息解密示例：

from Crypto.Cipher import AES
import base64
import xml.etree.ElementTree as ET

def decrypt_msg(encrypt_msg, encoding_aes_key):
    key = base64.b64decode(encoding_aes_key + "=")
    cipher = AES.new(key, AES.MODE_CBC, key[:16])
    decrypted = cipher.decrypt(base64.b64decode(encrypt_msg))
    pad = ord(decrypted[-1:])
    content = decrypted[:-pad].decode("utf-8")
    return ET.fromstring(content.split(">")[1] + ">")  # 提取XML内容

主动发送消息

通过企业微信API向用户发送文本消息：

def send_text_message(user_id, content):
    token = ensure_valid_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    payload = {
        "touser": user_id,
        "msgtype": "text",
        "agentid": WORKBUDDY_CONFIG["agent_id"],
        "text": {"content": content},
        "safe": 0
    }
    response = requests.post(url, json=payload).json()
    if response["errcode"] != 0:
        print(f"消息发送失败: {response}")

用户身份验证

通过OAuth2.0获取用户详细信息，实现WorkBuddy与企业微信账号绑定：

def get_user_info(code):
    token = ensure_valid_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo?access_token={token}&code={code}"
    response = requests.get(url).json()
    if response["errcode"] == 0:
        return response["UserId"]  # 企业微信用户唯一标识

事件处理框架

处理订阅、菜单点击等事件需实现回调接口：

from flask import Flask, request

app = Flask(__name__)

@app.route("/workbuddy/callback", methods=["POST"])
def handle_callback():
    msg_signature = request.args.get("msg_signature")
    timestamp = request.args.get("timestamp")
    nonce = request.args.get("nonce")
    data = request.get_data()
    
    # 验证签名并解密（略）
    xml_data = decrypt_msg(data, encoding_aes_key)
    msg_type = xml_data.find("MsgType").text
    
    if msg_type == "event":
        event = xml_data.find("Event").text
        if event == "click":
            handle_menu_click(xml_data)
    return "success"

关键注意事项

• IP白名单：调用企业微信API需配置服务器IP至企业微信后台。
• 消息加密：启用加密模式时需完整实现加解密逻辑。
• 频率限制：Access Token获取限制为2000次/天，需合理缓存。

通过上述技术实现，WorkBuddy可无缝集成至企业微信，提供消息互通、任务协同等能力。实际部署时需结合日志监控与错误重试机制确保稳定性。