509 lines
17 KiB
Python
509 lines
17 KiB
Python
# +======================================================================
|
||
# | 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
|