feat: 高精度离线GeoIP — MMDB支持 + 中文路径修复 + 分块自解压

- geoip.py: 修复中文路径 str()→Path 编码问题 (maxminddb Latin-1 bug)
- geoip.py: 启动自动重组geoip_chunks分块 + gunzip解压
- scripts/setup_geoip.py: 手动重组/解压脚本
- .gitignore: 忽略解压后的 .mmdb 文件

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
TianXuan Developer
2026-07-23 23:42:38 +08:00
parent 0f84497c82
commit e0a7400bd7
3 changed files with 103 additions and 2 deletions
+3
View File
@@ -18,6 +18,9 @@ runtime/
*.sqlite3-wal
*.sqlite3-shm
# GeoIP MMDB (decompressed — .gz versions are tracked)
*.mmdb
# Logs
logs/
*.log
+34 -2
View File
@@ -9,7 +9,9 @@
MMDB 查询使用 ``maxminddb`` 库,文本格式使用二分查找 O(log N)。
"""
import bisect
import gzip
import logging
import shutil
from pathlib import Path
from typing import Optional
@@ -58,6 +60,7 @@ def ip_to_int(ip_str: str) -> Optional[int]:
def _load_mmdb() -> None:
"""Try to load MaxMind MMDB databases from ``data/`` directory.
Auto-decompresses .mmdb.gz if the uncompressed file is missing.
Looks for:
- ``data/GeoLite2-City.mmdb`` (primary — city, country, lat/lon)
- ``data/GeoLite2-ASN.mmdb`` (optional — ISP, organization, ASN)
@@ -69,17 +72,46 @@ def _load_mmdb() -> None:
city_path = data_dir / 'GeoLite2-City.mmdb'
asn_path = data_dir / 'GeoLite2-ASN.mmdb'
# Auto-decompress gzipped databases (reassemble chunks first if needed)
chunks_dir = data_dir / 'geoip_chunks'
for mmdb_path in (city_path, asn_path):
gz_path = mmdb_path.with_suffix(mmdb_path.suffix + '.gz')
if not mmdb_path.exists() and not gz_path.exists() and chunks_dir.exists():
# Try reassembling from chunks
prefix = gz_path.name + '.part'
chunks = sorted(p for p in chunks_dir.iterdir() if p.name.startswith(prefix))
if chunks:
logger.info('[GEOIP] reassembling %d chunks → %s', len(chunks), gz_path.name)
try:
with open(gz_path, 'wb') as out:
for chunk in chunks:
with open(chunk, 'rb') as f:
shutil.copyfileobj(f, out, length=16 * 1024 * 1024)
logger.info('[GEOIP] reassembled %s', gz_path.name)
except Exception as e:
logger.warning('[GEOIP] failed to reassemble %s: %s', gz_path.name, e)
if not mmdb_path.exists() and gz_path.exists():
logger.info('[GEOIP] decompressing %s', gz_path.name)
try:
with gzip.open(gz_path, 'rb') as f_in:
with open(mmdb_path, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out, length=16 * 1024 * 1024)
logger.info('[GEOIP] decompressed → %s', mmdb_path.name)
except Exception as e:
logger.warning('[GEOIP] failed to decompress %s: %s', gz_path.name, e)
if not city_path.exists():
logger.info('[GEOIP] GeoLite2-City.mmdb not found at %s — will use text fallback', city_path)
return
try:
import maxminddb
_mmdb_city = maxminddb.open_database(str(city_path))
_mmdb_city = maxminddb.open_database(city_path)
logger.info('[GEOIP] loaded GeoLite2-City.mmdb (MMDB format)')
if asn_path.exists():
_mmdb_asn = maxminddb.open_database(str(asn_path))
_mmdb_asn = maxminddb.open_database(asn_path)
logger.info('[GEOIP] loaded GeoLite2-ASN.mmdb (ISP/ORG)')
except Exception as e:
logger.warning('[GEOIP] failed to open MMDB: %s', e)
+66
View File
@@ -0,0 +1,66 @@
#!/usr/bin/env python3
"""Reassemble split GeoIP database chunks into .mmdb.gz files, then decompress.
Chunks are stored in data/geoip_chunks/ to work around Git push size limits.
Run this once after cloning the repo, or let geoip.py auto-decompress at startup.
"""
import hashlib
import gzip
import shutil
from pathlib import Path
DATA_DIR = Path(__file__).parent.parent / 'data'
CHUNKS_DIR = DATA_DIR / 'geoip_chunks'
CHECKSUMS = {
'GeoLite2-City.mmdb.gz': 'a0b0fd27d587ab5caf1436bdbe8ab0a4',
'GeoLite2-ASN.mmdb.gz': '0b0fd27d587ab5caf1436bdbe8ab0a4', # placeholder
}
def reassemble(name: str) -> Path | None:
"""Reassemble chunks for `name` into a single .gz file. Returns path or None."""
prefix = name + '.gz.part'
chunks = sorted(
p for p in CHUNKS_DIR.iterdir()
if p.name.startswith(prefix)
)
if not chunks:
print(f'No chunks found for {name}.gz')
return None
out_path = DATA_DIR / (name + '.gz')
print(f'Reassembling {len(chunks)} chunks → {out_path.name}')
with open(out_path, 'wb') as out:
for chunk in chunks:
with open(chunk, 'rb') as f:
shutil.copyfileobj(f, out)
print(f'Done: {out_path.stat().st_size:,} bytes')
return out_path
def decompress(name: str) -> Path | None:
"""Decompress .mmdb.gz → .mmdb if not already present."""
gz_path = DATA_DIR / (name + '.gz')
mmdb_path = DATA_DIR / name
if mmdb_path.exists():
print(f'{name} already decompressed')
return mmdb_path
if not gz_path.exists():
print(f'{name}.gz not found')
return None
print(f'Decompressing {gz_path.name}{mmdb_path.name} ({gz_path.stat().st_size:,} bytes) …')
with gzip.open(gz_path, 'rb') as f_in:
with open(mmdb_path, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out, length=16 * 1024 * 1024)
print(f'Done: {mmdb_path.stat().st_size:,} bytes')
return mmdb_path
def main():
for name in ['GeoLite2-City.mmdb', 'GeoLite2-ASN.mmdb']:
gz_path = DATA_DIR / (name + '.gz')
if not gz_path.exists():
reassemble(name)
decompress(name)
print('GeoIP databases ready.')
if __name__ == '__main__':
main()