# +====================================================================== # | DjangoAdmin 企业数字化开发平台 [ 赋能开发者,助力企业发展 ] # +---------------------------------------------------------------------- # | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 # +---------------------------------------------------------------------- # | 开源协议: Apache-2.0 # | 重要声明: 未经书面许可,不得移除或修改本框架相关版权标识。 # +---------------------------------------------------------------------- # | 官方网站: https://www.djangoadmin.cn # +---------------------------------------------------------------------- # | 研发团队: 小蚂蚁云团队 荣誉出品 # +====================================================================== # | 版权和免责声明: # | 一、知识产权声明 # | 本团队对该软件框架产品(以下简称“本产品”)拥有完整知识产权,包括但不限于商标权、专利权、 # | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权, # | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 # | 二、使用限制与禁止条款 # | 您理解并同意,禁止将本产品用于以下任何行为: # | 1. 任何违反中华人民共和国法律法规的用途; # | 2. 任何侵害他人合法权益(包括但不限于人身权、财产权、知识产权等)的行为; # | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 # | 三、责任豁免 # | 因使用本产品而产生的任何损失或法律纠纷(包括但不限于直接损失、间接损失、附带损失、衍生损失 # | 以及因违约、侵权等引发的责任),本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 # | 方提起索赔或行政机关进行调查,您应独立承担全部责任。 # | 四、违约与法律责任 # | 如发现任何违反本声明的行为,本团队有权: # | 1. 无条件配合公安机关及其他有权机关的调查取证工作; # | 2. 依法追究相关方的民事责任,并要求赔偿全部损失; # | 3. 保留进一步采取刑事控告等法律措施的权利。 # | 五、许可范围说明 # | 1、被授权方应视本产品源代码为核心机密信息,负有最高标准的保密与保管义务。禁止以任何形式(包 # | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 # | 过网络服务提供给任何第三方)泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 # | 传播行为均构成对本团队知识产权的严重侵权,本团队有权依法追究其全部法律责任,并要求赔偿因此造 # | 成的一切经济损失及商誉损害。 # | 2、本产品仅限于公司或个人在法律允许范围内,用于内部且合规的软件产品研发。具体权利义务以《软件 # | 许可协议》附件为准。 # +====================================================================== # +====================================================================== # | 模块: IP地理位置查询 # | 说明: 基于 ip2region.xdb 离线数据库提供高性能IP地理位置查询 # +====================================================================== """ IP地理位置查询模块(基于ip2region离线数据库) ============================================================= 基于 ip2region.xdb 离线数据库提供高性能IP地理位置查询,无需网络请求。 核心功能: - get_ip_region: 根据IP获取完整区域信息字典 - get_ip_location: 根据IP获取地理位置文本(兼容现有调用) - get_ip_country / get_ip_province / get_ip_city / get_ip_isp: 按层级获取 xdb文件位于 utils/ip2region.xdb,启动时全量加载到内存,查询零IO延迟。 """ import logging import os from ip2region import searcher, util logger = logging.getLogger(__name__) # xdb数据库文件路径(与本模块同目录) _XDB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ip2region.xdb") # 模块级单例:启动时加载xdb到内存,后续查询零IO _searcher = None # ============================================================ # 搜索器单例 # ============================================================ def _get_searcher(): """获取搜索器单例(懒加载,首次调用时初始化)""" global _searcher if _searcher is not None: return _searcher if not os.path.exists(_XDB_PATH): logger.error(f"ip2region数据库文件不存在: {_XDB_PATH}") return None try: c_buff = util.load_content_from_file(_XDB_PATH) _searcher = searcher.new_with_buffer(util.IPv4, c_buff) logger.info(f"ip2region数据库加载成功: {_XDB_PATH}") except Exception as e: logger.error(f"ip2region数据库加载失败: {e}") return None return _searcher # ============================================================ # 区域信息解析 # ============================================================ def _parse_region(region_str): """ 解析ip2region原始结果字符串 原始格式: "国家|省份|城市|ISP|国家代码" 示例: "中国|浙江省|杭州市|阿里|CN", "United States|California|0|Google LLC|US" Args: region_str: ip2region返回的原始字符串 Returns: dict: 各层级区域信息 """ parts = region_str.split("|") if region_str else [] # 补齐到5段 while len(parts) < 5: parts.append("0") country = parts[0] province = parts[1] city = parts[2] isp = parts[3] country_code = parts[4] # 清理占位符 "0" → 空字符串 province = "" if province == "0" else province city = "" if city == "0" else city isp = "" if isp == "0" else isp country_code = "" if country_code == "0" else country_code return { "country": country, "province": province, "city": city, "isp": isp, "country_code": country_code, } # ============================================================ # 内网IP判断 # ============================================================ def _is_private_ip(ip): """判断是否为内网/保留IP(畸形输入不抛异常,按非内网处理交由查询侧兜底)""" if not ip: return True ip = ip.strip() if ip in ("127.0.0.1", "localhost", "0.0.0.0"): return True if ip.startswith("10.") or ip.startswith("192.168."): return True if ip.startswith("172."): parts = ip.split(".") if len(parts) >= 2: try: second = int(parts[1]) except (TypeError, ValueError): # 畸形IP(如 172.abc):无法判定内网,按非内网处理, # 避免抛异常冒泡导致 record_login_log/record_operation_log 整段中断 return False if 16 <= second <= 31: return True return False # ============================================================ # 公开接口 # ============================================================ def get_ip_region(ip): """ 根据IP获取完整区域信息字典 Args: ip: IP地址字符串 Returns: dict: 区域信息,包含 country/province/city/isp/country_code 字段。 内网IP返回 {"country": "内网IP", ...},查询失败返回 {"country": "未知", ...} """ try: if _is_private_ip(ip): return {"country": "内网IP", "province": "", "city": "", "isp": "", "country_code": ""} s = _get_searcher() if s is None: return {"country": "未知", "province": "", "city": "", "isp": "", "country_code": ""} region_str = s.search(ip) if not region_str: return {"country": "未知", "province": "", "city": "", "isp": "", "country_code": ""} return _parse_region(region_str) except Exception as e: # 畸形IP/查询异常一律降级"未知",绝不向上抛异常中断登录/操作日志落库 logger.warning(f"IP地理位置查询失败 [{ip}]: {e}") return {"country": "未知", "province": "", "city": "", "isp": "", "country_code": ""} def get_ip_location(ip): """ 根据IP获取地理位置文本描述 格式: "国家 省份 城市"(与原 utils/ip.py 的 get_ip_location 返回格式兼容) 内网IP返回 "内网IP",查询失败返回 "未知" Args: ip: IP地址字符串 Returns: str: 地理位置文本 """ region = get_ip_region(ip) if region["country"] in ("内网IP", "未知"): return region["country"] parts = [region["country"]] if region["province"]: parts.append(region["province"]) if region["city"] and region["city"] != region["province"]: parts.append(region["city"]) return " ".join(parts) def get_ip_country(ip): """根据IP获取国家""" return get_ip_region(ip)["country"] def get_ip_province(ip): """根据IP获取省份""" return get_ip_region(ip)["province"] def get_ip_city(ip): """根据IP获取城市""" return get_ip_region(ip)["city"] def get_ip_isp(ip): """根据IP获取运营商""" return get_ip_region(ip)["isp"]