微信公众号开发实战：从零构建一个智能客服机器人（基于Python Flask）

2次阅读

没有评论

共计 3448 个字符，预计需要花费 9 分钟才能阅读完成。

微信公众号开发中，新手常会遇到几个典型问题：

access_token 失效 ：微信的 access_token 有效期只有 2 小时，且频繁获取会被限流，如何高效管理成为难题
XML 消息解析复杂 ：微信的消息格式使用 XML，相比 JSON 处理更繁琐，容易出错
高并发消息丢失 ：用户高峰期时，同步处理模式容易导致消息堆积和丢失
签名验证复杂 ：每次请求都需要验证微信服务器的签名（sha1 加密），新手容易遗漏步骤

对比直接调用官方 SDK 与自研框架：

官方 SDK 封装完善但灵活性差，难以定制特殊需求
自研框架可以按需优化，比如实现异步处理、自定义缓存策略等

使用 Python Flask 构建轻量级服务端：

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/wechat', methods=['GET', 'POST'])
def wechat_handler():
    # 这里是处理微信消息的入口
    pass

if __name__ == '__main__':
    app.run(port=5000)

微信要求验证消息签名，确保请求来自微信服务器：

import hashlib

def check_signature(token, timestamp, nonce, signature):
    tmp_list = [token, timestamp, nonce]
    tmp_list.sort()
    tmp_str = ''.join(tmp_list)
    sha1 = hashlib.sha1(tmp_str.encode('utf-8')).hexdigest()
    return sha1 == signature

实现带自动刷新的 access_token 池：

import time
import requests

class TokenManager:
    def __init__(self, appid, secret):
        self.appid = appid
        self.secret = secret
        self._token = None
        self._expires_at = 0

    @property
    def token(self):
        if time.time() > self._expires_at - 300:  # 提前 5 分钟刷新
            self.refresh_token()
        return self._token

    def refresh_token(self):
        url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={self.appid}&secret={self.secret}"
        resp = requests.get(url).json()
        self._token = resp['access_token']
        self._expires_at = time.time() + resp['expires_in']

使用 Redis 实现消息去重和异步队列：

import redis
from rq import Queue

r = redis.Redis()
q = Queue(connection=r)

# 消息去重 Lua 脚本
DEDUP_SCRIPT = """
local key = KEYS[1]
local value = ARGV[1]
local ttl = tonumber(ARGV[2])

if redis.call("GET", key) == value then
    return 0
else
    redis.call("SET", key, value, "EX", ttl)
    return 1
end
"""

# 使用示例
msg_id = "unique_message_id"
if r.eval(DEDUP_SCRIPT, 1, f"dedup:{msg_id}", msg_id, 3600):
    q.enqueue(process_message, msg_content)

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

class WXBizMsgCrypt:
    def __init__(self, token, encoding_aes_key, app_id):
        self.key = base64.b64decode(encoding_aes_key + "=")
        self.iv = self.key[:16]
        self.app_id = app_id

    def decrypt(self, encrypted_msg):
        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
        decrypted = cipher.decrypt(base64.b64decode(encrypted_msg))
        # 处理 PKCS#7 填充
        pad = ord(decrypted[-1:])
        content = decrypted[16:-pad]
        xml_len = int(content[:4].decode('utf-8'))
        xml_content = content[4:xml_len+4]
        return ET.fromstring(xml_content)

@app.route('/wechat', methods=['GET', 'POST'])
def wechat_handler():
    if request.method == 'GET':
        # 验证签名
        signature = request.args.get('signature')
        timestamp = request.args.get('timestamp')
        nonce = request.args.get('nonce')
        echostr = request.args.get('echostr')

        if check_signature(WX_TOKEN, timestamp, nonce, signature):
            return echostr
        else:
            return 'Invalid signature', 403

    else:
        # 处理消息
        try:
            xml_data = request.data
            msg = parse_xml_message(xml_data)

            # 异步处理消息
            q.enqueue(process_user_message, msg)

            # 立即回复空响应
            return ""
        except Exception as e:
            app.logger.error(f"Message process error: {str(e)}")
            return "", 500

控制接口调用频率，使用令牌桶算法限流
对于非实时性要求的操作（如群发），使用延迟队列
建立本地缓存，减少重复请求

使用 KMS 服务管理加密密钥
敏感配置信息加密存储
实现自动化的密钥轮换机制

使用消息 ID 去重
实现乐观锁机制
记录处理状态，支持重试

使用 Locust 进行压力测试：

from locust import HttpUser, task, between

class WechatUser(HttpUser):
    wait_time = between(0.1, 0.5)

    @task
    def send_message(self):
        self.client.post("/wechat", data=test_xml, headers={"Content-Type": "text/xml"})

测试结果对比：