From 89ea22d1898a868c291baab32151b170ea37579b Mon Sep 17 00:00:00 2001 From: lycosa9527 Date: Sun, 28 Dec 2025 03:50:36 +0800 Subject: [PATCH] Initial commit: ip2region patch override implementation --- .gitignore | 29 ++++ README_CN.md | 211 ++++++++++++++++++++++++++ example_usage.py | 107 ++++++++++++++ patch_override.py | 366 ++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 23 +++ 5 files changed, 736 insertions(+) create mode 100644 .gitignore create mode 100644 README_CN.md create mode 100644 example_usage.py create mode 100644 patch_override.py create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..779b6ae --- /dev/null +++ b/.gitignore @@ -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 + diff --git a/README_CN.md b/README_CN.md new file mode 100644 index 0000000..c12d946 --- /dev/null +++ b/README_CN.md @@ -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/GB2312（Windows 回退） +- 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 项目的补丁格式和概念。 + diff --git a/example_usage.py b/example_usage.py new file mode 100644 index 0000000..6c755a3 --- /dev/null +++ b/example_usage.py @@ -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) + diff --git a/patch_override.py b/patch_override.py new file mode 100644 index 0000000..cd6f890 --- /dev/null +++ b/patch_override.py @@ -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() + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2485fb2 --- /dev/null +++ b/requirements.txt @@ -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. +