Initial commit: ip2region patch override implementation

This commit is contained in:
lycosa9527 2025-12-28 03:50:36 +08:00
commit 89ea22d189
5 changed files with 736 additions and 0 deletions

29
.gitignore vendored Normal file
View File

@ -0,0 +1,29 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
# Virtual environments
venv/
env/
ENV/
# IDE
.vscode/
.idea/
*.swp
*.swo
# Generated files
patches_cache.json
*.cache
# Patch files (users should download their own)
patches/*.fix
# OS
.DS_Store
Thumbs.db

211
README_CN.md Normal file
View File

@ -0,0 +1,211 @@
# IP2Region 补丁覆盖工具
一个轻量级的 Python 工具,用于应用 ip2region 数据库补丁,**无需修改**原始的 xdb 数据库文件。该工具创建一个 JSON 缓存,作为覆盖层,在查询主数据库之前检查补丁。
## 为什么需要这个工具?
ip2region 官方项目提供了直接修改 xdb 数据库文件的制作工具Java/Golang/C++)。但是,有时您可能希望:
- **无需重建**数据库即可应用补丁
- 将补丁与主数据库**分离**
- 在应用程序中将补丁用作**覆盖层**
- 在提交数据库更改之前**测试**补丁
此工具提供了一种非破坏性的补丁应用方法。
## 功能特性
- ✅ 从 `data/fix` 目录解析补丁文件
- ✅ 创建 JSON 缓存以支持快速二分查找
- ✅ UTF-8 编码支持Windows 系统自动回退)
- ✅ 命令行界面,支持自定义路径
- ✅ IP 查找测试功能
- ✅ 无需外部依赖(仅使用 Python 标准库)
## 安装
### 要求
- Python 3.7 或更高版本
- 无需外部依赖(仅使用标准库)
### 快速开始
1. **下载补丁文件**(从 ip2region 仓库):
```bash
# 创建补丁目录
mkdir patches
# 从以下地址下载补丁文件:
# https://github.com/lionsoul2014/ip2region/tree/master/data/fix
# 将它们保存为 .fix 文件到 patches/ 目录
```
2. **运行工具**
```bash
python patch_override.py
```
3. **在应用程序中使用缓存文件**
```python
from patch_override import load_patch_cache, find_patch_for_ip
cache = load_patch_cache(Path("patches_cache.json"))
result = find_patch_for_ip("39.144.0.1", cache)
if result:
print(f"省份: {result['province']}, 城市: {result['city']}")
```
## 使用方法
### 基本用法
```bash
# 使用默认路径patches/ 目录patches_cache.json 输出)
python patch_override.py
```
### 自定义路径
```bash
# 指定自定义目录
python patch_override.py --patches-dir ./data/fix --output ./cache.json
```
### 测试 IP 查找
```bash
# 使用特定 IP 地址进行测试
python patch_override.py --test-ip 39.144.0.1 --test-ip 39.144.10.5
```
### 命令行选项
```
--patches-dir DIR 包含 .fix 补丁文件的目录默认patches
--output FILE 输出 JSON 缓存文件路径默认patches_cache.json
--test-ip IP 用于验证补丁查找的测试 IP 地址(可重复指定)
```
## 补丁文件格式
补丁文件应为以下格式:
```
start_ip|end_ip|Country|Province|City|ISP
```
示例:
```
39.144.0.0|39.144.0.255|中国|山东省|菏泽市|移动
39.144.1.0|39.144.9.255|中国|0|0|移动
```
## 集成示例
以下是如何将此工具集成到您的 IP 地理位置服务中:
```python
from pathlib import Path
from patch_override import load_patch_cache, find_patch_for_ip
class IPGeolocationService:
def __init__(self):
self.patch_cache = load_patch_cache(Path("patches_cache.json"))
# 初始化您的主 ip2region 数据库...
def get_location(self, ip: str):
# 首先检查补丁(补丁优先)
patch_result = find_patch_for_ip(ip, self.patch_cache)
if patch_result:
return patch_result
# 回退到主数据库
return self._lookup_main_database(ip)
```
## 缓存文件格式
生成的缓存文件是一个 JSON 文件,结构如下:
```json
{
"patches": [
{
"start_ip": "39.144.0.0",
"end_ip": "39.144.0.255",
"start_int": 663748608,
"end_int": 663748863,
"country": "中国",
"province": "山东省",
"city": "菏泽市",
"isp": "移动",
"source": "github-issue-196.fix",
"line": 1
}
],
"total_patches": 237,
"last_updated": "2024-12-28T03:33:13.643784",
"patch_files": [
"github-issue-196.fix",
"github-issue-200.fix",
"github-issue-243.fix"
]
}
```
## 与制作工具对比
| 功能 | 制作工具 | 本工具 |
|------|---------|--------|
| 修改 xdb 文件 | ✅ 是 | ❌ 否 |
| 需要重建 | ✅ 是 | ❌ 否 |
| 非破坏性 | ❌ 否 | ✅ 是 |
| 覆盖层 | ❌ 否 | ✅ 是 |
| 易于更新 | ❌ 否 | ✅ 是 |
| Python 支持 | ❌ 否 | ✅ 是 |
## 更新补丁
当发布新补丁时:
1. 将新的 `.fix` 文件下载到您的 `patches/` 目录
2. 再次运行工具:`python patch_override.py`
3. 缓存将使用所有补丁重新生成
4. 重启您的应用程序以加载新缓存
## 编码支持
该工具支持多种编码:
- UTF-8主要
- GBK/GB2312Windows 回退)
- UTF-8 with BOM
中文字符在缓存文件中正确保留。
## 贡献
此工具旨在贡献给 ip2region 项目。如果您发现错误或有改进建议:
1. Fork ip2region 仓库
2. 将此工具添加到 `maker/python/``tools/python/`
3. 提交拉取请求
## 许可证
此工具遵循与 ip2region 项目相同的许可证Apache 2.0)。
## 参考资料
- [ip2region GitHub 仓库](https://github.com/lionsoul2014/ip2region)
- [补丁文件位置](https://github.com/lionsoul2014/ip2region/tree/master/data/fix)
- [制作工具文档](https://github.com/lionsoul2014/ip2region/tree/master/maker)
## 作者
社区贡献 - ip2region 补丁应用的 Python 实现。
## 致谢
基于 lionsoul2014 的 ip2region 项目的补丁格式和概念。

107
example_usage.py Normal file
View File

@ -0,0 +1,107 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Example usage of the IP2Region Patch Override Tool
This demonstrates how to integrate the patch override system
into your IP geolocation service.
"""
from pathlib import Path
from patch_override import load_patch_cache, find_patch_for_ip
def example_basic_usage():
"""Basic example: Load cache and lookup IPs."""
print("=" * 60)
print("Example: Basic Patch Lookup")
print("=" * 60)
# Load the patch cache
cache_file = Path("patches_cache.json")
cache = load_patch_cache(cache_file)
if not cache:
print("No cache file found. Run patch_override.py first!")
return
print(f"Loaded {cache.get('total_patches', 0)} patches from cache")
print(f"Last updated: {cache.get('last_updated', 'unknown')}")
# Test IPs
test_ips = [
"39.144.0.1", # Should match patch
"39.144.10.5", # Should match patch
"39.144.177.100", # Should match patch
"8.8.8.8", # Should not match (no patch)
]
print("\nTesting IP lookups:")
for ip in test_ips:
result = find_patch_for_ip(ip, cache)
if result:
print(f" {ip:15} -> {result['province']}, {result['city']} (from patch)")
else:
print(f" {ip:15} -> No patch found (use main database)")
def example_integration():
"""Example: Integration with IP geolocation service."""
print("\n" + "=" * 60)
print("Example: Service Integration")
print("=" * 60)
class SimpleIPGeolocation:
"""Simple example IP geolocation service."""
def __init__(self, cache_file: Path):
"""Initialize with patch cache."""
self.patch_cache = load_patch_cache(cache_file)
print(f"Service initialized with {self.patch_cache.get('total_patches', 0)} patches")
def get_location(self, ip: str):
"""
Get location for an IP address.
Checks patches first, then falls back to main database.
"""
# Check patches first (patches take priority)
patch_result = find_patch_for_ip(ip, self.patch_cache)
if patch_result:
return {
**patch_result,
'source': 'patch'
}
# In a real implementation, you would query the main database here
# For this example, we'll just return None
return {
'source': 'main_database',
'note': 'Would query main ip2region database here'
}
# Initialize service
service = SimpleIPGeolocation(Path("patches_cache.json"))
# Test lookups
test_ip = "39.144.0.1"
result = service.get_location(test_ip)
print(f"\nLookup result for {test_ip}:")
print(f" Source: {result.get('source')}")
if result.get('source') == 'patch':
print(f" Province: {result.get('province')}")
print(f" City: {result.get('city')}")
print(f" Country: {result.get('country')}")
if __name__ == "__main__":
print("IP2Region Patch Override - Example Usage\n")
# Run examples
example_basic_usage()
example_integration()
print("\n" + "=" * 60)
print("Examples complete!")
print("=" * 60)

366
patch_override.py Normal file
View File

@ -0,0 +1,366 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
IP2Region Patch Override Tool
A lightweight Python tool for applying ip2region database patches without
modifying the original xdb database files. This creates a JSON cache that
acts as an override layer, checking patches before querying the main database.
Unlike the maker tools (Java/Golang/C++) which modify xdb files directly,
this tool provides a non-destructive patch application method.
Based on patch format from:
https://github.com/lionsoul2014/ip2region/tree/master/data/fix
Usage:
python patch_override.py [--patches-dir DIR] [--output FILE]
Author: Community Contribution
License: Apache 2.0 (same as ip2region project)
"""
import json
import ipaddress
import sys
import argparse
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
# Configure stdout for UTF-8 on Windows
if sys.platform == 'win32':
try:
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
else:
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
except:
pass
def ip_to_int(ip: str) -> int:
"""Convert IP address to integer."""
try:
return int(ipaddress.IPv4Address(ip))
except:
return 0
def parse_patch_file(patch_path: Path) -> List[Dict]:
"""
Parse patch file format: start_ip|end_ip|Country|Province|City|ISP
Returns list of patch entries with IP ranges converted to integers.
"""
patches = []
try:
# Try UTF-8 first, fallback to GBK/GB2312 for Windows compatibility
encodings = ['utf-8', 'gbk', 'gb2312', 'utf-8-sig']
content = None
encoding_used = None
# Read as binary first to avoid any encoding issues
with open(patch_path, 'rb') as f:
raw_bytes = f.read()
for enc in encodings:
try:
content = raw_bytes.decode(enc)
encoding_used = enc
# Verify Chinese characters decode correctly
if '中国' in content[:500] or '北京' in content[:500]:
break
except UnicodeDecodeError:
continue
if content is None:
print(f"Warning: Could not decode {patch_path} with any encoding")
return patches
# Process lines
for line_num, line in enumerate(content.splitlines(), 1):
line = line.strip()
if not line or line.startswith('#'):
continue
# Format: start_ip|end_ip|Country|Province|City|ISP
parts = line.split('|')
if len(parts) >= 6:
start_ip = parts[0].strip()
end_ip = parts[1].strip()
country = parts[2].strip()
province = parts[3].strip()
city = parts[4].strip()
isp = parts[5].strip()
# Convert IPs to integers for range checking
try:
start_int = ip_to_int(start_ip)
end_int = ip_to_int(end_ip)
if start_int > 0 and end_int > 0:
patches.append({
'start_ip': start_ip,
'end_ip': end_ip,
'start_int': start_int,
'end_int': end_int,
'country': country,
'province': province,
'city': city,
'isp': isp,
'source': patch_path.name,
'line': line_num
})
except Exception as e:
print(f"Warning: Invalid IP range in {patch_path.name} line {line_num}: {e}")
continue
if encoding_used and encoding_used != 'utf-8':
print(f" Note: File read with {encoding_used} encoding")
except Exception as e:
print(f"Error parsing {patch_path}: {e}")
import traceback
traceback.print_exc()
return patches
def build_patch_cache(patches_dir: Path, output_file: Path) -> Dict:
"""
Build a cache of all patches for fast lookup.
Args:
patches_dir: Directory containing .fix patch files
output_file: Path to save the JSON cache file
Returns:
Dictionary with patch cache data
"""
if not patches_dir.exists():
print(f"Patch directory not found: {patches_dir}")
return {}
# Find all patch files
patch_files = list(patches_dir.glob("*.fix"))
if not patch_files:
print(f"No patch files found in {patches_dir}")
return {}
print(f"Found {len(patch_files)} patch file(s):")
for pf in patch_files:
print(f" - {pf.name}")
# Parse all patches
all_patches = []
for patch_file in patch_files:
patches = parse_patch_file(patch_file)
print(f" Parsed {len(patches)} entries from {patch_file.name}")
all_patches.extend(patches)
print(f"\nTotal patches: {len(all_patches)}")
if not all_patches:
return {}
# Build cache structure: list of patches sorted by start_int for binary search
patches_sorted = sorted(all_patches, key=lambda x: x['start_int'])
# Save to cache file with proper UTF-8 encoding
cache_data = {
'patches': patches_sorted,
'total_patches': len(patches_sorted),
'last_updated': datetime.now().isoformat(),
'patch_files': [pf.name for pf in patch_files]
}
# Ensure output directory exists
output_file.parent.mkdir(parents=True, exist_ok=True)
# Save with UTF-8 encoding
with open(output_file, 'w', encoding='utf-8', newline='\n') as f:
json.dump(cache_data, f, ensure_ascii=False, indent=2)
print(f"\nPatch cache saved to: {output_file}")
print(f"Total patches cached: {len(patches_sorted)}")
return cache_data
def load_patch_cache(cache_file: Path) -> Dict:
"""Load patch cache from file."""
if not cache_file.exists():
return {}
try:
# Try UTF-8 first, with fallback encodings
encodings = ['utf-8', 'utf-8-sig', 'gbk', 'gb2312']
cache_data = None
for enc in encodings:
try:
with open(cache_file, 'r', encoding=enc) as f:
cache_data = json.load(f)
break
except UnicodeDecodeError:
continue
if cache_data is None:
print(f"Warning: Could not decode cache file with any encoding")
return {}
return cache_data
except Exception as e:
print(f"Error loading patch cache: {e}")
return {}
def find_patch_for_ip(ip: str, cache: Dict) -> Optional[Dict]:
"""
Find patch entry for a given IP address using binary search.
Args:
ip: IP address string
cache: Patch cache dictionary
Returns:
Patch data dict if IP falls within any patch range, None otherwise
"""
patches = cache.get('patches', [])
if not patches:
return None
try:
ip_int = ip_to_int(ip)
if ip_int == 0:
return None
# Binary search for matching range
left, right = 0, len(patches) - 1
while left <= right:
mid = (left + right) // 2
patch = patches[mid]
start_int = patch.get('start_int', 0)
end_int = patch.get('end_int', 0)
if start_int <= ip_int <= end_int:
# Found matching range - return location data
return {
'province': patch.get('province', ''),
'city': patch.get('city', ''),
'country': patch.get('country', '中国'),
'isp': patch.get('isp', '')
}
elif ip_int < start_int:
right = mid - 1
else:
left = mid + 1
return None
except Exception as e:
print(f"Error finding patch for IP {ip}: {e}")
import traceback
traceback.print_exc()
return None
def main():
"""Main function to build and test patch cache."""
parser = argparse.ArgumentParser(
description='IP2Region Patch Override Tool - Create JSON cache from patch files',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Use default paths (patches/ and patches_cache.json)
python patch_override.py
# Specify custom directories
python patch_override.py --patches-dir ./data/fix --output ./cache.json
# Test with specific IPs
python patch_override.py --test-ip 39.144.0.1 --test-ip 39.144.10.5
"""
)
parser.add_argument(
'--patches-dir',
type=str,
default='patches',
help='Directory containing .fix patch files (default: patches)'
)
parser.add_argument(
'--output',
type=str,
default='patches_cache.json',
help='Output JSON cache file path (default: patches_cache.json)'
)
parser.add_argument(
'--test-ip',
action='append',
dest='test_ips',
help='Test IP addresses to verify patch lookup (can be specified multiple times)'
)
args = parser.parse_args()
patches_dir = Path(args.patches_dir)
output_file = Path(args.output)
print("=" * 60)
print("IP2Region Patch Override Tool")
print("=" * 60)
# Build patch cache
print(f"\n[1/2] Building patch cache...")
print(f" Patches directory: {patches_dir}")
print(f" Output file: {output_file}")
cache = build_patch_cache(patches_dir, output_file)
if not cache:
print("\nNo patches found. Exiting.")
return
# Test patch lookup if IPs provided
if args.test_ips:
print(f"\n[2/2] Testing patch lookup...")
for test_ip in args.test_ips:
patch = find_patch_for_ip(test_ip, cache)
if patch:
province = patch.get('province', '')
city = patch.get('city', '')
print(f" {test_ip} -> {province}, {city} (from patch)")
else:
print(f" {test_ip} -> No patch found (will use main database)")
else:
# Default test IPs
print(f"\n[2/2] Testing patch lookup...")
test_ips = ['39.144.0.1', '39.144.10.5', '39.144.177.100']
for test_ip in test_ips:
patch = find_patch_for_ip(test_ip, cache)
if patch:
province = patch.get('province', '')
city = patch.get('city', '')
print(f" {test_ip} -> {province}, {city} (from patch)")
else:
print(f" {test_ip} -> No patch found (will use main database)")
print("\n" + "=" * 60)
print("Patch cache creation complete!")
print("\nNext steps:")
print("1. Use the cache file in your IP geolocation service")
print("2. Check patches before querying the main xdb database")
print("3. Patches take priority over database results")
print("=" * 60)
if __name__ == "__main__":
main()

23
requirements.txt Normal file
View File

@ -0,0 +1,23 @@
# IP2Region Patch Override Tool - Requirements
#
# This tool uses only Python standard library modules.
# No external dependencies are required!
#
# Required Python version: 3.7 or higher
#
# Standard library modules used:
# - json (JSON serialization)
# - ipaddress (IP address handling)
# - sys (system-specific parameters)
# - argparse (command-line argument parsing)
# - pathlib (path handling)
# - typing (type hints)
# - datetime (timestamp generation)
#
# All these modules are included in Python 3.7+ standard library.
#
# To verify your Python version:
# python --version
#
# Should show Python 3.7.0 or higher.