第六章：企业微信回调接口完整实现（Webhook入口打通）

这一章是整个系统的第一个真实里程碑：

让企业微信成功“打通你的服务”，完成 URL 验证，并具备接收消息能力。

一旦这一章完成，你的系统就不再是“服务”，而是真正意义上的：

企业微信 AI 机器人入口系统

6.1 企业微信回调机制本质

企业微信回调不是普通 HTTP 请求，而是一个安全消息通道协议。

它包含三层机制：

HTTP层（请求入口）
   ↓
签名层（防伪造）
   ↓
加密层（AES消息内容）

📌 两种请求类型

① URL验证（GET）

用于首次绑定回调地址：

企业微信后台 → 验证回调URL

参数：

msg_signature
timestamp
nonce
echostr

② 消息推送（POST）

用户发送消息后：

企业微信 → POST → 你的服务

内容是：

<xml>
    <Encrypt>...</Encrypt>
    <MsgSignature>...</MsgSignature>
    <TimeStamp>...</TimeStamp>
    <Nonce>...</Nonce>
</xml>

6.2 项目结构准备（本章新增）

在 app/wecom/ 下新增模块：

app/wecom/
│
├── callback.py        # 回调入口
├── crypto.py          # 加解密
├── validator.py       # 签名验证
└── parser.py          # XML解析（本章新增）

6.3 FastAPI回调入口实现

📌 main.py 注册路由

from fastapi import FastAPI, Request
from app.wecom.callback import wecom_callback_router

app = FastAPI()

app.include_router(wecom_callback_router, prefix="/wecom")

📌 callback.py（核心入口）

from fastapi import APIRouter, Request
from app.wecom.crypto import WeComCrypto
from app.wecom.validator import verify_url

router = APIRouter()

crypto = WeComCrypto()


@router.get("/callback")
async def verify_callback(
    msg_signature: str,
    timestamp: str,
    nonce: str,
    echostr: str
):
    """
    企业微信URL验证
    """
    result = verify_url(
        msg_signature=msg_signature,
        timestamp=timestamp,
        nonce=nonce,
        echostr=echostr,
        crypto=crypto
    )

    return result

6.4 URL验证逻辑（第一关）

📌 verify_url.py

from app.wecom.crypto import WeComCrypto


def verify_url(msg_signature, timestamp, nonce, echostr, crypto: WeComCrypto):
    """
    企业微信URL验证逻辑
    """

    decrypted = crypto.decrypt_echo_str(
        msg_signature=msg_signature,
        timestamp=timestamp,
        nonce=nonce,
        echostr=echostr
    )

    return decrypted

📌 验证逻辑本质

企业微信发送加密 echostr
        ↓
服务端解密
        ↓
原样返回
        ↓
验证成功

6.5 企业微信加解密核心（crypto.py）

📌 WeComCrypto实现

from wechatpy.crypto import WeChatCrypto
import os


class WeComCrypto:
    def __init__(self):
        self.token = os.getenv("WECOM_TOKEN")
        self.aes_key = os.getenv("WECOM_AES_KEY")
        self.corp_id = os.getenv("WECOM_CORP_ID")

        self.crypto = WeChatCrypto(
            self.token,
            self.aes_key,
            self.corp_id
        )

    def decrypt_echo_str(self, msg_signature, timestamp, nonce, echostr):
        """
        URL验证解密
        """
        return self.crypto.check_signature(
            msg_signature,
            timestamp,
            nonce,
            echostr
        )

⚠️ 注意点（非常关键）

企业微信这里最容易错：

❌ 常见错误

  • Token不一致
  • AESKey写错
  • CorpID不对
  • URL未公网HTTPS
  • nginx未透传参数

6.6 POST消息入口（第二阶段）

📌 callback.py 增加 POST

@router.post("/callback")
async def receive_message(request: Request):
    """
    接收企业微信消息
    """

    body = await request.body()
    xml_data = body.decode("utf-8")

    print("收到原始XML:", xml_data)

    return "success"

6.7 XML结构解析（parser.py）

📌 解析加密消息

import xml.etree.ElementTree as ET


def parse_xml(xml_str: str):
    root = ET.fromstring(xml_str)

    data = {}

    for child in root:
        data[child.tag] = child.text

    return data

📌 企业微信POST结构

<xml>
    <ToUserName>...</ToUserName>
    <Encrypt>...</Encrypt>
</xml>

6.8 完整POST流程（当前简化版）

企业微信
   ↓
POST /wecom/callback
   ↓
FastAPI接收XML
   ↓
提取 Encrypt
   ↓
（下一步：解密）
   ↓
进入消息处理层

6.9 本章当前完成状态

✅ 已完成

✔ FastAPI回调路由
✔ GET URL验证入口
✔ AES解密封装（基础）
✔ XML解析工具
✔ POST入口接收

🚧 未完成（下一章要做）

下一步真正关键：

❌ msg_signature 校验（SHA1）
❌ AES解密 Encrypt消息
❌ 还原XML明文
❌ 消息分发 Dispatcher

🧠 本章核心总结

这一章完成后，你已经打通：

👉 企业微信 → 你的 FastAPI 服务入口

但目前还只是“收到消息”，还没有进入AI链路

🚀 下一章预告（第七章）

第七章：企业微信消息解密与完整回调链路

将完成：

msg_signature 校验
AES 解密 Encrypt
XML 明文解析
消息类型识别
Dispatcher 分发

👉 到这一章结束：

你的系统会真正开始“理解用户消息”。