浅念
e880dbe458
- 导出模块级 resolver 单例：Task 6 起 import resolver 使用， 防多实例各持 ~11MB xdb 副本、lru_cache 按实例分片 - 补公网 IPv6→None 测试，锁定 M1 v4-only 决策 - __init__ 降级路径（文件过短/结构无法识别/except）均记 warning 日志，运维可区分「没部署 xdb」与「库 API 崩了」 - 删掉测试未使用的 monkeypatch 参数 Co-Authored-By: Claude Code <noreply@anthropic.com>
92 lines
4.1 KiB
Python
92 lines
4.1 KiB
Python
import functools
|
||
import ipaddress
|
||
import logging
|
||
|
||
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:
|
||
logging.getLogger(__name__).warning(
|
||
"GeoResolver 降级运行：xdb 文件过短（%d 字节）", len(buff))
|
||
return
|
||
header = ip2region_util.Header(buff[: ip2region_util.HeaderInfoLength])
|
||
version = ip2region_util.version_from_header(header)
|
||
if version is None:
|
||
logging.getLogger(__name__).warning(
|
||
"GeoResolver 降级运行：无法识别的 xdb 结构/IP 版本（header=%s）", header)
|
||
return
|
||
# 结构版本同时决定数据段布局：2.0 旧布局 / 3.0 新布局
|
||
self._new_layout = header.version >= ip2region_util.XdbStructure30
|
||
self._searcher = ip2region_searcher.new_with_buffer(version, buff)
|
||
except Exception:
|
||
# 运维可据此区分"没部署 xdb"与"库 API 崩了"（含堆栈）
|
||
logging.getLogger(__name__).warning(
|
||
"GeoResolver 降级运行（xdb 缺失或加载失败）", exc_info=True)
|
||
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
|
||
|
||
|
||
# 进程级共享单例：Task 6 起 from tracker.services.geo import resolver 使用，
|
||
# 勿再各自实例化 GeoResolver()——每个实例会把 ~11MB xdb 完整读入内存，且 lru_cache 按实例分片。
|
||
resolver = GeoResolver()
|