手机归属地查询：用 Python 给 Excel 号码表批量补归属地
约 14 分钟 进阶手机号码归属地查询批量查询Python 脚本Excel 处理
# 手机归属地查询：用 Python 给 Excel 号码表批量补归属地
接口：手机归属地查询（apiCode=6，接入点 6-1）｜是否免费：是，注册后默认可调用，设使用档次限制｜请求方式：POST / GET｜返回格式：JSON｜适用人群：需要给号码清单补地域标签的数据与运营同学｜阅读时间：约 9 分钟｜最后实测核对：2026-09-18
拿到一张号码表要补归属地，手工一条条查不现实，直接写个并发循环打接口又会撞上并发上限。先把两件事确认清楚：手机归属地查询接口没有官方批量能力，接入点 6-1 是单条同步请求；但归属地结果只跟前 7 位号段有关，先去重再查，请求数会明显下降。
## 核心要点
- 接入点 6-1 是单条同步请求-响应，接口没有批量入口，批量靠客户端循环实现。
- 号码表先取前 7 位去重，实测后 4 位不影响归属地结果，几万个号码通常只剩几百个不同号段。
- 并发上限由档位决定，L0 档 1 QPS，写循环时不要盲目开多线程。
## 两步预处理：去重和排序
号码表里同一个号段的号码往往很多，一个公司内部的号码通常集中在少数几个号段。实测确认 `13800130000` 和 `13800139999` 的返回完全一致，`13800240000` 才落到另一个城市，说明匹配键是前 7 位。
```python
import re
from collections import Counter
def to_segment(raw) -> str | None:
"""清理并取前 7 位号段，不合规返回 None。"""
if raw is None:
return None
cleaned = re.sub(r"[-\s]", "", str(raw).strip())
if not re.fullmatch(r"\d{7,}", cleaned):
return None
if cleaned[0] == "0":
return None
return cleaned[:7]
raw_numbers = [row[0] for row in rows] # 从表里读出来的原始号码列
segments = [s for s in (to_segment(n) for n in raw_numbers) if s]
unique_segments = sorted(set(segments))
print(f"号码 {len(raw_numbers)} 个，其中格式不合规 {len(raw_numbers) - len(segments)} 个")
print(f"去重后需要查询的号段：{len(unique_segments)} 个")
```
先把这两个数字打出来再决定要不要跑。多数情况下这一步就能把请求数压下来一个数量级。
## 逐条查询并写回
```python
import csv
import time
import requests
API_URL = "https://route.showapi.com/6-1"
APP_KEY = "YOUR_APPKEY" # 正式环境从配置或密钥管理服务读取
SEGMENT_FIELDS = ("prov", "city", "name", "type", "num",
"areaCode", "postCode", "provCode", "cityCode")
def locate(seg: str, retries: int = 3) -> dict:
"""查询单个号段，带指数退避重试。"""
for i in range(retries):
try:
resp = requests.get(
API_URL,
params={"appKey": APP_KEY, "num": seg},
timeout=10, # 对齐官方 x-read-timeout
)
data = resp.json()
if data.get("showapi_res_code") != 0:
raise RuntimeError(data.get("showapi_res_error"))
body = data.get("showapi_res_body", {})
if body.get("ret_code") != 0:
# 失败不扣点数，号段问题直接记录，不再重试
return {"ok": False, "ret_code": body.get("ret_code"),
"remark": body.get("remark")}
return {"ok": True, **{k: body.get(k) for k in SEGMENT_FIELDS}}
except requests.RequestException as exc:
if i == retries - 1:
return {"ok": False, "ret_code": None, "remark": f"网络异常：{exc}"}
time.sleep(2 ** i) # 1s / 2s / 4s 退避
return {"ok": False, "ret_code": None, "remark": "未知错误"}
# 断点续跑：结果落盘，中断后重新执行会跳过已查过的号段
segment_map = {}
try:
with open("segment_result.jsonl", encoding="utf-8") as fh:
for line in fh:
rec = __import__("json").loads(line)
segment_map[rec["seg"]] = rec
except FileNotFoundError:
pass
with open("segment_result.jsonl", "a", encoding="utf-8") as fh:
for idx, seg in enumerate(unique_segments, 1):
if seg in segment_map:
continue
result = locate(seg)
rec = {"seg": seg, **result}
fh.write(__import__("json").dumps(rec, ensure_ascii=False) + "\n")
fh.flush() # 立即落盘，中断不丢进度
segment_map[seg] = rec
print(f"[{idx}/{len(unique_segments)}] {seg} -> {result.get('prov', '')}{result.get('city', '')}")
time.sleep(1.0) # L0 档 1 QPS，逐条之间留 1 秒
```
写回表格：
```python
import csv
with open("numbers_in.csv", encoding="utf-8-sig", newline="") as fin, \
open("numbers_out.csv", "w", encoding="utf-8-sig", newline="") as fout:
reader = csv.DictReader(fin)
writer = csv.DictWriter(fout, fieldnames=reader.fieldnames + [
"prov", "city", "carrier", "carrier_type", "area_code", "post_code"])
writer.writeheader()
for row in reader:
seg = to_segment(row["phone"])
rec = segment_map.get(seg) or {}
row.update({
"prov": rec.get("prov", ""),
"city": rec.get("city", ""),
"carrier": rec.get("name", ""),
"carrier_type": rec.get("type", ""),
"area_code": rec.get("areaCode", ""),
"post_code": rec.get("postCode", ""),
})
writer.writerow(row)
```
Excel 文件用 `openpyxl` 或 `pandas.read_excel` 读进来、`to_excel` 写回去，逻辑与上面完全一致，只是换掉读写层。
## 并发节奏怎么定
免费档位的并发上限按等级划分，取自[免费接口与档位说明](https://www.showapi.com/free-api)：
| 等级 | 每日调用量 | 并发限制 | 所需积分 |
|------|-----------|----------|----------|
| L0 基础版 | 100 次/天 | 1 QPS | 免费 |
| L1 进阶版 | 500 次/天 | 5 QPS | 5,000 积分/年 |
| L2 高级版 | 2,000 次/天 | 10 QPS | 15,000 积分/年 |
| L3 专业版 | 10,000 次/天 | 30 QPS | 50,000 积分/年 |
| L4 旗舰版 | 50,000 次/天 | 50 QPS | 100,000 积分/年 |
L0 档并发是 1 QPS，脚本里逐条之间留 1 秒就是合适的节奏，开多线程只会拿到限流错误。档位提高后可以把 `time.sleep` 换成令牌桶，桶容量设为当前档位的 QPS 值。
每日调用量也要对一下。去重后号段数如果超过当日上限，把任务拆成多天跑，脚本的断点续跑机制能直接支撑这种做法。
## 号码列的常见清洗项
表里的号码列经常不干净，读进来之后先清一遍再取号段。
| 原始形态 | 处理 | 原因 |
|----------|------|------|
| `138 0013 8000` | 去掉空格 | 实测接口会清理空格，本地也一并清掉更稳 |
| `138-0013-8000` | 去掉连字符 | 实测该形态能正常查询 |
| `+8613800138000` | 剥离国际区号后再取前 7 位 | 带 `+` 的输入接口返回 `-2` |
| `１３８００１３８０００` | 全角转半角 | 全角数字接口返回 `-4` |
| 空单元格 | 跳过并计数 | 避免拼出空请求 |
| 首位为 0 | 跳过 | 不存在首位为 0 的号段 |
这些形态对应的错误码细节在[错误码逐条对照](https://www.showapi.com/guides/phone-attribution-error-codes-6)。
## FAQ
**Q1：接口有批量查询入口吗？**
没有。接入点 6-1 是单条同步请求-响应，没有订阅推送、没有回调。所谓批量是客户端循环调用实现的。
**Q2：几万个号码要调多少次接口？**
取决于不同号段的数量，不是号码总数。先取前 7 位去重，实测后 4 位不影响结果，所以同一个号段只需要查一次。
**Q3：能开多线程加速吗？**
取决于档位。L0 档并发限制是 1 QPS，多线程会触发限流。档位升到 L1 以上（5 QPS 起）再考虑并发，并且把并发数控制在档位 QPS 之内。
**Q4：脚本跑到一半中断了怎么办？**
结果按行落盘到 JSONL 文件，重新执行时会先读回已完成记录并跳过。这个设计不影响已消耗的额度，因为跳过的号段不会再发请求。
**Q5：查不到归属地的号码怎么处理？**
`ret_code` 为 `-1`、`-5`、`-6` 的号段查不到归属地，失败不扣点数。建议单独输出一份清单，标记待人工核验，不要混在成功结果里。
**Q6：Excel 里写回时区号和邮编为什么会变形？**
区号 `"010"` 和邮编 `"100000"` 带前导零，写进 Excel 时会被当成数字。输出列的单元格格式设为文本，或者写成 CSV 再用 Excel 打开，前导零能保住。
## 下一步阅读
- [手机归属地查询返回 -2、-3、-4 是什么意思？错误码逐条对照与排查](https://www.showapi.com/guides/phone-attribution-error-codes-6) —— 号码列清洗的依据
- [手机归属地查询按前 7 位号段做缓存：实测粒度与 Redis 实现](https://www.showapi.com/guides/phone-attribution-cache-6) —— 把去重结果沉淀成缓存
- [手机归属地查询返回字段全解：11 个字段的类型、取值与粒度](https://www.showapi.com/guides/phone-attribution-response-fields-6) —— 决定写回哪几列
- **本系列共 12 篇**：查看[手机归属地查询指南总目录](https://www.showapi.com/guides/phone-attribution-guides-6)
