企业官网建设流程全解析

项目里直接在各处散落 api("sendText", ...)、api("getContactList", ...) 调用，短期写得快，长期维护成本高。统一接口层解决的是"所有微信能力调用走同一扇门"的问题。

一、不封装会遇到什么问题

鉴权散落各处：每个调用点都要自己拼 wId 和 Token，漏传或传错只能到运行时才发现。错误处理不统一：有的调用点判断了返回码，有的没判断，同一种限频错误在不同地方表现不同。日志缺失：出问题时查不到"什么时候调了什么接口、参数是什么、返回了什么"。频控各写各的：发送消息的间隔控制散落在业务代码里，改频率要改十几处。

二、统一接口层的四个职责

鉴权注入：所有请求自动带上 wId 和 Token，业务代码不碰鉴权参数。错误处理：统一判断返回码，限频自动等待重试，鉴权失效触发告警，业务错误返回结构化异常。日志记录：每次调用记录接口名、参数（脱敏）、返回码、耗时，形成完整调用链。频控管理：按接口类型维护调用间隔和并发数，业务代码只管调，频控由接口层兜底。

三、接口层的边界——封装到什么程度

接口层封装的是"通用能力"（鉴权、错误、日志、频控），不封装业务逻辑。sendText 就是 sendText，不做"发送欢迎语"这种业务封装。业务封装放在更上层的服务层。

判断标准：如果一个封装逻辑换个业务就不适用，它不该放在接口层；如果所有业务都需要（如鉴权、频控），它必须放在接口层。

接口层职责对照

统一接口层实现

import time import logging class WeChatAPI: def __init__(self, base_url, token, wid): self.base_url = base_url self.token = token self.wid = wid self.rate_limiter = RateLimiter() def call(self, method, params=None, retry=2): """统一调用入口""" params = params or {} params["wId"] = self.wid # 鉴权注入 headers = {"Authorization": f"Bearer {self.token}"} for attempt in range(retry + 1): self.rate_limiter.wait(method) # 频控 start = time.time() try: resp = http_post(f"{self.base_url}/{method}", params, headers) elapsed = int((time.time() - start) * 1000) self._log(method, params, resp, elapsed) code = resp.get("code") if code == "1000": return resp if code == "1004" and attempt < retry: time.sleep(30) # 限频等待重试 continue raise APIError(code, resp.get("msg", "")) except (Timeout, ConnectionError): if attempt < retry: time.sleep(3) continue raise def _log(self, method, params, resp, elapsed): logging.info("api_call", extra={ "method": method, "params": self._mask(params), # 脱敏 "code": resp.get("code"), "elapsed_ms": elapsed }) def _mask(self, params): masked = dict(params) if "content" in masked: masked["content"] = masked["content"][:50] return masked # 业务代码只调封装好的方法 api = WeChatAPI(BASE_URL, TOKEN, WID) api.call("sendText", {"toUser": wxid, "content": msg})

落地建议

接口层在项目第一天就建好——哪怕只封装鉴权和日志两个职责，后续所有业务代码都走这一层。等散落调用多了再重构成本极高。频控配置放配置文件，不同接口的间隔限制按文档调整。接口返回码和频控说明参考 Eyun 开发文档。