批量查询号码归属地：从数据源选型到脚本落地的完整方案

简介：批量查询号码归属地是市场营销、客户服务与数据分析中的常见需求，尤其在理解用户分布、优化服务策略时需要快速准确地匹配运营商与地区信息。这里提供的工具包整合了覆盖电信199号段在内的43万条最新号码归属地数据，兼顾新鲜度与全面性。压缩包共4个文件，约6.92MB，内含json数据文件、txt使用说明、xlsx表格样例和可直接运行的exe查询工具。exe工具支持单个即时查询和批量文件上传，批量结果可导出为Excel，便于进一步清洗与市场分析。目前已有1827人学习使用。除现成查询工具外，资源还展示了从数据库管理、文件处理到异步任务调度、Web接口设计的完整技术思路，适合希望提升综合开发能力的开发者参考实践。

1. 批量查询号码归属地：先想清楚要查哪一个“归属地”

把一批手机号丢进脚本，批量查询号码归属地，几秒后拿回省份、城市和运营商——这听起来像个小工具，但真正去做的同学往往卡在三个地方：数据源该选在线接口还是离线号段库、批量请求会不会被对方限流、以及携号转网以后查回来的“归属地”到底还准不准。这篇文章不讨论玄学，只讲一套我自己在维护的批量查询方案，覆盖从单号查询到万级号码调度的完整落地方案，给运营、开发或对接供应商的同学作参考。

在往下读之前，你只需要有一个最小认知：手机号前三位指向运营商，前七位在绝大多数情况下指向发卡省份和城市。准确的归属地数据，就分布在这些号段前缀的映射关系里。有了这个认知，后面的选型和脚本才不会变成黑匣子操作。

2. 数据源选型：在线接口、离线库和自建之间的取舍

手机号码归属地的本质是“号段映射”。一个号码的前三位决定了它的运营商前缀，前七位能锁定放号省份和城市。运营商每隔一段时间会释放一批新号段，所以任何一份归属地数据都只是某个时间点的快照。理解这一点，选型思路就会很清晰：你需要的是一个实时查询通道，还是一份能反复扫描的本地数据。

2.1 三种数据源都长什么样

常见做法可以分成三类：在线接口、离线号段库、自建混合库。在线接口最直观，把号码作为参数发给某个开放平台，平台返回JSON，返回内容一般长这样：

{ "code": "0", "data": { "province": "浙江", "city": "杭州", "carrier": "中国移动", "area_code": "0571" } }

在线接口的好处是数据新鲜、能拿到运营商字段，缺点是每一笔都要付钱，而且受QPS限制。对只有几百个号码且要实时结果的场景，这是最合适的。如果批量任务里混着大量重复号码，把全部号码都压给在线接口，相当于把钱扔在水里。

离线号段库则是一个静态文件，通常按手机号前7位建立映射，比如1380571映射到“浙江杭州”。文件格式可以是CSV、SQLite或二进制索引。它的特点是百万甚至千万级号码也能在内存里跑完，因为查询只是在字典里做一次前缀查找，没有网络IO。但离线库的时效性取决于更新频率，新号段放号后如果不更新，查出来就是“未知”。

自建库是在前两者之上再包一层自己的存储：把在线接口的结果沉淀到本地数据库，定期用离线库补全，更新自己的索引。批量查询时优先查本地，本地没有再走在线接口，回填后给下一次复用。常见做法是在内部系统里用一个表或哈希结构保存已经查过的号段结果，避免重复消耗额度。

三种方式可以放在一起对比：

补充一点：所谓“离线号段库”不是网上随便找个归属地包就行。它的来源必须是合法采购或有明确授权协议的数据文件。拿到的文件先看字段说明，有些只给省市区，有些附带运营商和长途区号，还有些会单独提供“是否携号转网”的标记。字段越全，后面判断越不容易踩坑。

2.2 我怎么选：存量清扫选离线，实时校验选在线

如果你手头是十万个历史号码要做城市分群，用在线接口一万一万地跑，大概率会被限流，额度也烧得心疼。我一般会先采购一份离线号段库，把十万个号码本地跑一遍，5分钟内出结果。这里说的采购不是从网上抓数据，而是向数据服务商购买合法的号段文件。签约时一定要问清楚数据使用期限，以及能不能用于内部系统二次加工，很多合作协议对存储、分发和二次处理都有明确限制。

反过来，如果你的业务是用户注册时实时校验归属地，比如判断这个号码属于哪个分拣中心，那就必须用在线接口。离线库做不到实时更新，新放出的号段和刚转网的用户都查不准。

无论选哪一条路，都要先做一轮数据校验：拿一批已知归属的号码对照查询结果，统计准确率。不要拿一个号段库直接上生产，曾经有同事拿了一份过期两年的库去切分短信通道，结果把一批新号段全归成“未知城市”，这就是黑匣子没打开过的后果。

补一个实用的选型判断公式：需要查询的号码总数乘接口单价，对比离线库采购价。当总量超过接口预算数倍时，离线库更划算。同时考虑系统对实时性的要求，只跑一次的数据任务完全可以用离线库，不需要在线接口。

我通常在项目里先建一个“离线优先、在线补漏”的混合结构：本地库命中就用本地结果；本地缺失时，把号码排队去请求在线接口，回填后入库。这样既控制了成本，也保证新号段至少有个实时结果兜底。

2.3 号段库的文件格式、索引与版本管理

离线号段库在工程上要准备一个meta表，记录当前数据的来源、发布时间、覆盖号段数量、生效日期。批量脚本启动时先读这个表，如果发现数据超过30天未更新，就把离线查询降级成“参考数据”，同时把未命中率上报到日志。这个习惯帮我避免过好几次“老库害死人”的事故。

文件格式方面，最常见的是CSV，Python直接读取后写入SQLite即可。字段顺序建议固定为：

SQLite索引直接用prefix作为主键，不需要额外加索引，因为查询就是按主键点查。导入时可以将表建成WITHOUT ROWID模式，能少一层B树查找，但对百万级数据来说提升有限。真正有用的是让数据库连接使用只读模式，并在内存中放大Cache。查询瓶颈不在SQLite，而在把号码转成前缀的那一步，所以批量脚本里要优先复用同一个连接，同时把号码标准化逻辑收敛到函数入口，避免每个线程各写一套。

版本管理还有一个容易忽略的环节：在线接口回填到本地后，要能区分“本地号段”和“接口回填”。我习惯在表里增加一个data_source字段，取值为base或online。以后排查数据错误时，可以直接按data_source=online筛出回填数据，逐个核对接口返回是否合理。后面的脚本会用到这个字段。

3. 核心脚本：从单号查询到批量轮询的落地写法

3.1 先写一个离线查询函数

我习惯用SQLite做离线号段库，原因有三个：Python内置支持、单文件便于分发、索引查询足够快。号段文件导入SQLite的表结构如下：

CREATE TABLE IF NOT EXISTS phone_segments ( prefix TEXT PRIMARY KEY, province TEXT, city TEXT, carrier TEXT, data_source TEXT DEFAULT 'base' );

这比上一章的字段设计多了data_source，用来标记数据来自离线库还是在线接口回填。导入完成后，查询一个手机号归属地的函数可以写成这样：

import sqlite3 DB_PATH = "phone_segments.db" def lookup_offline(phone: str, conn: sqlite3.Connection) -> dict | None: # 先清理号码，只保留数字 phone = phone.strip() if phone.startswith("+86"): phone = phone[3:] if len(phone) != 11 or not phone.isdigit(): return None prefix = phone[:7] row = conn.execute( "SELECT province, city, carrier, data_source FROM phone_segments WHERE prefix = ?", (prefix,) ).fetchone() if row is None: return None return { "province": row[0], "city": row[1], "carrier": row[2], "source": "offline" if row[3] == "base" else "online" }

逻辑说明：把手机号规范化以后截取前7位号段，去SQLite里按主键查一次。这一步不走网络，常驻内存式连接每秒能跑几十万次。参数上需要关注两点：

• phone参数：建议调用方只传11位数字，但函数内部仍然做一层防御。因为Excel里常见的号码是“+86 138 0571 1234”这种格式，如果前置清理交给SQL，可读性和排查会差很多。
• DB_PATH：根据你存放数据库文件的位置配置。连接建议用check_same_thread=False，否则后面配合多线程会报错。

实际项目里我一般不用裸sqlite3，而是把库连接封装在模块里，或者在整个批量任务里只创建一个连接传给多个worker。但最小脚本这样写已经够用。

3.2 再加一个在线查询封装

离线库查不到时，需要在线接口兜底。在线接口的封装要同时处理HTTP层和业务层，一个最小版本如下：

import requests API_URL = "https://open.example.com/api/phone/locate" def lookup_online(phone: str, api_key: str, timeout: float = 3.0) -> dict | None: payload = { "phone": phone, "key": api_key, "format": "json" } try: resp = requests.get(API_URL, params=payload, timeout=timeout) except requests.Timeout: # 超时不要立刻重试，交给上层的重试队列控制 return None if resp.status_code != 200: return None data = resp.json() if data.get("code") != "0": return None result = data.get("data") or {} return { "province": result.get("province"), "city": result.get("city"), "carrier": result.get("carrier"), "source": "online" }

逻辑说明：这里把业务码和数据字段都封装好，外层只关心拿没拿到结果。要注意返回码和字段名的映射，不同平台的返回结构完全不一样，对接前先打印3到5个样本。

参数说明：

• timeout：默认3秒，网络抖动时宁可失败让整个队列重试，也不要卡住线程。
• api_key：不要写死在代码里，批量脚本建议从环境变量读取。
• 每一笔在线查询都要记录耗时和结果状态，方便月末核账和排查限流。

3.3 把单号查询拼成批量流程

批量流程分成三步：读入号码清单 → 逐号查离线库 → 未命中收集到在线队列。这样可以先覆盖90%以上的号码，在线请求量降到最低。

def batch_lookup(path: str, offline_conn: sqlite3.Connection, api_key: str): from collections import deque online_queue = deque() results = [] with open(path, "r", encoding="utf-8") as f: for line in f: phone = line.strip() if not phone: continue r = lookup_offline(phone, offline_conn) if r: # 命中本地，直接记录 results.append((phone, r["province"], r["city"], r["carrier"], r["source"])) else: online_queue.append(phone) for phone in online_queue: r = lookup_online(phone, api_key) if r: results.append((phone, r["province"], r["city"], r["carrier"], r["source"])) else: results.append((phone, "", "", "", "unknown")) return results

这里故意用deque而不是list，是为了号码量大了以后，如果要断点续跑，队列可以更容易转成文件记录。真正的工程中不会把所有号码都丢进内存，而是分批落盘。批量流程的一个关键参数是每批数量，常见做法是1000条一批，写一个独立的flush动作。原因有两个：一是防止内存占用失控，二是便于中途kill后从最后一批继续。

如果复现时发现大量号码跑到在线队列，先检查是否没做号码标准化。lookup_offline内部能处理+86前缀，但对带横杠和空格的号码并不友好，第5章会给出一个更完整的清洗函数。

4. 并发与调度：把一万个号码跑完需要哪些参数

4.1 并发模型：本地查询和在线接口要分开看

在讲并发参数前，先把模型理清楚。离线库查询拼的是CPU与IO，适合开多线程；在线接口拼的是平台QPS，你开多少线程都没用，只会触发限流。所以正确做法是：本地离线用线程池，在线请求用固定速率。

我先给一个通用任务拆分。把号码清单拆成每批1000条，每批内部先本地查，再收集在线缺失项。这样每批就是一个独立任务，可以交给线程池执行。

from concurrent.futures import ThreadPoolExecutor def process_batch(phone_list, offline_conn, api_key): results = [] for phone in phone_list: r = lookup_offline(phone, offline_conn) if r: results.append((phone, r["province"], r["city"], r["carrier"], r["source"])) else: r = lookup_online(phone, api_key) if r: results.append((phone, r["province"], r["city"], r["carrier"], r["source"])) else: results.append((phone, "", "", "", "unknown")) return results def run_batches(phones, offline_conn, api_key, workers=4): batches = [phones[i:i + 1000] for i in range(0, len(phones), 1000)] with ThreadPoolExecutor(max_workers=workers) as pool: futures = [pool.submit(process_batch, batch, offline_conn, api_key) for batch in batches] for future in futures: yield future.result()

这个版本有个明显的问题：在线接口会被4个线程同时打，容易触发限流。正确的习惯是把在线查询单独限速，或者在process_batch中对在线请求加锁。后面讲限流器时会解决这一点。

4.2 三个必调参数：qps、超时、重试

在线接口调用必须显式配置三个参数：qps、timeout、retry。随意设置的后果不是跑得慢就是被封。我常用的初始值如下：

用Python内置的time.sleep实现qps限制最直接。如果要求更高，可以用令牌桶。令牌桶的好处是允许短突发，比如某1秒内发了6笔，但下一秒补回来。不过对于批量任务，固定sleep已经够用。

import threading import time class RateLimiter: def __init__(self, qps: float): self.interval = 1.0 / qps self._lock = threading.Lock() self._next_call = time.time() def wait(self): with self._lock: now = time.time() if now < self._next_call: time.sleep(self._next_call - now) self._next_call = max(time.time(), self._next_call) + self.interval

逻辑说明：每个线程在发起在线请求前调用limiter.wait()，保证全进程的请求速率不超过设定qps。注意线程锁不能省略，否则多线程下两个worker可能同时通过检查。

参数细节：

• qps从1开始调，观察返回码。如果频繁出现限流码，把qps往下调，而不是往上加。
• retry并不是越多越好。离线任务可以容忍失败，在线任务重试次数多了会拖慢整批，甚至把对方系统打到告警。
• timeout与qps配合：qps=5时，每个线程每200ms发一条，如果timeout=10秒，线程数必须小于等于qps×timeout才不会堆积。

提示：qps并不是越高越好，对接前先查接口文档里的频控说明，把初始值设低，再慢慢往上调。

4.3 断点续跑与结果落盘

万级任务的主要风险是跑到一半进程死掉。死在哪、下一次从哪继续，这是调度里必须处理的事。

我的习惯是输出JSONL而不是CSV，理由是JSONL能边跑边写，某一行坏了不影响前面的数据。如果要CSV，就每查完一批重写一次。

import json FIELD_NAMES = ["phone", "province", "city", "carrier", "source"] def append_result(path: str, records: list) -> None: with open(path, "a", encoding="utf-8") as f: for record in records: f.write(json.dumps(dict(zip(FIELD_NAMES, record)), ensure_ascii=False) + "\n")

同时记录一个进度文件，每完成一批就写当前偏移量。重启时从偏移量接着读，这样在线查询的额度不会浪费在已经查完的号码上。程序正常结束后，把临时JSONL转成CSV，再做一次去重。

如果连进度都不想管理，还有更省事的方案：结果表提前建好唯一约束，主键就是phone。每条结果插入用INSERT OR REPLACE。这样即使任务重复跑，也只是覆盖旧结果，不会污染数据。这个做法在小规模批量场景里非常省心，但到了十万级以上，插入开销会变大，所以常见做法是两个结合：批量用临时JSONL，结束统一入数据库。

5. 避坑指南：携号转网、虚拟号段与返回码的五处翻车

5.1 携号转网后，离线库查出来的运营商不可信

现象：一个138开头的号码，离线库返回“中国移动”，但用户实际已经转到联通。用这个结果做短信通道会翻车。

原因：携号转网放开后，号段前缀只代表初始放号运营商，不代表当前归属。离线号段库普遍只记录初始运营商，没有实时转网标记。

解决：需要向数据服务商确认是否提供转网后的数据；如果没有，就在结果里去掉“运营商”字段，或者标注为“初始运营商”。归属地省市区通常不随转网变化，可以放心使用。

补充一点：即使转网后省市区不变，在线接口的实时运营商结果也可能与用户实际持卡不符。想验证自己手里数据源的准确性，可以拿一百个号码做人工抽样核对。虽然费时间，但能让你对数据源的真实情况心里有数。

5.2 虚拟运营商号段查不出城市

现象：170、171、162、165、167开头的号码，离线库大部分查不到，在线接口偶尔返回“未知”。

原因：虚拟运营商租用基础运营商网络，号段没有自己的归属地；不同平台对虚拟号段的处理逻辑不一致，有的直接返回空。

解决：在批量流程里单独处理。用前3位判断，落在虚拟号段时，结果写“虚拟运营商”，省份和城市留空。千万不要因为查不到就标记为“无效号码”，这会误删用户。

还要注意物联网卡和特殊号段。106开头的服务号、400开头的固话业务号，以及当前运营商放出的13、14、15、16、17、18、19开头的号段规则都不太一样。批量任务里如果混进了这类号码，先过滤掉，再进入归属地查询链路，能省下不少在线额度。

5.3 在线接口报错被误判成“无归属地”

现象：批量跑的时候，某一段号码全部返回source=unknown，但业务上这些号码是真实在用的。

原因：最常见是触发了QPS限制，平台返回429，脚本把非200都当成失败，最终把“查询失败”和“查无此号”混在一起。

解决：在结果里至少要分三态：SUCCESS、FAILED、UNKNOWN。UNKNOWN表示接口明确查无数据；FAILED包括超时、5xx、429。下游统计时不要把FAILED当成UNKNOWN。这类问题在批量任务里占大头。

状态码设计可以像HTTP那样自解释，把FAILED再细分出LIMITED和TIMEOUT。这样日志一旦报警，直接看状态码就知道是限流还是网络问题，不需要再翻原始响应去猜。下游对接的同学也不用反复问你“unknown到底代表什么”。

5.4 号码格式不统一

现象：Excel里“+86 138-0571-1234”在批处理时丢失了前7位，全部落到在线队列，白白消耗额度。

原因：没有做号码标准化，直接把整行字符串当phone传递。

解决：在查询之前统一清洗。示例：

import re def normalize_phone(raw: str) -> str | None: if not raw: return None phone = re.sub(r"[^\d]", "", raw) if phone.startswith("86") and len(phone) == 13: phone = phone[2:] if len(phone) == 11 and phone[0] == "1": return phone return None

处理完以后，把无法识别的行单独落一个文件，不参与查询。这一步看起来琐碎，却是很多批量任务翻车的起点。

如果直接从Excel导出数据，还要提防科学计数法。手机号在Excel里经常变成1.38E+10，导出CSV时应该选文本格式，或者用\t分隔的tsv文件，避免长号码被截断。反正我的习惯是，任何外部数据进来都先过一遍normalize_phone，再决定下一步。

5.5 离线库数据陈旧

现象：某个新号段（比如199开头）上线几个月，离线库查出来是空的，但在线接口能查到。

原因：离线号段库文件是采购时的版本，不是一个实时更新的服务。

解决：记录号段库的版本与更新时间，在脚本里做一个检查：如果库的更新时间超过30天，就触发在线补缺队列。让在线接口专门消化新号段，回填后更新本地缓存。

如果要自动更新，不要一上来就全量覆盖。先拿小样本做pacing check，把新旧库的差异统计出来，确认覆盖率和准确率达标后再切换。回填数据写入时也要保持data_source字段，这样后面发现某个新号段被错误归属，还能追踪到是哪个接口返回的。

翻车记录不只这五类，但把这五类处理好，批量查询的准确率基本能回到一个可接受的范围。排查时记住一个原则：每一次查询都要留下状态码和来源字段，不然出了问题很难对照日志还原现场。

6. 进阶技巧：用缓存与增量更新把查询成本压到最低

6.1 用Redis缓存在线查询结果

对于反复出现号码的业务场景，同一个号码的在线查询结果没必要每次都花钱。正确做法是在在线接口前面加一层Redis。

import json import redis CACHE_KEY_PREFIX = "phone_locate_v1:" def lookup_with_cache(phone: str, api_key: str, r: redis.Redis, ttl: int = 86400) -> dict | None: cache_key = CACHE_KEY_PREFIX + phone cached = r.get(cache_key) if cached: data = json.loads(cached) data["source"] = "cache" return data result = lookup_online(phone, api_key) if result is None: return None r.setex(cache_key, ttl, json.dumps(result, ensure_ascii=False)) return result

这里把TTL设为86400秒，也就是一天。注意不要设永久缓存，因为号码转网后归属地可能变化。当天结果至少能支撑一天内的重复查询。

6.2 增量更新机制

离线号段库每季度或每月更新一次。更新动作可以做成增量。做法：从数据服务商那里收到新号段文件后，先入库，再用版本号标记当前生效版本。查询时读到的数据携带版本，如果发现版本过低，就在后台触发在线接口补缺。这样批量任务的准确率会快速回暖。如果要把这套方案接到公司内部系统，我建议把查询结果再写进一个宽表，记录查询时间、命中来源、是否转网等字段。运维同学想排查时，不用去翻日志。

6.3 验证与成本压测

做了缓存和增量后，不要只看功能正常，还要验证两点。第一是准确率：拿一百个真实号码，比对离线库、在线接口、缓存三者的结果，逐个人工核对省市区，发现不一致再调优先级。第二是成本：把在线接口调用量和缓存命中率打点，监控指标可以设在缓存命中率90%以上、离线库命中率95%以上，低于阈值就发告警。

有一次我在内部系统加了缓存，命中率从零涨到80%之后出现了一个奇怪现象：有几个号码在缓存里存了“未知”，后来数据库更新了能查到归属地，但缓存没过期，查询结果一直是“未知”。这就是我没把查询失败也缓存进去造成的。后来改成：只有成功的结果才写Redis，失败结果一律不缓存。这个坑不是很大，但正好说明缓存也要分状态处理。

批量查询号码归属地这个方向，做到后面已经不是“怎么查”的问题，而是“查多少次、信哪份数据、怎么低成本地维持准确”。把离线库、在线接口、Redis三级结构搭好，再配一个清晰的状态标记，大部分花式需求都能接住。希望帮到你。

本文还有配套的精品资源，点击获取