☰
用SQLite实现手机号归属地解析:离线号段库与API兜底方案
2026/10/12 1:03:26 网站建设 项目流程

简介:这是一款基于JavaScript开发的手机号码归属地查询小工具,面向需要快速定位号码所在城市的营销人员、客服人员及个人用户,也适合前端或Node.js开发者作为练手项目。压缩包共53个文件,主体为40个dat号段数据文件,覆盖国内主要运营商号段;同时包含TypeScript源码、测试文件、构建脚本及package.json配置,整体仅163KB,轻量易部署。项目采用TypeScript编写核心逻辑,清晰展示号码前缀匹配、按号段分片存储的实现思路,数据文件可独立更新,lib目录提供编译后的JS版本,便于直接接入业务系统。此外还附带README、LICENSE和持续集成配置,涵盖从编码规范到构建发布的完整实践。当前已有962人学习下载,对于想了解数据查询工具开发与JavaScript工程化流程的读者有较好参考价值。

1. icity 是什么:手机号码城市解析到底在解决什么

icity 这类“获取手机号码所在城市的工具”,叫法很多,有人叫号码归属地服务,有人叫手机号城市解析,落到实现上就一句话:输入 11 位手机号,输出它所在的城市。电商预填收货地、活动页面判断用户地域、短信模板按城市做差异化投放,都依赖这个能力。很多团队起初会直接调付费 API,一旦查询量涨起来,成本、限流、时延全来了,于是又把目光转回本地号段库。这篇文章会从号码结构讲起,分析离线库与在线 API 的选型边界,给出一套用 SQLite 就能直接跑起来的最小方案,再把更新、兜底、缓存这些生产环境才遇到的事补齐,适合正准备自己动手做归属地解析的后端开发。

2. 号段原理与数据选型:icity 靠前 7 位定位城市的依据

2.1 号码结构:为什么主键是前 7 位而不是前 3 位

手机号是 11 位数字,不是随机分配的,从第 1 位到第 7 位携带了明确的语义信息。第 1 位固定是 1,第 2、3 位用来区分运营商大类,比如 13、15、18 开头是早期就分配的号段,17 和 19 开头的号段是后补的。第 4 到 7 位才是真正决定地市归属的部分,业内把它和前三们合在一起,统称“前 7 位号段”。

前 3 位只能区分到运营商,而且同一个前缀下跨省复用非常普遍;前 4 位能圈定一个较大的地理范围,但做不到稳定的地市级定位;前 7 位在绝大多数历史放号记录里,能唯一对应到某一个地级市。这就是icity这类工具统一把号码截取前 7 位去建索引的原因。下面这张结构表可以直观看到每一段位承担的责任:

号码位置位数含义粒度
第 1 位1移动通信标识固定为 1
第 2~3 位2运营商号段大类运营商级
第 4~7 位4地区编码与放号批次地市级
第 8~11 位4用户序列号无地域含义

这里要强调一个边界:前 7 位“绝大多数情况”唯一定位地市,不是绝对。历史上少数号段存在相邻地市共用的情况,比如某些新放号批次按省级池子分配,地市边界模糊;携号转网不影响归属地,但会让人误以为归属地变了。所以icity的定位必须收敛到“返回号码初始登记的城市”,而不是“用户当前所在的城市”,这两个概念在产品上要分清楚。

2.2 离线号段库 vs 在线 API:两种数据方式的成本边界

做号码城市解析,数据获取方式基本分成两条路:一条是维护一张本地号段表,查询时直接查本地库;另一条是调用现成的 API,把号码发出去等结果。两者不是二选一的对立关系,而是成本和时效的权衡。下面是我自己整理的一张对比表,做方案选型时可以拿它当评判框架:

对比维度离线号段库在线 API
定位粒度地市级地市级,部分能到基站级
查询时延亚毫秒级50~150ms 不等
并发上限取决于自身服务能力受套餐 QPS 限制
单次成本存储与带宽按次计费
数据更新自己维护,按月/按季服务商负责
运维成本需要管数据链路几乎为零

选型结论很直接:如果一天查询量只有几千次,对时延不敏感,直接接 API 是最省事的;如果日查询量到了几万、几十万,或者在线 API 的 QPS 卡住了活动页面,那就必须自建本地号段库。我自己经手的项目里,大多数是从 API 起步,等请求量上来了再切到本地库,这样前期的数据积累也能直接复用,后面细说。

2.3 数据更新是长期成本:号段增量从哪来

本地号段库最大的隐性成本不是查询,是维护。号段不是静态数据,新的放号批次会不断产生新号段,老号段也会因为运营商资源回收变得不再活跃。icity 这类工具如果只做一次性导入,三个月后覆盖率就会开始崩。

常见的号段数据来源有三个。第一,运营商公示的放号材料,通常按省份和地市批次公布,信息分散但权威;第二,专业数据服务商整理的号段文件,字段规整、更新及时,是需要付费的;第三,把自己早期调用 API 的查询结果沉淀下来,回写进本地库。第三条路径最稳定,因为它能精准反映你业务里真实出现的号码分布,但需要时间积累。

我的建议是:基础号段表每 3 到 6 个月更新一次,出现大规模新号段投放时按季度核对。号段库本质上是一张快照,导入完成的那一刻就可能已经过期。所以从设计第一天起就要给每条记录加上数据版本号和时间戳,否则后面做增量更新时根本不知道哪些数据是新哪些是旧。

3. 用 SQLite 自建本地号段库:icity 的最小可复现实现

3.1 初始化号段库:建表与导入完整命令

有了号段数据文件,第一步是把数据洗干净放进 SQLite。我会建一张 phone_segment 表,主键直接用前 7 位的文本类型,而不是整数。原因有两个:一是号段前导数字如果是 0 开头的固话号段用整数会丢精度,二是 TEXT 类型在 SQLite 里做等值查询时同样走索引,性能没有损失。SQL 如下:

CREATE TABLE IF NOT EXISTS phone_segment ( prefix TEXT PRIMARY KEY, province TEXT NOT NULL, city TEXT NOT NULL, carrier TEXT, source TEXT DEFAULT 'import', updated_at TEXT DEFAULT (date('now')) ); CREATE INDEX IF NOT EXISTS idx_phone_segment_city ON phone_segment(city);

这个索引不是为了等值查询,是为了后续做“按城市反查号段”的运营分析,比如想统计某个城市的号段覆盖量时直接走索引。真正查询用的是主键 prefix,SQLite 对主键等值查询走的是 B 树定位,不会全表扫。

数据文件拿到后,一般是个 CSV,列顺序建议和表结构对齐:prefix,province,city,carrier。导入命令用 sqlite3 自带的 .import:

sqlite3 city.db <<'SQL' .mode csv .import phone_segment.csv phone_segment SQL

有一点要留意:.import 不会跳过表头,如果 CSV 第一行是列名,导入前先删掉或者用 sed 剔除,否则会把一行“prefix”字符串插进表里。导入完成后跑一句SELECT COUNT(*) FROM phone_segment;,确认行数和源文件一致再继续,这里翻车的人不少。

3.2 查询函数:从手机号到城市的核心代码

表建好后,查询逻辑其实非常薄。核心是三步:清洗号码、截取前 7 位、查表。我会把清洗和查询分开成两个函数,因为输入清洗在真实场景里远比想象中复杂:

import re import sqlite3 DB_PATH = "city.db" _local = { "version": None, "cache": {} } def normalize_phone(raw: str) -> str: digits = re.sub(r"[\s\-\(\)]", "", raw or "") if digits.startswith("+86"): digits = digits[3:] elif digits.startswith("86") and len(digits) == 13: digits = digits[2:] if len(digits) == 11 and digits.startswith("1"): return digits return "" def lookup(phone: str): p = normalize_phone(phone) if not p: return {"code": 400, "msg": "invalid phone"} prefix = p[:7] cached = _local["cache"].get(prefix) if cached is not None: return {"code": 0, "data": cached} conn = sqlite3.connect(DB_PATH) row = conn.execute( "SELECT province, city, carrier FROM phone_segment WHERE prefix = ?", (prefix,) ).fetchone() conn.close() if row is None: return {"code": 404, "msg": "segment not found"} info = {"province": row[0], "city": row[1], "carrier": row[2]} _local["cache"][prefix] = info return {"code": 0, "data": info}

normalize_phone 里做的清洗和格式兼容非常关键。用户从通讯录复制出的号码经常带“+86”前缀,或者中间被人为加了空格和短横线;直接截前 7 位会把“+86”这三位算进去,得到完全错误的结果。这段处理完,非法号码直接返回 400,不会把脏数据带进查询。

进程内缓存字典 _local 也很重要。同一个前 7 位号段被反复查询是常态,缓存能省掉绝大多数 SQLite 查询。不过要注意,这个缓存只是最简单的 dict,生产环境要加上限和淘汰策略,3.3 节给出配置办法。

3.3 性能参数调节:索引、内存缓存与并发读

SQLite 默认配置偏向保守,直接拿来扛并发读会吃亏。我会在连接建立时执行三个 PRAGMA,分别是开启 WAL、降低同步级别、打开内存映射。它们的组合效果是:读写不互相阻塞、减少磁盘 fsync 次数、让大表的一部分直接映射到内存地址空间。

PRAGMA 配置推荐值作用
journal_modeWAL读不阻塞写,写不阻塞读
synchronousNORMALWAL 模式下降低提交成本
mmap_size268435456(256MB)对表的扫描和查找走内存映射

多线程场景下,我不建议每个查询都新建连接,也不建议所有线程共用一个连接。常见做法是用 threading.local 给每个线程绑一个独立连接,省去重复建连的开销,也避免连接对象在线程间乱跑:

import sqlite3 import threading _db_local = threading.local() def get_conn(): if not hasattr(_db_local, "conn"): c = sqlite3.connect(DB_PATH, check_same_thread=False) c.execute("PRAGMA journal_mode=WAL") c.execute("PRAGMA synchronous=NORMAL") c.execute("PRAGMA mmap_size=268435456") _db_local.conn = c return _db_local.conn

进程内缓存这一层,我会把 3.2 节的普通 dict 换成有上限的 LRU。做法是给缓存记录加时间戳,或者在插入时判断字典长度,超出阈值就把最旧的一半清掉。缓存上限一般设在 3000 到 5000 条就够了,因为业务里活跃号段通常只有几千个,留太多只是白白占内存。这里有一个血泪教训:缓存清理策略一定要做,否则跑一个月后 Python 进程的内存会涨到让你怀疑自己的代码有泄漏。

4. 在线 API 兜底与级联查询:icity 的生产级混合方案

4.1 API 接入的通用请求与参数说明

本地号段库覆盖率再高,也总有查不到的号段。所以生产方案里,API 不是替代品,而是兜底。API 的接入各家服务商接口路径和返回字段不一样,但核心参数就那几个:手机号、鉴权 key、超时时间。封装成一个通用函数,内部把差异挡住:

import requests def api_lookup(phone: str, timeout: int = 3): p = normalize_phone(phone) if not p: return None resp = requests.get( API_BASE_URL + "/phone/lookup", params={ "mobile": p, "key": API_KEY }, timeout=timeout ) if resp.status_code != 200: return None data = resp.json() if data.get("status") != "ok": return None # 不同服务商字段可能是 province/city,也可能是 loc,这里统一转换 return { "province": data.get("province") or data.get("loc", ""), "city": data.get("city") or "", "carrier": data.get("carrier", ""), "fresh": True }

这段代码有几个参数值得注意。timeout 设成 3 秒是平衡值,太短在弱网环境容易误伤,太长会拖慢整个查询链路。API_KEY 不要暴露在前端调用里,icity 这种工具应该部署在服务端。字段兼容那两行很有必要,市面上的 API 返回结构并不统一,花半小时做一层薄薄的适配层,后面切服务商就不用改业务代码。

4.2 本地优先、API 兜底的级联查询流程

有了本地库和 API,就要决定两者的先后顺序。我的习惯是“本地优先,API 兜底,旧数据最后保底”。具体流程是:先查本地号段表,命中就直接返回;查不到时走 API,API 成功就把结果回写本地缓存;如果本地有一条旧数据而 API 失败,宁可返回这条旧数据也不要让调用方一无所获。

def resolve(phone: str): local = try_local(phone) if local is not None and not local.get("stale"): return local api = api_lookup(phone) if api is not None: update_local_cache(phone[:7], api) return api if local is not None: return {**local, "stale": True} return {"province": "", "city": "", "unknown": True, "code": 404}

这里“stale”标记是核心设计。本地数据不会自动变旧,但号段库有数据版本概念,我会在服务启动时记录号段文件的更新时间,如果超过 6 个月,就把这个标记赋值给所有本地查询结果。外部调用方看到 stale 就能判断:这个城市信息可信,但时效性不足,适合做低风险场景的降级。

回写本地缓存这个动作别小看,它能让高频新号段在第一次 API 查询后就不再重复走网络链路,长期运行下来 API 调用量会越来越低,成本也随之下降。

4.3 超时、失败重试与结果缓存的三个配置

生产环境里最坑的不是查询逻辑,是依赖 API 时的网络抖动。我会把超时和重试的参数集中配置,让行为可预测:

配置项推荐值说明
connect_timeout2 秒连接超时,不包含服务端处理时间
read_timeout3 秒响应体读取超时
重试次数2 次超过即放弃本次查询
退避策略1s、2s 递增重试间隔指数递增,避免雪崩
结果缓存 TTL24 小时控制 API 结果对后续查询的影响窗口

重试不是无限重试,网络上对同一个号码反复请求没有意义,两次就够。缓存 TTL 这块,我会用带过期时间的 LRU,实现比普通 dict 多一步时间判断:

from collections import OrderedDict import time _api_cache = OrderedDict() API_TTL = 86400 def get_api_with_ttl(prefix: str): item = _api_cache.get(prefix) if item is None: return None value, ts = item if time.time() - ts > API_TTL: del _api_cache[prefix] return None return value

缓存 key 不要只放手机号,放前 7 位号段就够了,号码后 4 位对归属地没有任何影响,放全号只会浪费内存。TTL 到了之后,同一个号段会自然触发下一次 API 查询,服务商那边数据有更新也能被感知到。

5. icity 落地避坑:携转、虚商、新号段的 5 个典型坑

5.1 携号转网:归属地没变,运营商变了

现象:本地库返回的运营商和用户实际使用的运营商对不上。比如一个号码本地表里记录的是 A 运营商,但用户已经携号转到了 B 运营商,接投诉说是查询不准。

原因:号段表记录的是号码首次放号时的归属运营商,携号转网不会改号码,也不会改归属地,它只改变这个号码由哪家运营商提供当前服务。归属地查询工具如果硬要连运营商一起返回,必然翻车。

解决:icity 这个场景只承诺城市维度,运营商字段如果保留,就明确标注为init_carrier(初始运营商),在文档里写清楚不反映当前服务商。这样既避免了误导,又保住了号码初始归属信息在风控场景里的参考价值。

5.2 虚商与物联网卡号段:不在常规号段表里

现象:查询 170、171、162、165、167 等号段时,本地库要么返回空,要么返回一个明显错误的默认城市。

原因:虚拟运营商和物联网卡用的是单独分配的号段池,很多号段文件压根不收录,或者收录了但只到省级粒度。用普通号段表去查这些号码,基本是盲区。

解决:维护一张小的“特殊号段”集合,命中后不打常规查询逻辑,直接标记为虚拟运营商或物联网卡,根据业务需要返回省级信息或走 API。重点是不要尝试给这些号段猜测城市,猜错的成本比返回未知高得多。

5.3 新号段放号:前 7 位查不到时的默认策略

现象:上级要求统计新上线号段覆盖的城市,本地库查不到;新号段用户在活动页面上看到默认城市是错的。

原因:新号段投入商用后,本地号段表没有同步更新。号段库是快照,放号公告和实际商用之间存在时间差,这个窗口期靠本地库永远覆盖不到。

解决:查询未命中时走级联逻辑去查 API,成功后回写本地库。季度更新时,把线上查询日志里所有命中 API 的号段整理出来,合并进下一次号段文件。这样即使不买新数据,工具也会在运行中自己长出覆盖能力。

5.4 输入格式五花八门:范围过滤不能省

现象:用户从通讯录复制的号码带着“+86”、空格、短横线,直接截前 7 位得到“8613800”这种错误结果,查出来一个不存在的城市。

原因:查询代码没有做输入归一化。很多刚上手的方案只在页面端做了简单的 11 位校验,后端没有二次清洗,脏数据就进来了。

解决:后端必须有一道 normalize,先去掉所有非数字字符,再处理“+86”和“86”前缀,最后用len(digits) == 11 and digits.startswith("1")做门槛过滤。港台地区和国际号码也在这个环节挡掉,因为这些号码的前 7 位不在大陆号段语义里。

5.5 命中率突然下降:先查数据文件,别怀疑代码

现象:线上命中率从 97% 掉到 80%,没有发过新版本,代码逻辑也没动过。

原因:最常见的是数据文件出了问题,比如 .import 导入时中途失败造成部分数据缺失、数据文件被纳入了代码仓库走了 LFS 截断、定时任务覆盖了号段文件。

解决:先跑一个自检,看表行数和文件行数是否一致,再执行PRAGMA integrity_check;看完整性。为了防止这类问题再次出现,导入操作不要直接写业务表,先导入 staging 表,校验通过后再用事务整体替换目标表。这个习惯能挡掉一半以上的数据事故。

6. 数据质量检验与增量更新:icity 的进阶收尾技巧

6.1 全量抽样自检:覆盖率与准确率的量化方法

本地号段库有没有变准,不能用感觉衡量。我会从线上脱敏后的号码日志里随机抽 1 万条,用本地库查一遍,再调 API 查一遍,把两个结果做对比,输出总准确率和各号段覆盖率:

def sample_check(sample_phones): hit_local = 0 same = 0 missing = [] for phone in sample_phones: local = lookup(phone) api = api_lookup(phone) if local is None or "city" not in local: missing.append(phone) continue hit_local += 1 if local["city"] == api["city"]: same += 1 return { "coverage": hit_local / len(sample_phones), "accuracy": same / max(hit_local, 1) }

统计口径上,省一致但城市不一致的记半对,城市一致才算全对。把覆盖率低于 90% 的号段列出来,这些就是下一次更新要优先补的数据。

6.2 增量更新脚本:按月合并新号段的标准流程

新号段进来后,我会先在 staging 表里准备好数据,校验数量无误后,用一个事务完成合并并清空进程内缓存:

BEGIN; INSERT OR REPLACE INTO phone_segment SELECT prefix, province, city, carrier, 'monthly_update', date('now') FROM staging; COMMIT;

注意,这里用的是 OR REPLACE 而不是 OR IGNORE,因为同一前缀在旧文件里可能城市信息已经变了,要以新文件为准。更新完成后,把内存里的缓存整个清掉,避免旧数据继续被命中。

最后说一个我自己的教训:有一次我图省事把缓存 TTL 调成了 7 天,结果运营换了一波城市投放策略,两个小时里大量请求还在拿缓存里的旧城市。排查到最后才发现不是查询代码的问题,是缓存版本没跟上。后来我把号段文件的版本号拼进缓存 key,每次更新号段库就全局换一个 key,旧缓存自动失效。这个习惯一直留到现在。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询