批量查询号码归属地：Python与SQLite离线号段库实战

简介在市场营销、客户服务与数据分析场景中批量识别电话号码归属地是一项高频需求。这份资源提供一套覆盖电信199号段等最新号码段的离线查询方案内置总计43万条归属地数据适合需要本地化、高效率处理海量号码的开发者和业务人员。压缩包共4个文件大小仅6.92MB包含JSON数据源、可直接运行的exe查询工具、Excel表格及使用说明。使用时无需搭建复杂环境既可单号实时查询也可导入号码文件批量处理并输出Excel结果方便后续分析。已有1820人学习下载。相比在线接口该工具不受网络限制数据更新及时既可用于企业客户数据清洗也可作为学习号码归属地匹配逻辑、数据库查询与文件处理技术的实用参考兼顾业务价值与学习价值。 先聊个真实场景。前几年我做用户运营手里攒了一批几万条的用户名单要做区域化运营策略第一步就得先知道这些号码都分布在哪些省市。一开始我直接用网页挨个查查了不到一百条就崩溃了——复制、粘贴、回车、记结果机械动作重复几百次之后脑子基本是木的而且手一抖就容易抄错。后来我花了半天时间把“批量查询号码归属地”这活儿给工具化了几万条数据跑完只要几分钟。如果你也在处理类似需求——不管是做用户分群、短信发送前的号段分析还是渠道投放的反作弊排查——这篇文章里讲到的方案、代码和坑应该能帮你省下不少时间。1. 需求场景梳理与方案选型“批量查询号码归属地”这个需求拆开来看其实就是两件事准确识别号码所属的省市运营商以及用尽量高的效率处理大量号码。不同场景对这两点的侧重点完全不一样所以方案不能一把抓。1.1 先搞清楚你的真实场景我粗略把这需求分成三类你可以自己对号入座运营分析场景比如用户列表分群、区域活动策划。特点是量大、要求快但对单个号码的精度容忍度稍高差个把号段一般影响不大。业务系统实时校验场景比如注册环节想判断用户是否来自目标省份。特点是必须实时返回、响应要快不能因为查个归属地把接口拖慢。数据清洗与建模场景比如给存量数据打标签之后还要拿去做分析建模。特点是必须准确、可复现而且处理过程要能随时回溯。分清场景之后再选方案思路就清晰了。实时接口类方案适合小批量和实时场景离线方案才是批量处理的正道。1.2 三条主流技术路线对比我实际试下来市面上的方案基本就三条路各有各的适用边界方案实现方式优点缺点适合场景运营商/三方API逐个号码请求HTTP接口数据实时准确有QPS限制、按次计费、批量慢实时校验、小批量离线号段数据库本地SQLite/内存加载号段数据极快、零成本、无限制数据有滞后性、需更新大批量处理、本地分析大数据平台UDF写MapReduce/UDF函数跑批适合超大规模工程成本高千万级以上数据这里我想多说一句很多人一上来就找api觉得“在线“就一定比“离线“准。真实情况是号段归属地本身是个低频变化的数据运营商新增一个号段从发号到被各类离线库更新时间差通常在几天到几周之间。对绝大多数业务来说离线库的滞后性根本感知不到。所以如果你的需求是“这批几万条号码分别属于哪个省”直接用离线库就够了别花那个冤枉钱。2. 号码归属地数据结构和号段规则要写代码批量查询你得先理解号码本身的结构。手机号码虽然看起来是11位随机数字但前几位是有明确含义的理解了这个你才能看懂号段库为什么这么设计。2.1 手机号11位数字的构成以标准的11位手机号为例它其实分三段前3位网络识别号对应运营商和号段大类。比如139是移动老号段、186是联通、189是电信这是最粗粒度的区分。第4到第7位地区编码决定了这个号码归属哪个省市。这是判断归属地的核心依据。第8到第11位用户号码随机分配和归属地无关。所以判断归属地本质上就是把前7位号码作为“特征码”去号段库里匹配对应的省市运营商记录。这也是所有离线库的基本原理——你不需要存全量11位组合那数据量太大了存前7位再映射就够用。注意这里说的“前7位定归属”是针对绝大多数常规号码。实际库里会有一些特殊兼容处理比如携号转网后归属地不变、物联网卡号段特殊等后面我会专门讲坑。2.2 离线号段库的核心字段一个能用的离线号段库核心字段就这么几个prefix前7位作为主键。province所属省份。city所属城市。isp运营商类型移动/联通/电信/广电。area_code区号这个字段不是必须的但有些场景会用到。post_code邮编同样视情况保留。入库之后你可能还会额外加一些字段比如更新日期、数据来源标记方便以后排查问题。拿到一个原始的号段json或csv之后第一步不是急着写查询代码而是先做数据清洗。2.3 数据清洗的细节这个环节非常枯燥但特别重要。我踩过一次数据源给的csv编码不对全表乱码差点直接拿来上线。清洗阶段至少要做这几件事统一数据源编码为UTF-8别信原始文件的编码标记。去重。同一个prefix可能在不同批次数据源里重复出现以最新日期为准。校验省份/城市字段脏数据很有可能是空值或乱码。转成标准格式入库推荐直接用SQLite单文件、零依赖、查询性能足够。清洗完再验证一下数据量级。国内常规号段大概在几十万量级加上物联网和虚拟运营商号段一般不会超过百万条。这个量级用SQLite和Python处理性能完全够用不需要上重型数据库。3. 实战写一个批量查询工具这个部分直接上代码。我拿当前比较通用的Python 3来完成整套流程方案不绑定任何特定框架你拿到之后稍微改改路径就能跑起来。3.1 初始化数据库并加载号段数据假设你已经把号段数据整理成了一份CSV文件字段为prefix,province,city,isp下面这段代码的作用是把它导入SQLite并建好索引import sqlite3 import csv DB_PATH phone_prefix.db CSV_PATH phone_prefix.csv def init_db(): conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS prefix_map ( prefix TEXT PRIMARY KEY, province TEXT, city TEXT, isp TEXT ) ) # prefix 已经是主键自带索引查询性能足够 conn.commit() return conn def load_csv(conn): cur conn.cursor() with open(CSV_PATH, r, encodingutf-8) as f: reader csv.reader(f) next(reader, None) # 跳过表头 batch [] for row in reader: if len(row) 4: continue prefix, province, city, isp row[0].strip(), row[1].strip(), row[2].strip(), row[3].strip() if not prefix: continue batch.append((prefix, province, city, isp)) # 每 5000 条批量写入一次减少事务开销 if len(batch) 5000: cur.executemany( INSERT OR REPLACE INTO prefix_map VALUES (?, ?, ?, ?), batch ) batch [] if batch: cur.executemany( INSERT OR REPLACE INTO prefix_map VALUES (?, ?, ?, ?), batch ) conn.commit() if __name__ __main__: conn init_db() load_csv(conn) print(init done) conn.close()这段代码里有两个我特别想强调的点。第一INSERT OR REPLACE是为了处理前后批次数据有重复的情况以当前导入的数据为准这个策略比先删后插更稳。第二批量写入控制在5000条一次SQLite的事务提交压力会小很多实测导入五十万条数据也就几秒钟的事。3.2 单条查询与批量查询数据库准备好之后查询逻辑反而简单。单条查询直接取号码前7位去SQLite查就行但我想在这里多写几个防御性检查实际生产中这些细节能帮你省掉很多排查时间。import sqlite3 import json import re from concurrent.futures import ThreadPoolExecutor DB_PATH phone_prefix.db class PhoneQuery: def __init__(self, db_pathDB_PATH): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.conn.row_factory sqlite3.Row def normalize_phone(self, phone): 号码清洗去空格、横线转字符串 phone str(phone).strip() phone re.sub(r[\s\-], , phone) return phone def query_one(self, phone): 查询单条号码返回结果字典 phone self.normalize_phone(phone) if not re.fullmatch(r1\d{10}, phone): return {phone: phone, valid: False, reason: invalid_number} prefix phone[:7] cur self.conn.execute( SELECT province, city, isp FROM prefix_map WHERE prefix ?, (prefix,) ) row cur.fetchone() if row: return { phone: phone, valid: True, province: row[province], city: row[city], isp: row[isp], prefix: prefix, } return {phone: phone, valid: False, reason: prefix_not_found} def query_many(self, phones, max_workers8): 批量查询使用线程池并发执行 with ThreadPoolExecutor(max_workersmax_workers) as executor: results list(executor.map(self.query_one, phones)) return results def query_many_sql_batch(self, phones): 批量查询的另一种更快的写法一次性用 SQL IN 查出全部再映射 normalized [self.normalize_phone(p) for p in phones] valid_prefixes [p[:7] for p in normalized if re.fullmatch(r1\d{10}, p)] if not valid_prefixes: return {} # 返回 prefix - 归属信息 的映射 placeholders ,.join([?] * len(valid_prefixes)) sql fSELECT prefix, province, city, isp FROM prefix_map WHERE prefix IN ({placeholders}) cur self.conn.execute(sql, valid_prefixes) mapping {row[prefix]: dict(row) for row in cur.fetchall()} return mappingquery_one是基础query_many用线程池实现并发。不过我要提醒你SQLite的并发读在开了check_same_threadFalse之后确实能跑但线程数别开太猛实测8个线程基本就到头了再多对SQLite没意义反而让CPU上下文切换变多。如果你追求极致性能直接用query_many_sql_batch这一版它通过IN语句一次查出所有前缀再在内存里做映射比逐条查快一个数量级。3.3 主流程整合与结果导出有了查询模块剩下的就是组装主流程。核心流程是读入待查询号码 - 批量查询 - 结果导出为csv或json。我建议导出时把结果统一成flat结构这样后续无论是透视表还是进数据库都方便。import csv def main(): phones [] with open(input_phones.csv, r, encodingutf-8) as f: reader csv.reader(f) next(reader, None) for row in reader: if row: phones.append(row[0].strip()) pq PhoneQuery() # 使用 SQL 批量查询方式速度快 mapping pq.query_many_sql_batch(phones) results [] for phone in phones: norm pq.normalize_phone(phone) if not re.fullmatch(r1\d{10}, norm): results.append({phone: phone, valid: False, province: , city: , isp: }) continue info mapping.get(norm[:7]) if info: results.append({ phone: phone, valid: True, province: info[province], city: info[city], isp: info[isp], }) else: results.append({phone: phone, valid: False, province: , city: , isp: }) with open(output_result.csv, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnames[phone, valid, province, city, isp]) writer.writeheader() writer.writerows(results) print(done, total:, len(results)) if __name__ __main__: main()导出的时候编码我特意用了utf-8-sig这样生成的CSV文件直接用Excel打开不会乱码省得运维同事又要来找我问“为什么文件是乱码”。就这种小细节实际工作中特别加分。4. 性能优化进阶思路如果说前面的代码是能跑那这一节就是想让你跑得足够快。批量查询的场景里性能通常不是瓶颈——几十万条数据用上面的写法几秒就能跑完但到了几百万、上千万条就值得认真抠一抠了。4.1 内存映射代替频繁IOSQLite已经算是轻量级里很快的但如果你查询请求密集到每秒成千上万次SQLite的IO和锁竞争还是会有压力。这时候可以绕过数据库把号段数据直接加载进内存用字典做映射。class MemoryPhoneQuery: def __init__(self, db_pathDB_PATH): self.conn sqlite3.connect(db_path) self.load_to_memory() def load_to_memory(self): cur self.conn.execute(SELECT prefix, province, city, isp FROM prefix_map) self.mapping { row[prefix]: {province: row[province], city: row[city], isp: row[isp]} for row in cur.fetchall() } # 内存中的字典查询时间 O(1) def query_one(self, phone): phone str(phone).strip() if not re.fullmatch(r1\d{10}, phone): return None return self.mapping.get(phone[:7])几十万条数据load进内存占用的内存可能就几十MB非常划算。之后每次查询就是一次哈希查找速度比SQLite快一个量级。4.2 并行处理框架的选择如果数据量到了千万级单机Python的GIL会成为瓶颈。这时候有两条路用multiprocessing做多进程把号码列表分片后交给各个worker处理最后合并结果。如果需要持续跑批直接上Pandas的apply配合swifter或者dask但要注意Pandas本身的内存开销上亿条数据得谨慎。个人经验是Python单机处理千万级没问题只要你的分片逻辑和并发粒度设置合理。真正要注意的是别开几百个进程去抢同一个SQLite文件那一顿操作下来比串行还慢。稳妥做法是数据库只负责一次性导出全量号段到内存然后分片给多进程每个进程独立用内存字典查询。4.3 结果去重与合并批量查询还有一个容易被忽略的优化待查询号码列表往往存在大量重复。比如你导出的用户名单里同一个手机号可能出现多次。处理这类数据时可以先把号码set去重只查询唯一号码最后再通过映射关系还原到原始行。这个技巧在号码重复率高的场景下能省掉一大半没必要查询时间。5. 常见问题与排查经验这部分内容我原本没打算写但后来想了想批量查询号码归属地这个活儿坑都在细节里。代码本身不难难的是被乱七八糟的数据坑到怀疑人生。所以我总结了几个自己踩过的坑以及对应的排查思路。5.1 号码格式清洗的坑最大的坑就是号码格式不统一。真实数据里手机号经常会带86、-、空格甚至有人把号码做成了科学计数法比如13812345678在Excel里被转成1.38123E10导出的CSV里那列直接变成了浮点数。处理办法是在清洗阶段统一把号码转成字符串再做正则校验re.fullmatch(r1\d{10}, phone)这个规则虽然简单但能挡掉90%的脏数据。5.2 虚拟运营商和物联网卡号段常规手机号好办但虚拟运营商170/171/162/165/167等和物联网卡10648、10649等特殊号段经常在离线库里查不到或者查询结果不准。你最好在查询结果里加一个isp标记遇到这些特殊号段时明确提示而不是默默返回“查不到”。业务上怎么处理是产品的事至少技术上要能识别出来。5.3 携号转网和归属地漂移严格来说一个号码携号转网后归属地不变但运营商变了。号段库一般只记录发号时的初始运营商所以遇到携转用户会显示旧运营商。这个问题在离线库里无解除非接运营商的实时接口。批量场景下我一般选择忽略——因为比例太低不影响区域分析。但你心里要有数别到时候被业务方拿着一个携转用户的截图来问“为什么这里显示不对”你得能解释清楚。5.4 常见的几个问题速查问题可能原因排查思路查出来的省份明显不对号段库数据错乱抽样检查原始数据源重点看第4-7位大批量号码全部查不到prefix清洗错误或拼接bug打印前几条prefix对照号段库确认格式Excel打开CSV乱码编码用了UTF-8导出改用utf-8-sig程序内存飙升一次性加载了过多号码用分批迭代器每批处理10万条查询速度越来越慢连接池未关闭或数据库被反复写确认连接复用读写分离5.5 数据更新机制离线库的更新同样要设计好。建议不要手动覆盖线上文件而是用“新库先在测试环境验证再切换目录软链接指向新库”的方式发布。这一步看起来费事但能有效防止数据源本身有问题时线上直接出事故。6. 一些实操心得最后再分享几点我做这个批量查询工具时的体会。第一工具要做得可配置不要硬编码。数据库路径、线程数、输入输出文件路径这些最好像我代码里那样做成常量或参数甚至支持命令行传参。别小看这一步需求稍微一迭代你就能感受到它的价值。第二日志必须打。批量处理程序尤其需要记录日志处理了多少条、成功多少条、失败多少条、失败的原因分布是什么。这个信息不仅是让你心里有数更是你之后和业务方对接时最有力的沟通依据。你直接丢给业务方一份“5万条里有300条异常主要原因是号码格式错误”比一句“跑完了”专业太多。第三写自动化测试覆盖关键行为。哪怕是几行简单的单测把“正常手机号”、“座机号”、“乱码”、“带86的号”这几种情况固定下来也能防止你改了上游逻辑之后不小心把查询结果搞崩。这种东西不写的时候没人觉得必要写完之后再改代码你会感谢当年的自己。第四与其在接口调用上花精力不如把离线库维护好。离线库的正确性和时效性决定这个工具的天花板代码优化做到后面反倒是最基础的数据源质量决定了项目成败。我现在的习惯是每季度检查一次号段库版本并且我会用抽样的人工核对来验证准确性——从结果里随机挑20个号码上网核对一遍很快就能评估出这个数据源靠不靠谱。批量查询号码归属地本质上是件小而实用的工具活投入成本不高但能节省的重复劳动非常可观。希望这篇内容里的代码和经验对你实际工作有直接帮助。本文还有配套的精品资源点击获取