djangoadmin/DjangoAdmin_Django_EleVue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
gitadmin
2026-09-12 10:26:08 +08:00
parent 2548231e88
commit faa046bec9
12 changed files with 2950 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/ip_location.py
+508
View File
@@ -0,0 +1,508 @@
# +======================================================================
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
# +----------------------------------------------------------------------
# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有
# +----------------------------------------------------------------------
# | 开源协议: Apache-2.0
# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
# +----------------------------------------------------------------------
# | 官方网站: https://www.djangoadmin.cn
# +----------------------------------------------------------------------
# | 研发团队: 小蚂蚁云团队 荣誉出品
# +======================================================================
# | 版权和免责声明:
# | 一、知识产权声明
# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
# | 二、使用限制与禁止条款
# | 您理解并同意，禁止将本产品用于以下任何行为：
# | 1. 任何违反中华人民共和国法律法规的用途；
# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
# | 三、责任豁免
# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
# | 四、违约与法律责任
# | 如发现任何违反本声明的行为，本团队有权：
# | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
# | 3. 保留进一步采取刑事控告等法律措施的权利。
# | 五、许可范围说明
# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
# | 成的一切经济损失及商誉损害。
# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
# | 许可协议》附件为准。
# +======================================================================
"""
IP地理位置解析模块
=============================================================
该模块提供根据IP地址获取地理位置信息的功能。
支持多种IP数据库格式（如纯真IP数据库、GeoIP2等），
并提供缓存机制，减少重复查询。
主要功能：
1. 根据IP地址获取省市信息
2. 支持多种IP数据库格式
3. 提供缓存机制，提高查询效率
4. 支持本地数据库和在线API两种方式
依赖模块：
- requests: 调用在线API（可选）
- ipip-ipdb: 支持IPIP.NET数据库（可选）
- geoip2: 支持MaxMind GeoIP2数据库（可选）
"""
import os
import json
import logging
import hashlib
from functools import lru_cache
from typing import Optional, Tuple, Dict
import requests
from django.core.cache import cache
logger = logging.getLogger(__name__)
# =============================================================
# 配置项
# =============================================================
# IP数据库类型配置
# 可选值: 'local' - 本地数据库, 'api' - 在线API
IP_LOCATION_SOURCE = getattr(__import__('config.env', fromlist=['IP_LOCATION_SOURCE']),
'IP_LOCATION_SOURCE', 'api')
# 本地数据库文件路径
IP_DATABASE_PATH = getattr(__import__('config.env', fromlist=['IP_DATABASE_PATH']),
'IP_DATABASE_PATH',
os.path.join(os.path.dirname(__file__), 'data', 'qqwry.ipdb'))
# 在线API配置
IP_API_URL = getattr(__import__('config.env', fromlist=['IP_API_URL']),
'IP_API_URL', 'https://ip.taobao.com/outGetIpInfo')
IP_API_TIMEOUT = 5 # API请求超时时间（秒）
# 缓存配置
IP_CACHE_ENABLED = True # 是否启用缓存
IP_CACHE_TIMEOUT = 86400 # 缓存时间（秒），默认1天
# 默认返回信息（当无法获取时）
DEFAULT_LOCATION = "未知"
DEFAULT_PROVINCE = "未知"
DEFAULT_CITY = "未知"
# =============================================================
# IP数据库加载器
# =============================================================
class IPDatabaseLoader:
"""
IP数据库加载器
支持多种IP数据库格式：
1. 纯真IP数据库（QQWry）
2. IPIP.NET数据库
3. MaxMind GeoIP2数据库
"""
_instance = None
_db = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
self._db = None
self._load_database()
def _load_database(self):
"""加载IP数据库"""
if IP_LOCATION_SOURCE != 'local':
return
try:
# 尝试加载IPIP.NET数据库（.ipdb格式）
if os.path.exists(IP_DATABASE_PATH) and IP_DATABASE_PATH.endswith('.ipdb'):
import ipip
self._db = ipip.IPDatabase.open(IP_DATABASE_PATH)
logger.info(f"加载IPIP数据库成功: {IP_DATABASE_PATH}")
# 尝试加载纯真IP数据库（.dat格式）
elif os.path.exists(IP_DATABASE_PATH) and IP_DATABASE_PATH.endswith('.dat'):
import qqwry
self._db = qqwry.QQWry(IP_DATABASE_PATH)
logger.info(f"加载纯真IP数据库成功: {IP_DATABASE_PATH}")
# 尝试加载GeoIP2数据库
elif os.path.exists(IP_DATABASE_PATH) and IP_DATABASE_PATH.endswith('.mmdb'):
import geoip2.database
self._db = geoip2.database.Reader(IP_DATABASE_PATH)
logger.info(f"加载GeoIP2数据库成功: {IP_DATABASE_PATH}")
else:
logger.warning(f"IP数据库文件不存在或格式不支持: {IP_DATABASE_PATH}")
except ImportError as e:
logger.warning(f"IP数据库依赖未安装: {e}")
except Exception as e:
logger.error(f"加载IP数据库失败: {e}")
def get_location(self, ip: str) -> Tuple[str, str, str]:
"""
从本地数据库获取IP地理位置
Args:
ip: IP地址
Returns:
Tuple[str, str, str]: (省份, 城市, 详细地址)
"""
if not self._db:
return DEFAULT_PROVINCE, DEFAULT_CITY, DEFAULT_LOCATION
try:
# IPIP.NET数据库查询
if hasattr(self._db, 'find'):
info = self._db.find(ip)
if info:
# IPIP.NET返回格式: ['中国', '广东', '深圳', '电信']
province = info[1] if len(info) > 1 else DEFAULT_PROVINCE
city = info[2] if len(info) > 2 else DEFAULT_CITY
location = f"{province} {city}".strip()
return province, city, location
# 纯真IP数据库查询
elif hasattr(self._db, 'lookup'):
result = self._db.lookup(ip)
if result:
# 解析纯真数据库返回的地址
location_str = str(result)
parts = location_str.split()
province = parts[0] if parts else DEFAULT_PROVINCE
city = parts[1] if len(parts) > 1 else DEFAULT_CITY
return province, city, location_str
# GeoIP2数据库查询
elif hasattr(self._db, 'city'):
response = self._db.city(ip)
province = response.country.name if response.country.name else DEFAULT_PROVINCE
city = response.city.name if response.city.name else DEFAULT_CITY
location = f"{province} {city}".strip()
return province, city, location
except Exception as e:
logger.error(f"IP数据库查询失败: {ip}, 错误: {e}")
return DEFAULT_PROVINCE, DEFAULT_CITY, DEFAULT_LOCATION
# =============================================================
# 在线API查询器
# =============================================================
class IPAPILoader:
"""
在线IP查询API加载器
支持多种在线API服务：
1. 淘宝IP地址库（默认）
2. 百度IP定位API
3. 高德IP定位API
4. ip-api.com
"""
@staticmethod
def get_location(ip: str) -> Tuple[str, str, str]:
"""
通过在线API获取IP地理位置
Args:
ip: IP地址
Returns:
Tuple[str, str, str]: (省份, 城市, 详细地址)
"""
# 本地IP不需要查询
if ip in ['127.0.0.1', 'localhost', '::1']:
return "本地", "本地", "本地"
# 内网IP不需要查询
if ip.startswith(('192.168.', '10.', '172.16.', '172.17.', '172.18.',
'172.19.', '172.20.', '172.21.', '172.22.', '172.23.',
'172.24.', '172.25.', '172.26.', '172.27.', '172.28.',
'172.29.', '172.30.', '172.31.')):
return "内网", "内网", "内网IP"
# 尝试多种API
apis = [
IPAPILoader._query_taobao,
IPAPILoader._query_ip_api,
IPAPILoader._query_pconline,
]
for api in apis:
try:
province, city, location = api(ip)
if province != DEFAULT_PROVINCE:
return province, city, location
except Exception as e:
logger.debug(f"API查询失败 {api.__name__}: {e}")
continue
return DEFAULT_PROVINCE, DEFAULT_CITY, DEFAULT_LOCATION
@staticmethod
def _query_taobao(ip: str) -> Tuple[str, str, str]:
"""
淘宝IP地址库API
API文档: https://ip.taobao.com/
"""
url = f"https://ip.taobao.com/outGetIpInfo"
params = {
'ip': ip,
'accessKey': 'alibaba-inc'
}
response = requests.get(url, params=params, timeout=IP_API_TIMEOUT)
data = response.json()
if data.get('code') == 0:
ip_data = data.get('data', {})
province = ip_data.get('region', DEFAULT_PROVINCE)
city = ip_data.get('city', DEFAULT_CITY)
location = f"{province} {city}".strip()
if location == " XX":
location = province
return province, city, location
raise Exception(f"淘宝API返回错误: {data.get('msg')}")
@staticmethod
def _query_ip_api(ip: str) -> Tuple[str, str, str]:
"""
ip-api.com API (免费版)
API文档: http://ip-api.com/
"""
url = f"http://ip-api.com/json/{ip}"
params = {
'lang': 'zh-CN',
'fields': 'status,country,regionName,city'
}
response = requests.get(url, params=params, timeout=IP_API_TIMEOUT)
data = response.json()
if data.get('status') == 'success':
province = data.get('regionName', DEFAULT_PROVINCE)
city = data.get('city', DEFAULT_CITY)
location = f"{data.get('country', '')} {province} {city}".strip()
return province, city, location
raise Exception(f"ip-api返回错误")
@staticmethod
def _query_pconline(ip: str) -> Tuple[str, str, str]:
"""
太平洋电脑网IP查询API
"""
url = f"https://whois.pconline.com.cn/ipJson.jsp"
params = {
'ip': ip,
'json': 'true'
}
response = requests.get(url, params=params, timeout=IP_API_TIMEOUT)
# 太平洋API返回的是GBK编码
response.encoding = 'gbk'
data = response.text
# 解析JSONP格式的返回
if data.startswith('try{'):
data = data[4:-2]
result = json.loads(data)
if result:
province = result.get('pro', DEFAULT_PROVINCE)
city = result.get('city', DEFAULT_CITY)
location = result.get('addr', DEFAULT_LOCATION)
return province, city, location
raise Exception("太平洋API返回错误")
# =============================================================
# 主入口函数
# =============================================================
def get_ip_location(ip: str, use_cache: bool = True) -> Dict[str, str]:
"""
根据IP地址获取地理位置信息
Args:
ip: IP地址字符串
use_cache: 是否使用缓存，默认True
Returns:
Dict[str, str]: 地理位置信息，包含以下字段：
- province: 省份
- city: 城市
- location: 详细地址
- ip: 原始IP地址
使用示例:
>>> location = get_ip_location('183.14.30.110')
>>> print(location['province']) # 广东省
>>> print(location['city']) # 深圳市
>>> print(location['location']) # 广东省 深圳市
Notes:
1. 支持IPv4和IPv6地址
2. 内网IP返回"内网"标识
3. 查询结果会缓存，提高性能
"""
# 参数校验
if not ip or ip == 'unknown':
return _build_location_result(DEFAULT_PROVINCE, DEFAULT_CITY, DEFAULT_LOCATION, ip)
# 检查缓存
cache_key = f"ip_location_{ip}"
if use_cache and IP_CACHE_ENABLED:
cached = cache.get(cache_key)
if cached:
logger.debug(f"从缓存获取IP地理位置: {ip} -> {cached}")
return cached
try:
# 根据配置选择查询方式
if IP_LOCATION_SOURCE == 'local':
# 使用本地数据库
loader = IPDatabaseLoader()
province, city, location = loader.get_location(ip)
else:
# 使用在线API
province, city, location = IPAPILoader.get_location(ip)
# 构建结果
result = _build_location_result(province, city, location, ip)
# 缓存结果
if use_cache and IP_CACHE_ENABLED:
cache.set(cache_key, result, IP_CACHE_TIMEOUT)
return result
except Exception as e:
logger.error(f"获取IP地理位置失败: {ip}, 错误: {e}")
return _build_location_result(DEFAULT_PROVINCE, DEFAULT_CITY, DEFAULT_LOCATION, ip)
def get_ip_province(ip: str) -> str:
"""
获取IP所在省份
Args:
ip: IP地址
Returns:
str: 省份名称
"""
location = get_ip_location(ip)
return location.get('province', DEFAULT_PROVINCE)
def get_ip_city(ip: str) -> str:
"""
获取IP所在城市
Args:
ip: IP地址
Returns:
str: 城市名称
"""
location = get_ip_location(ip)
return location.get('city', DEFAULT_CITY)
def _build_location_result(province: str, city: str, location: str, ip: str) -> Dict[str, str]:
"""
构建地理位置结果字典
Args:
province: 省份
city: 城市
location: 详细地址
ip: IP地址
Returns:
Dict[str, str]: 标准化的地理位置信息
"""
# 清理数据
province = province.strip() if province else DEFAULT_PROVINCE
city = city.strip() if city else DEFAULT_CITY
location = location.strip() if location else f"{province} {city}".strip()
# 如果省份和城市相同，只显示一个
if province == city:
location = province
return {
'province': province,
'city': city,
'location': location,
'ip': ip
}
def batch_get_ip_locations(ips: list) -> Dict[str, Dict[str, str]]:
"""
批量获取多个IP的地理位置信息
Args:
ips: IP地址列表
Returns:
Dict[str, Dict[str, str]]: IP地址到地理位置信息的映射
"""
result = {}
for ip in ips:
result[ip] = get_ip_location(ip)
return result
# =============================================================
# 便捷函数
# =============================================================
def is_internal_ip(ip: str) -> bool:
"""
判断是否为内网IP
Args:
ip: IP地址
Returns:
bool: 是否为内网IP
"""
if ip in ['127.0.0.1', 'localhost', '::1']:
return True
if ip.startswith(('192.168.', '10.', '172.16.', '172.17.', '172.18.',
'172.19.', '172.20.', '172.21.', '172.22.', '172.23.',
'172.24.', '172.25.', '172.26.', '172.27.', '172.28.',
'172.29.', '172.30.', '172.31.')):
return True
return False