
项目里直接在各处散落 api(sendText, ...)、api(getContactList, ...) 调用短期写得快长期维护成本高。统一接口层解决的是所有微信能力调用走同一扇门的问题。一、不封装会遇到什么问题鉴权散落各处每个调用点都要自己拼 wId 和 Token漏传或传错只能到运行时才发现。错误处理不统一有的调用点判断了返回码有的没判断同一种限频错误在不同地方表现不同。日志缺失出问题时查不到什么时候调了什么接口、参数是什么、返回了什么。频控各写各的发送消息的间隔控制散落在业务代码里改频率要改十几处。二、统一接口层的四个职责鉴权注入所有请求自动带上 wId 和 Token业务代码不碰鉴权参数。错误处理统一判断返回码限频自动等待重试鉴权失效触发告警业务错误返回结构化异常。日志记录每次调用记录接口名、参数脱敏、返回码、耗时形成完整调用链。频控管理按接口类型维护调用间隔和并发数业务代码只管调频控由接口层兜底。三、接口层的边界——封装到什么程度接口层封装的是通用能力鉴权、错误、日志、频控不封装业务逻辑。sendText 就是 sendText不做发送欢迎语这种业务封装。业务封装放在更上层的服务层。判断标准如果一个封装逻辑换个业务就不适用它不该放在接口层如果所有业务都需要如鉴权、频控它必须放在接口层。接口层职责对照职责不封装的问题接口层做什么鉴权漏传wId/Token自动注入错误处理各处判断不一致统一码表重试日志出问题无法追溯全量调用记录频控改频率要改多处集中配置管理统一接口层实现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, paramsNone, retry2): 统一调用入口 params params or {} params[wId] self.wid # 鉴权注入 headers {Authorization: fBearer {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 开发文档。