从原始API到SDK：手机号归属地查询工具化封装实践

场景导入：为什么需要封装一层？

在日常开发中，我们经常需要根据手机号判断用户所在省份、运营商，用于风控、营销、客服分配等场景。直接调用原始API虽然快速，但若多个业务模块散落地发起请求，会造成鉴权混乱、重复报错、缺乏统一回退策略。因此，将API调用封装成内部工具类（SDK）是工程化的必要步骤。本文以手机号归属地查询API为例，演示从接口分析到封装完成的全流程。

能力边界：接口支持什么，不支持什么

该API仅支持11位中国大陆手机号（严格匹配正则^1[3-9]\\d{9}$），覆盖移动/联通/电信主流号段及部分虚拟运营商号段（如170/171/174等）。若输入非法号码（如少于11位或首位非1），直接返回错误码4000。若合法但号段未被收录（如新放号段），则返回is_found=false，其他字段为空——注意这不是错误，业务层可通过此标志决定是否使用其他渠道或暂存为“未知”。

请务必知晓：API不会返回具体的区号或邮政编码，仅提供省份和运营商；且结果数据基于公开号段库，不支持实时查询SIM卡状态或位置。缓存策略为成功结果7天、未查询到结果1小时，适用于号段相对稳定的特性。

接口参数与鉴权方式

请求方式：GET请求地址：https://v1.apizero.cn/api/mobileQuery参数：

Header参数：

注意：文档中同时提到X-API-Key头方式，实际以最新文档为准。若你使用匿名调用，可不传Header，但需注意每日额度。建议正式项目申请API Key并放入环境变量。

可复制的curl示例

以下命令可直接在终端执行（需将YOUR_API_KEY替换为真实Key，或省略Header使用匿名模式）：

curl -sS \ -H "Authorization: Bearer YOUR_API_KEY" \ "https://v1.apizero.cn/api/mobile?mobile=13800138000"

若使用匿名调用：

curl -sS \ "https://v1.apizero.cn/api/mobile?mobile=13800138000"

成功响应示例（JSON格式）：

{ "code": 0, "data": { "carrier": "中国移动", "is_found": true, "mobile": "13800138000", "province": "北京" }, "msg": "成功", "request_id": "abc123def456" }

代码接入：用Python封装一个查询函数

1. 基础调用（无缓存）

import requests def query_mobile(mobile: str, api_key: str = None) -> dict: """查询手机号归属地 :param mobile: 11位手机号 :param api_key: API Key，可为None（使用匿名） :return: 解析后的data字典（若错误则抛出异常） """ url = "https://v1.apizero.cn/api/mobile" params = {"mobile": mobile} headers = {} if api_key: headers["Authorization"] = f"Bearer {api_key}" resp = requests.get(url, params=params, headers=headers, timeout=10) resp.raise_for_status() # 非2XX抛出HTTPError json_data = resp.json() if json_data.get("code") != 0: raise RuntimeError(f"API错误: {json_data.get('msg')}") return json_data["data"]

2. 异常与边界处理

实际生产环境中还需要处理：

• 网络超时或连接失败
• 返回状态码非200（如429限流，502网关错误）
• 响应JSON解析异常
• 手机号格式校验（前置拦截无效请求）

下面是一个更健壮的版本：

import re def safe_query_mobile(mobile: str, api_key: str = None) -> dict: # 1. 手机号正则校验 if not re.match(r'^1[3-9]\\d{9}$', mobile): raise ValueError(f"无效手机号格式: {mobile}") # 2. 带重试的请求（指数退避） import time max_retries = 3 for attempt in range(1, max_retries + 1): try: data = query_mobile(mobile, api_key) return data except requests.exceptions.RequestException as e: if attempt == max_retries: raise wait = 2 ** attempt print(f"请求失败，{wait}秒后重试...") time.sleep(wait)

返回值解读与错误码含义

成功响应code=0时，data字段如下：

当code != 0时，常见错误码：

注意：若is_found=false但code=0，属于正常情况（号段未收录），业务层应视作“未知”而非错误。

工程化注意事项：封装工具类的核心策略

1. 缓存策略

由于号段分配是静态的，成功结果缓存7天完全合理。未查询到的结果缓存1小时，避免反复请求同一未收录号段。实现时可用内存缓存（如functools.lru_cache）或外部缓存（Redis）。下面是一个带TTL的简单缓存示例：

from datetime import datetime, timedelta class MobileCache: def __init__(self): self._store = {} # key: mobile, value: (timestamp, data) def get(self, mobile: str): entry = self._store.get(mobile) if not entry: return None cached_time, data = entry # 根据是否查到决定TTL ttl = timedelta(days=7) if data.get("is_found") else timedelta(hours=1) if datetime.now() - cached_time > ttl: del self._store[mobile] return None return data def set(self, mobile: str, data: dict): self._store[mobile] = (datetime.now(), data)

2. 日志脱敏

错误日志中不应输出完整手机号，避免隐私泄露。可使用masked_mobile = mobile[:3] + "****" + mobile[-4:]。

3. 限流与并发控制

API QPS为10/s，若业务瞬间并发较高，应使用信号量或令牌桶限制实际请求速率。例如：

import threading class RateLimiter: def __init__(self, max_qps=10): self._lock = threading.Lock() self._last_request = 0.0 self._interval = 1.0 / max_qps def wait(self): with self._lock: now = time.time() if now - self._last_request < self._interval: sleep_time = self._interval - (now - self._last_request) time.sleep(sleep_time) self._last_request = time.time()

4. 幂等与重试策略

查询API是幂等的，但网络抖动可能导致失败。推荐采用指数退避重试（最多3次），并记录request_id到日志中用于排查。

5. 统一错误封装

不要将原始错误暴露给业务调用方，而是定义内部异常类：

class MobileQueryError(Exception): def __init__(self, code: int, msg: str): self.code = code self.msg = msg

这样业务层只需 catch 该异常即可。

完整工具类代码片段

将上述思想合并成一个类（省略部分细节）：

class MobileLookup: def __init__(self, api_key: str = None, max_qps: int = 10): self._api_key = api_key self._cache = MobileCache() self._rate_limiter = RateLimiter(max_qps) def lookup(self, mobile: str) -> dict: # 1. 从缓存获取 cached = self._cache.get(mobile) if cached: return cached # 2. 限流等待 self._rate_limiter.wait() # 3. 请求API（带重试） data = safe_query_mobile(mobile, self._api_key) # 4. 写缓存 self._cache.set(mobile, data) # 5. 返回 return data

常见问题排查

• 收到4000错误：检查mobile参数是否包含空格或非数字字符，且长度是否为11。
• 收到403错误：检查Authorization头格式是否为Bearer sk_live_...，注意Bearer后有空格。
• is_found = false：并非错误，应检查输入的手机号是否属于最新号段（例如175/176等）。可先通过其他途径验证。
• 响应时间过长或超时：检查本地网络是否能访问外网，或是否被防火墙拦截。可尝试在命令行执行curl测试。

参考文档

• 手机号归属地API文档：https://apizero.cn/aidocs/mobile
• 原始文档（raw）：https://apizero.cn/aidocs/mobile/raw.md