feat: ip2region 离线地理解析(优雅降级)
GeoResolver 进程级单例,c_buffer 模式一次读入内存(线程安全)。 适配 py-ip2region 3.0.4 实际 API:new_with_buffer(version, buff) 需 Version 对象,由 xdb 头 256 字节探测(2.0 老库自动归 IPv4)。 数据段布局按结构版本分派:3.0 新库 国家|省|城市|ISP|国家代码, 2.0 老库 国家|区域|省|市|ISP——错位会把 ISP 串进 city。 包/xdb 缺失或损坏时 resolve 一律返回 None,测试全绿。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
import functools
|
||||
import ipaddress
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
|
||||
def parse_region(raw, new_layout):
|
||||
"""ip2region 五段地区串 → {"country", "province", "city"};不足 4 段/空串返回 None。
|
||||
|
||||
两种数据布局(以 xdb 文件头的结构版本区分):
|
||||
- 旧版 2.0 结构:"国家|区域|省|市|ISP",省=parts[2]、市=parts[3];
|
||||
- 新版 3.0 结构(现行 v4/v6 库):"国家|省/区域|城市|ISP|国家代码",
|
||||
省=parts[1]、市=parts[2],ISP 段在 parts[3]——映射错位会把 ISP 串进 city。
|
||||
缺失段为 ip2region 的 "0" 占位,原样保留,由展示层决定如何呈现。
|
||||
"""
|
||||
parts = (raw or "").split("|")
|
||||
if len(parts) < 4:
|
||||
return None
|
||||
if new_layout:
|
||||
country, province, city = parts[0], parts[1], parts[2]
|
||||
else:
|
||||
country, province, city = parts[0], parts[2], parts[3]
|
||||
return {"country": country, "province": province, "city": city}
|
||||
|
||||
|
||||
class GeoResolver:
|
||||
"""离线 IP 地理解析(ip2region xdb,进程级单例)。
|
||||
|
||||
py-ip2region 3.0.4 实际 API(以其源码为准,与官方旧 README 不同):
|
||||
- ``searcher.new_with_buffer(version, c_buffer)`` 需要先传 ``util.IPv4``/``util.IPv6``
|
||||
版本对象(由 xdb 文件头 256 字节经 ``util.Header`` + ``util.version_from_header``
|
||||
探测;2.0 结构老库自动归为 IPv4);
|
||||
- ``Searcher`` 不支持 with 上下文管理器(无 __enter__/__exit__);
|
||||
- c_buffer 模式下 ``search(ip)`` 是纯内存切片(无文件 IO、无共享可变状态,
|
||||
官方声明线程安全),返回上述五段字符串。
|
||||
|
||||
包缺失 / xdb 未下载 / 文件损坏时优雅降级:resolve 一律返回 None。
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._searcher = None
|
||||
self._new_layout = True
|
||||
try:
|
||||
from ip2region import searcher as ip2region_searcher
|
||||
from ip2region import util as ip2region_util
|
||||
|
||||
with open(settings.IP2REGION_XDB, "rb") as f:
|
||||
buff = f.read()
|
||||
if len(buff) < ip2region_util.HeaderInfoLength:
|
||||
return # 文件过短,绝非合法 xdb
|
||||
header = ip2region_util.Header(buff[: ip2region_util.HeaderInfoLength])
|
||||
version = ip2region_util.version_from_header(header)
|
||||
if version is None:
|
||||
return # 无法识别的 xdb 结构/IP 版本
|
||||
# 结构版本同时决定数据段布局:2.0 旧布局 / 3.0 新布局
|
||||
self._new_layout = header.version >= ip2region_util.XdbStructure30
|
||||
self._searcher = ip2region_searcher.new_with_buffer(version, buff)
|
||||
except Exception:
|
||||
self._searcher = None
|
||||
|
||||
@functools.lru_cache(maxsize=10000)
|
||||
def resolve(self, ip: str):
|
||||
"""返回 {"country", "province", "city"},内网/环回/非法 IP 或降级时返回 None。
|
||||
结果被 lru_cache 缓存(键含 self,单例常驻故无泄漏问题),调用方勿修改返回的 dict。"""
|
||||
if not ip:
|
||||
return None
|
||||
try:
|
||||
if ipaddress.ip_address(ip).is_private:
|
||||
return None
|
||||
except ValueError:
|
||||
return None
|
||||
if not self._searcher:
|
||||
return None
|
||||
try:
|
||||
raw = self._searcher.search(ip)
|
||||
return parse_region(raw, self._new_layout)
|
||||
except Exception:
|
||||
return None
|
||||
@@ -0,0 +1,73 @@
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from django.conf import settings
|
||||
|
||||
from tracker.services.geo import GeoResolver, parse_region
|
||||
|
||||
XDB_EXISTS = os.path.exists(settings.IP2REGION_XDB)
|
||||
|
||||
|
||||
def test_private_ip_returns_local():
|
||||
r = GeoResolver()
|
||||
assert r.resolve("192.168.1.5") is None # 内网不解析(后台显示"本地")
|
||||
assert r.resolve("127.0.0.1") is None
|
||||
|
||||
|
||||
def test_public_ip_parsed_or_none(monkeypatch):
|
||||
r = GeoResolver()
|
||||
# xdb 未下载/解析失败时优雅降级返回 None,不抛异常
|
||||
out = r.resolve("8.8.8.8")
|
||||
assert out is None or set(out) == {"country", "province", "city"}
|
||||
|
||||
|
||||
def test_parse_region_two_layouts():
|
||||
# 旧版 2.0 结构:"国家|区域|省|市|ISP"
|
||||
assert parse_region("中国|0|江苏省|南京市|电信", new_layout=False) == {
|
||||
"country": "中国", "province": "江苏省", "city": "南京市"}
|
||||
# 新版 3.0 结构(v4/v6 库):"国家|省/区域|城市|ISP|国家代码"
|
||||
assert parse_region("中国|江苏省|南京市|0|CN", new_layout=True) == {
|
||||
"country": "中国", "province": "江苏省", "city": "南京市"}
|
||||
# ISP 段不得串进 city
|
||||
assert parse_region("United States|California|0|Google LLC|US", new_layout=True) == {
|
||||
"country": "United States", "province": "California", "city": "0"}
|
||||
assert parse_region("", new_layout=True) is None
|
||||
assert parse_region("中国", new_layout=True) is None # 不足 4 段
|
||||
assert parse_region(None, new_layout=False) is None
|
||||
|
||||
|
||||
class _FakeSearcher:
|
||||
def __init__(self, raw):
|
||||
self._raw = raw
|
||||
|
||||
def search(self, ip):
|
||||
return self._raw
|
||||
|
||||
|
||||
def test_resolve_uses_layout_dispatch():
|
||||
r = GeoResolver()
|
||||
r._searcher = _FakeSearcher("中国|北京市|北京市|百度|CN")
|
||||
r._new_layout = True
|
||||
assert r.resolve("1.2.3.4") == {
|
||||
"country": "中国", "province": "北京市", "city": "北京市"}
|
||||
|
||||
r2 = GeoResolver()
|
||||
r2._searcher = _FakeSearcher("中国|华北|北京市|北京市|百度")
|
||||
r2._new_layout = False
|
||||
assert r2.resolve("1.2.3.4") == {
|
||||
"country": "中国", "province": "北京市", "city": "北京市"}
|
||||
|
||||
|
||||
@pytest.mark.skipif(not XDB_EXISTS, reason="ip2region.xdb 未下载,跳过真实解析探针")
|
||||
def test_real_probe_public_ip():
|
||||
r = GeoResolver()
|
||||
assert r._searcher is not None, "xdb 存在时必须成功加载"
|
||||
raw = r._searcher.search("8.8.8.8")
|
||||
assert len(raw.split("|")) == 5, f"应为五段格式,实际: {raw!r}"
|
||||
out = r.resolve("8.8.8.8")
|
||||
assert set(out) == {"country", "province", "city"}
|
||||
assert out["country"] in ("美国", "United States")
|
||||
assert out["city"] != "Google LLC", "ISP 不得串进 city(布局映射错误)"
|
||||
cn = r.resolve("114.114.114.114")
|
||||
assert cn["country"] == "中国"
|
||||
assert cn["province"] not in ("", "0"), "国内公网 IP 应解析出省份"
|
||||
Reference in New Issue
Block a user