企业微信 API 二次开发架构设计：个人微信与企业微信 API 双栈消息路由与状态同步方案

有团队同时用个人号 API 和企业微信 API，一边接私域个人好友，一边接企微客户。两套凭证、两套字段、两套返回格式，代码很容易写成一锅粥。这篇讲怎么在架构层把两栈分开。

先认清两条栈的差异

不是"换个 base_url"那么简单。核心字段对不上：

对比项	个人号 API	企业微信 API
实例标识	wId	appid（we_xxx）
联系人 ID	wcId	uin（int64）
会话 ID	—	conversationId（int64）
路径前缀	/sendText	/wx-api/api/message/sendText
成功 code	字符串 "1000"	数字 0
限流码	45009	-301002

code 一个是字符串一个是数字，这个差异会让统一的响应处理逻辑直接失效。用 if resp["code"] == 0 判断，个人号那边永远进不去成功分支。

统一消息模型

上层业务不该关心消息从哪来。先定义中间模型：

from dataclasses import dataclass, field
from typing import Any, Dict

@dataclass
class UnifiedMessage:
    stack: str                    # "personal" | "wecom"
    instance_id: str              # wId 或 appid
    sender_id: str                # wcId 或 uin（转成字符串）
    conversation_id: str          # 个人号用 sender_id 代替
    content: str
    content_type: int
    raw: Dict[str, Any] = field(default_factory=dict)

stack 字段必须带，它是后面所有路由、幂等、日志的区分依据。

适配器模式：两栈各自归一

class Adapter:
    stack = ""

    def normalize_webhook(self, payload): ...
    def send_text(self, msg): ...
    def is_success(self, resp): ...

class WecomAdapter(Adapter):
    stack = "wecom"

    def normalize_webhook(self, payload):
        d = payload["data"]
        return UnifiedMessage(
            stack="wecom",
            instance_id=payload["appid"],
            sender_id=str(d["senderUin"]),
            conversation_id=str(d["conversationId"]),
            content=d.get("content", ""),
            content_type=d.get("contentType", 1),
            raw=d,
        )

    def is_success(self, resp):
        return resp.get("code") == 0          # 数字

class PersonalAdapter(Adapter):
    stack = "personal"

    def is_success(self, resp):
        return resp.get("code") == "1000"     # 字符串

is_success 交给适配器自己判断，上层只调这一个方法。这是隔离差异最省事的位置。

路由表：实例到适配器

ADAPTERS = {}

def register(adapter):
    ADAPTERS[adapter.stack] = adapter

def route(stack, instance_id):
    adapter = ADAPTERS.get(stack)
    if not adapter:
        raise ValueError(f"未知栈: {stack}")
    if not instance_whitelist(stack, instance_id):
        raise ValueError(f"实例未授权: {stack}/{instance_id}")
    return adapter

路由的是"栈 + 实例"，不是一个笼统的"账号"。同一个栈下多个实例共用适配器，但凭证各自独立。

个人号和企业微信的凭证互不通用，实例也不能混用，这点必须在上线前明确写进配置校验里。

状态同步：最难的部分

两栈里"同一个人"是两个人。个人号那边是 wcId，企微那边是 uin，没有任何官方字段能可靠关联。

能做的只有弱关联：

手机号。两边都拿到手机号时，写进一张映射表。但手机号可能变，也可能拿不到。

外部标识。客户在个人号里留的微信 ID，和企微外部联系人 ID 不是一回事。

业务主键。如果客户在你自己的系统里有客户 ID，用这个做锚点最靠谱。

def link_identity(stack, sender_id, biz_customer_id, phone=None):
    if not biz_customer_id:
        return
    db.execute("""
        INSERT OR REPLACE INTO identity_map
        (stack, sender_id, customer_id, phone, updated_at)
        VALUES (?, ?, ?, ?, datetime('now'))
    """, (stack, sender_id, biz_customer_id, phone))

映射表要允许 "stack + sender_id" 唯一，但 customer_id 一对多。一个客户可能在两个栈都有会话，这是常态，不是脏数据。

幂等键必须带栈前缀

一个隐蔽的坑：两栈的 delivery id 或消息 id 可能长得像，甚至撞号。

def idem_key(stack, delivery_id):
    return f"cb:{stack}:{delivery_id}"

不带前缀，个人号的消息可能被企微的去重表吃掉。

日志和监控也要分栈

指标按栈打标签：

metrics.inc("webhook_received", tags={"stack": msg.stack})
metrics.inc("send_success", tags={"stack": msg.stack})

混在一起统计，个人号限流飙高会把企微的指标带偏，排查时误判。

三个设计取舍

要不要做统一发送接口。做，但入口参数里必须带 stack，不要试图在内部猜。

要不要共用一个队列。可以共用，但消费端按栈分不同 worker，避免个人号那边慢把企微堵死。

要不要同一套重试策略。不要。个人号限流码是 45009、企微是 -301002，退避曲线和阈值要各自配。

最后

双栈的价值是覆盖面，代价是复杂度。如果业务上能用一套栈解决，别为了"全都要"把架构撑大。

真要双栈，把差异锁在适配器里，上层只处理 UnifiedMessage，是唯一能长期维护的写法。

两条栈的路径前缀、字段类型和返回格式，在开发文档里分得很清楚，Eyun 控制台里个人号和企业微信实例是分开开通的，凭证也不通用。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2603_96320844/article/details/168882505