1. 纯真IP库API简介与适用场景纯真IP库是国内历史最悠久的IP地理位置数据库之一由纯真网络团队维护超过15年。这个数据库以高准确率和广泛覆盖率著称特别在中国大陆地区的IP定位精度可以达到区县级。数据库每周更新一次包含全球IPv4和IPv6地址的归属地信息。在实际工作中我们经常遇到需要批量处理IP地址归属地的场景网络安全分析时溯源攻击源地理位置用户行为分析中的地域分布统计CDN节点优化时的访问来源分析广告投放的地域定向验证传统的手动查询方式存在明显瓶颈网页查询每次只能处理单个IP多数免费接口有查询频率限制结果需要手动复制整理无法与现有系统集成纯真提供的API接口正好解决了这些问题。通过编程方式调用API我们可以实现单次请求处理数百个IP地址结构化返回JSON/XML格式数据直接集成到自动化流程中避免人工操作错误重要提示纯真API免费版有严格的调用限制商用场景建议购买企业授权。免费用户每日限额500次查询每分钟不超过10次请求。2. API接入准备与认证配置2.1 获取API访问密钥访问纯真官网注册开发者账号后进入控制台创建应用即可获得API Key。这个密钥需要妥善保管每个请求都必须携带。密钥格式通常为32位MD5字符串例如a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p62.2 请求基础参数说明纯真API采用标准的HTTP GET/POST请求核心参数包括参数名必填示例值说明key是a1b2...o5p6开发者密钥ip否8.8.8.8单IP查询时使用ips否8.8.8.8,114.114.114.114多IP查询时使用format否json返回格式(json/xml)lang否zh返回语言(zh/en)2.3 开发环境准备推荐使用Python 3.7环境需要安装requests库pip install requests准备测试用的IP列表文件ip_list.txt每行一个IP地址220.181.38.148 114.114.114.114 8.8.8.83. 批量查询实现方案详解3.1 基础单次查询实现我们先实现最基本的单个IP查询功能这有助于理解API的工作机制import requests def query_single_ip(api_key, ip_address): url http://api.cz88.net/data.php params { key: api_key, ip: ip_address, format: json } try: response requests.get(url, paramsparams) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f查询失败: {e}) return None # 使用示例 api_key your_api_key_here result query_single_ip(api_key, 8.8.8.8) print(result)典型返回结果示例{ code: 0, data: { ip: 8.8.8.8, country: 美国, province: , city: , isp: Google } }3.2 高效批量查询方案直接使用ips参数虽然可以批量查询但官方限制每次最多50个IP。我们实现分块处理机制import time from itertools import islice def batch_query_ips(api_key, ip_list, batch_size50, interval1): url http://api.cz88.net/batch.php results [] # 将IP列表分块处理 it iter(ip_list) while batch : list(islice(it, batch_size)): params { key: api_key, ips: ,.join(batch), format: json } try: response requests.get(url, paramsparams) response.raise_for_status() batch_result response.json() if batch_result.get(code) 0: results.extend(batch_result[data]) else: print(f批量查询出错: {batch_result.get(message)}) except requests.exceptions.RequestException as e: print(f请求异常: {e}) # 遵守API调用频率限制 time.sleep(interval) return results # 使用示例 with open(ip_list.txt) as f: ip_list [line.strip() for line in f if line.strip()] all_results batch_query_ips(api_key, ip_list)3.3 文件批处理完整实现结合文件读写实现完整的批处理工作流import json from datetime import datetime def process_ip_file(input_file, output_file, api_key): # 读取IP列表 with open(input_file) as f: ip_list [line.strip() for line in f if line.strip()] # 执行批量查询 start_time datetime.now() print(f开始处理 {len(ip_list)} 个IP地址...) results batch_query_ips(api_key, ip_list) # 保存结果 with open(output_file, w) as f: json.dump(results, f, ensure_asciiFalse, indent2) duration (datetime.now() - start_time).total_seconds() print(f处理完成耗时 {duration:.2f} 秒结果已保存到 {output_file}) # 使用示例 process_ip_file(ip_list.txt, results.json, api_key)4. 性能优化与异常处理4.1 查询性能优化策略合理设置批处理大小免费账户建议30-40个IP/次企业账户可提高到官方上限(通常100个/次)多线程加速实现from concurrent.futures import ThreadPoolExecutor def threaded_batch_query(api_key, ip_list, workers3): def query_worker(batch): return batch_query_ips(api_key, batch) # 将IP列表分成worker数量的块 chunk_size len(ip_list) // workers 1 chunks [ip_list[i:ichunk_size] for i in range(0, len(ip_list), chunk_size)] with ThreadPoolExecutor(max_workersworkers) as executor: results list(executor.map(query_worker, chunks)) return [item for sublist in results for item in sublist]缓存机制实现import os import pickle from hashlib import md5 def get_cache_key(ip_list): return md5(,.join(sorted(ip_list)).encode()).hexdigest() def cached_query(api_key, ip_list, cache_dir.ip_cache): os.makedirs(cache_dir, exist_okTrue) cache_key get_cache_key(ip_list) cache_file os.path.join(cache_dir, f{cache_key}.pkl) # 检查缓存 if os.path.exists(cache_file): with open(cache_file, rb) as f: return pickle.load(f) # 执行查询并缓存 results batch_query_ips(api_key, ip_list) with open(cache_file, wb) as f: pickle.dump(results, f) return results4.2 常见异常处理方案频率限制错误处理def safe_batch_query(api_key, ip_list, max_retries3): retries 0 while retries max_retries: try: return batch_query_ips(api_key, ip_list) except Exception as e: if limit exceeded in str(e).lower(): wait_time 2 ** retries # 指数退避 print(f触发频率限制等待 {wait_time} 秒后重试...) time.sleep(wait_time) retries 1 else: raise raise Exception(超过最大重试次数)IP格式校验函数import re def validate_ip(ip): pattern r^(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$ return re.match(pattern, ip) is not None def clean_ip_list(raw_ips): return [ip.strip() for ip in raw_ips if validate_ip(ip.strip())]结果后处理示例def process_results(raw_results): processed [] for item in raw_results: if not item: continue processed.append({ ip: item.get(ip, ), country: item.get(country, ), region: f{item.get(province, )}{item.get(city, )}, isp: item.get(isp, ), coordinates: get_coordinates(item) # 可添加地理编码功能 }) return processed5. 实际应用案例扩展5.1 与安全日志分析系统集成将IP查询功能集成到SIEM系统中自动标记异常登录的地理位置def analyze_security_logs(log_file, api_key): # 从日志中提取IP ip_pattern r\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3} with open(log_file) as f: log_ips set(re.findall(ip_pattern, f.read())) # 查询地理位置 ip_info batch_query_ips(api_key, log_ips) # 分析异常登录 suspicious [] for info in ip_info: if info[country] ! 中国 and info[isp] ! 阿里云: suspicious.append(info[ip]) print(f发现 {len(suspicious)} 个可疑IP:) for ip in suspicious: print(f- {ip})5.2 自动化报告生成将查询结果转换为可视化报告import pandas as pd import matplotlib.pyplot as plt def generate_report(results, output_file): # 转换为DataFrame df pd.DataFrame(results) # 统计地域分布 location_counts df[region].value_counts() # 生成图表 plt.figure(figsize(10, 6)) location_counts.head(10).plot(kindbarh) plt.title(IP地域分布TOP10) plt.tight_layout() plt.savefig(f{output_file}.png) # 保存详细数据 with pd.ExcelWriter(f{output_file}.xlsx) as writer: df.to_excel(writer, sheet_name原始数据) location_counts.to_frame(数量).to_excel(writer, sheet_name地域统计)5.3 长期IP监控系统建立持续监控关键IP变化的系统import schedule import time class IPMonitor: def __init__(self, api_key, watch_ips): self.api_key api_key self.watch_ips watch_ips self.history {} def check_ips(self): current batch_query_ips(self.api_key, self.watch_ips) changes [] for info in current: ip info[ip] if ip in self.history: old self.history[ip] if old[region] ! info[region]: changes.append(fIP {ip} 位置变化: {old[region]} → {info[region]}) self.history[ip] info if changes: self.alert(\n.join(changes)) def alert(self, message): # 实现邮件/短信通知 print(f发现变更:\n{message}) def run_daily(self): schedule.every().day.at(09:00).do(self.check_ips) while True: schedule.run_pending() time.sleep(60) # 使用示例 monitor IPMonitor(api_key, [8.8.8.8, 114.114.114.114]) monitor.run_daily()经验分享在实际使用中我们发现纯真API对教育网、企业专线等特殊IP段的解析精度较高但对某些云服务商的IP可能出现归属地漂移现象。建议关键业务场景配合多个IP库交叉验证。 
长河君 · 2026/11/2 8:12:26 · 阅读 1.2k 纯真IP库API批量查询与性能优化实践
1. 纯真IP库API简介与适用场景纯真IP库是国内历史最悠久的IP地理位置数据库之一由纯真网络团队维护超过15年。这个数据库以高准确率和广泛覆盖率著称特别在中国大陆地区的IP定位精度可以达到区县级。数据库每周更新一次包含全球IPv4和IPv6地址的归属地信息。在实际工作中我们经常遇到需要批量处理IP地址归属地的场景网络安全分析时溯源攻击源地理位置用户行为分析中的地域分布统计CDN节点优化时的访问来源分析广告投放的地域定向验证传统的手动查询方式存在明显瓶颈网页查询每次只能处理单个IP多数免费接口有查询频率限制结果需要手动复制整理无法与现有系统集成纯真提供的API接口正好解决了这些问题。通过编程方式调用API我们可以实现单次请求处理数百个IP地址结构化返回JSON/XML格式数据直接集成到自动化流程中避免人工操作错误重要提示纯真API免费版有严格的调用限制商用场景建议购买企业授权。免费用户每日限额500次查询每分钟不超过10次请求。2. API接入准备与认证配置2.1 获取API访问密钥访问纯真官网注册开发者账号后进入控制台创建应用即可获得API Key。这个密钥需要妥善保管每个请求都必须携带。密钥格式通常为32位MD5字符串例如a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p62.2 请求基础参数说明纯真API采用标准的HTTP GET/POST请求核心参数包括参数名必填示例值说明key是a1b2...o5p6开发者密钥ip否8.8.8.8单IP查询时使用ips否8.8.8.8,114.114.114.114多IP查询时使用format否json返回格式(json/xml)lang否zh返回语言(zh/en)2.3 开发环境准备推荐使用Python 3.7环境需要安装requests库pip install requests准备测试用的IP列表文件ip_list.txt每行一个IP地址220.181.38.148 114.114.114.114 8.8.8.83. 批量查询实现方案详解3.1 基础单次查询实现我们先实现最基本的单个IP查询功能这有助于理解API的工作机制import requests def query_single_ip(api_key, ip_address): url http://api.cz88.net/data.php params { key: api_key, ip: ip_address, format: json } try: response requests.get(url, paramsparams) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f查询失败: {e}) return None # 使用示例 api_key your_api_key_here result query_single_ip(api_key, 8.8.8.8) print(result)典型返回结果示例{ code: 0, data: { ip: 8.8.8.8, country: 美国, province: , city: , isp: Google } }3.2 高效批量查询方案直接使用ips参数虽然可以批量查询但官方限制每次最多50个IP。我们实现分块处理机制import time from itertools import islice def batch_query_ips(api_key, ip_list, batch_size50, interval1): url http://api.cz88.net/batch.php results [] # 将IP列表分块处理 it iter(ip_list) while batch : list(islice(it, batch_size)): params { key: api_key, ips: ,.join(batch), format: json } try: response requests.get(url, paramsparams) response.raise_for_status() batch_result response.json() if batch_result.get(code) 0: results.extend(batch_result[data]) else: print(f批量查询出错: {batch_result.get(message)}) except requests.exceptions.RequestException as e: print(f请求异常: {e}) # 遵守API调用频率限制 time.sleep(interval) return results # 使用示例 with open(ip_list.txt) as f: ip_list [line.strip() for line in f if line.strip()] all_results batch_query_ips(api_key, ip_list)3.3 文件批处理完整实现结合文件读写实现完整的批处理工作流import json from datetime import datetime def process_ip_file(input_file, output_file, api_key): # 读取IP列表 with open(input_file) as f: ip_list [line.strip() for line in f if line.strip()] # 执行批量查询 start_time datetime.now() print(f开始处理 {len(ip_list)} 个IP地址...) results batch_query_ips(api_key, ip_list) # 保存结果 with open(output_file, w) as f: json.dump(results, f, ensure_asciiFalse, indent2) duration (datetime.now() - start_time).total_seconds() print(f处理完成耗时 {duration:.2f} 秒结果已保存到 {output_file}) # 使用示例 process_ip_file(ip_list.txt, results.json, api_key)4. 性能优化与异常处理4.1 查询性能优化策略合理设置批处理大小免费账户建议30-40个IP/次企业账户可提高到官方上限(通常100个/次)多线程加速实现from concurrent.futures import ThreadPoolExecutor def threaded_batch_query(api_key, ip_list, workers3): def query_worker(batch): return batch_query_ips(api_key, batch) # 将IP列表分成worker数量的块 chunk_size len(ip_list) // workers 1 chunks [ip_list[i:ichunk_size] for i in range(0, len(ip_list), chunk_size)] with ThreadPoolExecutor(max_workersworkers) as executor: results list(executor.map(query_worker, chunks)) return [item for sublist in results for item in sublist]缓存机制实现import os import pickle from hashlib import md5 def get_cache_key(ip_list): return md5(,.join(sorted(ip_list)).encode()).hexdigest() def cached_query(api_key, ip_list, cache_dir.ip_cache): os.makedirs(cache_dir, exist_okTrue) cache_key get_cache_key(ip_list) cache_file os.path.join(cache_dir, f{cache_key}.pkl) # 检查缓存 if os.path.exists(cache_file): with open(cache_file, rb) as f: return pickle.load(f) # 执行查询并缓存 results batch_query_ips(api_key, ip_list) with open(cache_file, wb) as f: pickle.dump(results, f) return results4.2 常见异常处理方案频率限制错误处理def safe_batch_query(api_key, ip_list, max_retries3): retries 0 while retries max_retries: try: return batch_query_ips(api_key, ip_list) except Exception as e: if limit exceeded in str(e).lower(): wait_time 2 ** retries # 指数退避 print(f触发频率限制等待 {wait_time} 秒后重试...) time.sleep(wait_time) retries 1 else: raise raise Exception(超过最大重试次数)IP格式校验函数import re def validate_ip(ip): pattern r^(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$ return re.match(pattern, ip) is not None def clean_ip_list(raw_ips): return [ip.strip() for ip in raw_ips if validate_ip(ip.strip())]结果后处理示例def process_results(raw_results): processed [] for item in raw_results: if not item: continue processed.append({ ip: item.get(ip, ), country: item.get(country, ), region: f{item.get(province, )}{item.get(city, )}, isp: item.get(isp, ), coordinates: get_coordinates(item) # 可添加地理编码功能 }) return processed5. 实际应用案例扩展5.1 与安全日志分析系统集成将IP查询功能集成到SIEM系统中自动标记异常登录的地理位置def analyze_security_logs(log_file, api_key): # 从日志中提取IP ip_pattern r\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3} with open(log_file) as f: log_ips set(re.findall(ip_pattern, f.read())) # 查询地理位置 ip_info batch_query_ips(api_key, log_ips) # 分析异常登录 suspicious [] for info in ip_info: if info[country] ! 中国 and info[isp] ! 阿里云: suspicious.append(info[ip]) print(f发现 {len(suspicious)} 个可疑IP:) for ip in suspicious: print(f- {ip})5.2 自动化报告生成将查询结果转换为可视化报告import pandas as pd import matplotlib.pyplot as plt def generate_report(results, output_file): # 转换为DataFrame df pd.DataFrame(results) # 统计地域分布 location_counts df[region].value_counts() # 生成图表 plt.figure(figsize(10, 6)) location_counts.head(10).plot(kindbarh) plt.title(IP地域分布TOP10) plt.tight_layout() plt.savefig(f{output_file}.png) # 保存详细数据 with pd.ExcelWriter(f{output_file}.xlsx) as writer: df.to_excel(writer, sheet_name原始数据) location_counts.to_frame(数量).to_excel(writer, sheet_name地域统计)5.3 长期IP监控系统建立持续监控关键IP变化的系统import schedule import time class IPMonitor: def __init__(self, api_key, watch_ips): self.api_key api_key self.watch_ips watch_ips self.history {} def check_ips(self): current batch_query_ips(self.api_key, self.watch_ips) changes [] for info in current: ip info[ip] if ip in self.history: old self.history[ip] if old[region] ! info[region]: changes.append(fIP {ip} 位置变化: {old[region]} → {info[region]}) self.history[ip] info if changes: self.alert(\n.join(changes)) def alert(self, message): # 实现邮件/短信通知 print(f发现变更:\n{message}) def run_daily(self): schedule.every().day.at(09:00).do(self.check_ips) while True: schedule.run_pending() time.sleep(60) # 使用示例 monitor IPMonitor(api_key, [8.8.8.8, 114.114.114.114]) monitor.run_daily()经验分享在实际使用中我们发现纯真API对教育网、企业专线等特殊IP段的解析精度较高但对某些云服务商的IP可能出现归属地漂移现象。建议关键业务场景配合多个IP库交叉验证。