diff --git a/.gitignore b/.gitignore index ac10cc8..aa9d67a 100644 --- a/.gitignore +++ b/.gitignore @@ -19,4 +19,5 @@ cbuild .svn evui/ document/ -node_modules/ \ No newline at end of file +node_modules/ +uploads/ \ No newline at end of file diff --git a/utils/.gitignore b/utils/.gitignore deleted file mode 100644 index c96a04f..0000000 --- a/utils/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -* -!.gitignore \ No newline at end of file diff --git a/utils/R.py b/utils/R.py new file mode 100644 index 0000000..079605f --- /dev/null +++ b/utils/R.py @@ -0,0 +1,221 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +""" +统一响应结果模块 +============================================================= + +该模块提供了系统统一的JSON响应格式，确保所有API接口返回格式一致。 +包含成功响应和失败响应两种基础方法，支持扩展参数。 + +响应格式规范： +- code: 状态码，0表示成功，1表示失败（可扩展其他状态码） +- data: 响应数据，成功时返回具体数据，失败时为None +- msg: 提示信息，对操作结果的文字描述 +- ok: 布尔值，标识操作是否成功（兼容旧版前端） +- 其他扩展参数：可通过kwargs动态添加 + +使用示例： + # 成功响应 + return R.ok(data={"id": 1, "name": "张三"}) + return R.ok(msg="操作成功") + + # 失败响应 + return R.failed(msg="用户名或密码错误") + return R.failed(msg="服务器内部错误", code=500) + + # 带扩展参数的响应 + return R.ok(data=user_info, total=100, pages=10) +""" + +from typing import Any, Optional, Dict +from django.http import JsonResponse + +from constant.constants import MESSAGE_OK, MESSAGE_FAIL + + +def ok( + data: Optional[Any] = None, + msg: str = MESSAGE_OK, + code: int = 0, + **kwargs: Any +) -> JsonResponse: + """ + 生成成功响应 + + 构建一个标准的成功响应JSON对象，包含状态码、数据、提示信息等字段。 + 支持通过kwargs添加额外的响应字段，便于扩展。 + + Args: + data: 响应数据，可以是任意类型（dict, list, str, int等），默认为None + msg: 成功提示信息，默认使用常量MESSAGE_OK + code: 状态码，0表示成功，默认为0 + **kwargs: 额外的键值对参数，会合并到响应体中 + + Returns: + JsonResponse: 格式化的JSON响应对象 + + Examples: + >>> return R.ok() + {"code": 0, "data": None, "msg": "操作成功", "ok": True} + + >>> return R.ok(data={"id": 1, "name": "张三"}) + {"code": 0, "data": {"id": 1, "name": "张三"}, "msg": "操作成功", "ok": True} + + >>> return R.ok(msg="创建成功", total=100) + {"code": 0, "data": None, "msg": "创建成功", "ok": True, "total": 100} + """ + # 构建基础响应体 + response_body: Dict[str, Any] = { + "code": code, + "data": data, + "msg": msg, + "ok": True + } + + # 合并扩展参数 + if kwargs: + response_body.update(kwargs) + + # 返回JSON响应 + return JsonResponse(data=response_body) + + +def failed( + msg: str = MESSAGE_FAIL, + code: int = 1, + data: Optional[Any] = None, + **kwargs: Any +) -> JsonResponse: + """ + 生成失败响应 + + 构建一个标准的失败响应JSON对象，包含状态码、错误信息等字段。 + 支持通过kwargs添加额外的响应字段，便于扩展。 + + Args: + msg: 错误提示信息，默认使用常量MESSAGE_FAIL + code: 状态码，1表示失败，默认为1（可根据业务需求定义其他错误码） + data: 错误时的附加数据，通常为None，默认为None + **kwargs: 额外的键值对参数，会合并到响应体中 + + Returns: + JsonResponse: 格式化的JSON响应对象 + + Examples: + >>> return R.failed() + {"code": 1, "data": None, "msg": "操作失败", "ok": False} + + >>> return R.failed(msg="用户名或密码错误") + {"code": 1, "data": None, "msg": "用户名或密码错误", "ok": False} + + >>> return R.failed(msg="未授权访问", code=401) + {"code": 401, "data": None, "msg": "未授权访问", "ok": False} + + >>> return R.failed(msg="参数错误", errors={"name": "该字段不能为空"}) + {"code": 1, "data": None, "msg": "参数错误", "ok": False, "errors": {...}} + """ + # 构建基础响应体 + response_body: Dict[str, Any] = { + "code": code, + "data": data, + "msg": msg, + "ok": False + } + + # 合并扩展参数 + if kwargs: + response_body.update(kwargs) + + # 返回JSON响应 + return JsonResponse(data=response_body) + + +def response( + data: Optional[Any] = None, + msg: str = MESSAGE_OK, + code: int = 0, + success: bool = True, + **kwargs: Any +) -> JsonResponse: + """ + 通用响应生成器（高级用法） + + 提供更灵活的响应生成方式，允许自定义所有字段。 + 通常推荐使用ok()和failed()两个专用方法，但在需要完全控制响应格式时可以使用此方法。 + + Args: + data: 响应数据 + msg: 提示信息 + code: 状态码 + success: 操作是否成功标识 + **kwargs: 额外的键值对参数 + + Returns: + JsonResponse: 格式化的JSON响应对象 + + Examples: + >>> return R.response(data=[], msg="查询成功", code=200, success=True) + {"code": 200, "data": [], "msg": "查询成功", "ok": True} + """ + response_body: Dict[str, Any] = { + "code": code, + "data": data, + "msg": msg, + "ok": success + } + + if kwargs: + response_body.update(kwargs) + + return JsonResponse(data=response_body) + + +# 为方便使用，定义常用错误码常量（可选） +class Codes: + """常用状态码定义""" + SUCCESS = 0 + FAILED = 1 + UNAUTHORIZED = 401 # 未认证 + FORBIDDEN = 403 # 无权限 + NOT_FOUND = 404 # 资源不存在 + VALIDATE_ERROR = 422 # 验证错误 + SERVER_ERROR = 500 # 服务器错误 diff --git a/utils/common.py b/utils/common.py new file mode 100644 index 0000000..4bc993f --- /dev/null +++ b/utils/common.py @@ -0,0 +1,670 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +""" +通用工具函数模块 +============================================================= + +该模块提供项目中广泛使用的通用工具函数，涵盖token解析、类型判断、 +文件处理、文件操作等常用功能。这些工具函数在整个项目中重复使用， +提供了统一的处理逻辑，减少代码冗余。 + +主要功能： +1. Token解析工具 - 从请求中解析JWT token，获取用户信息 +2. 类型判断工具 - 判断变量类型并返回可读的类型名称 +3. 文件处理工具 - 文件URL处理、临时文件迁移、远程文件保存 +4. 富文本内容处理 - 提取并保存富文本中的文件 +5. IP地址处理 - 获取客户端IP和地理位置 +6. User-Agent解析 - 解析操作系统和浏览器信息 +7. 参数格式化 - 格式化请求参数和返回结果 + +所属包： utils +所在文件： utils/utils.py +依赖模块： os, re, time - 标准库 + config.env - 项目配置 + utils.file.mkdir - 目录创建工具 +""" +import logging +import os +import time + +from django.conf import settings + +from config.env import DJANGO_FILE_URL, DJANGO_TEMP_PATH, DJANGO_UPLOAD_DIR +import json +import re +import urllib.request + +from utils.file import mkdir + +# 获取日志记录器 +logger = logging.getLogger(__name__) + + +# ============================= 文件处理工具 ============================= + +# 获取文件地址 +def get_file_url(path): + """ + 获取完整的文件访问URL + ============================================================= + + 将文件的相对路径拼接上文件域名，生成完整的可访问URL。 + 如果路径为空，直接返回None，避免空值拼接。 + + 参数: + path (str): 文件的相对路径，可以是空字符串、None或有效路径 + + 返回: + str or None: 完整的文件访问URL，如果path为空则返回None + """ + # 参数验证：如果路径为空或None，直接返回None + if not path: + return None + + # 拼接文件域名和相对路径 + return DJANGO_FILE_URL + path + + +def save_file(url, directory): + """ + 保存文件到指定目录 + ============================================================= + + 处理文件保存逻辑，主要针对本站临时文件进行迁移。 + 将临时目录中的文件移动到正式存储目录，并返回新的相对路径。 + + 参数: + url (str): 文件的完整URL或相对路径 + dirname (str): 目标子目录名称，用于分类存储 + + 返回: + str: 文件的相对路径 + 失败时返回空字符串 + + 处理流程: + 1. 检查URL有效性 + 2. 判断是否为本站文件 + 3. 判断是否为临时文件 + 4. 创建目标目录 + 5. 移动文件到正式目录 + 6. 返回新路径 + + 文件分类说明: + - 临时文件：存储在 temp 目录下，需要迁移 + - 正式文件：已存储在 images 目录下，直接返回相对路径 + - 远程文件：其他站点的文件，暂未实现处理逻辑 + """ + # 判断文件地址是否为空 + if not url: + return "文件地址不能为空" + + # 判断是否是本站文件 + # 说明：通过判断URL是否包含文件域名来识别本站文件 + if url.find(DJANGO_FILE_URL) != -1: + # 本站文件处理 + if url.find("/temp") != -1: + # 临时文件处理 + # 说明：临时文件需要移动到正式目录 + + # 构建目标目录路径 + # 格式：正式文件路径/分类目录/日期目录 + path = DJANGO_UPLOAD_DIR + "/" + directory + "/" + time.strftime('%Y%m%d') + + # 创建存储目录 + # 函数：mkdir(path) + # 说明：确保目标目录存在 + mkdir(path) + + # 原始文件地址 + # 说明：将URL转换为本地文件系统路径 + old_path = url.replace(DJANGO_FILE_URL, DJANGO_UPLOAD_DIR) + + # 目标目录地址 + # 说明：构建移动后的新路径，移除temp目录部分 + new_path = DJANGO_UPLOAD_DIR + "/" + directory + old_path.replace(DJANGO_TEMP_PATH, "") + + # 移动文件 + # 函数：os.rename(oldPath, newPath) + # 说明：将文件从临时位置移动到正式位置 + os.rename(old_path, new_path) + + # 返回结果 + # 说明：返回新路径的相对部分（不包含上传目录） + return new_path.replace(DJANGO_UPLOAD_DIR, "") + else: + # 非临时文件处理 + # 说明：已经是正式文件，直接返回相对路径 + return url.replace(DJANGO_FILE_URL, "") + else: + # 远程文件处理 + # 说明：非本站文件，暂未实现下载保存逻辑 + print("远程文件处理") + + # 返回空字符串 + return "" + + +def save_content(content, title, directory): + """ + 保存富文本编辑器内容 + ============================================================= + + 处理富文本编辑器提交的内容，提取其中的文件并进行保存。 + 同时处理文件的alt属性，将其设置为文章标题。 + + 参数: + content (str): 富文本HTML内容 + title (str): 文章标题，用于设置文件alt属性 + dirname (str): 文件存储的子目录名称 + + 返回: + str: 处理后的富文本内容，文件地址已被替换 + + 处理流程: + 1. 使用正则表达式提取所有img标签的src属性 + 2. 遍历每个文件URL，调用save_file保存 + 3. 将原URL替换为新URL（带[IMG_URL]标记） + 4. 设置空alt为文章标题 + 5. 返回处理后的内容 + + 正则表达式说明: + 'img src="(.*?)"' + - img src=" : 匹配img标签的src属性开始 + - (.*?) : 非贪婪匹配，捕获引号内的URL + - " : 匹配结束引号 + - re.S : 使.匹配包括换行符在内的所有字符 + """ + # 正则取出富文本文件地址 + # 说明：找出所有img标签中的src属性值 + image_urls = re.findall('img src="(.*?)"', content, re.S) + + # 遍历文件地址 + for url in image_urls: + # 保存文件至本地 + # 函数：save_file(url, dirname) + # 说明：处理文件保存，返回新路径 + image = save_file(url, directory) + if image: + # 替换原文件地址 + # 说明：将原URL替换为新地址，添加[IMG_URL]标记用于后续处理 + content = content.replace(url, "[IMG_URL]" + image) + + # 设置alt标题 + # 说明：为没有alt属性的文件设置标题，有利于SEO + # 确保 title 是字符串类型 + title = str(title) if title else "" + if content.find("alt=\"\"") and title: + content = content.replace("alt=\"\"", "alt=\"" + title + "\"") + + # 返回结果 + return content + + +def get_table_name(base_name): + """获取带前缀的表名""" + prefix = getattr(settings, 'DATABASE_PREFIX', '') + return f"{prefix}{base_name}" + + +# ============================= IP地址处理工具 ============================= + +def get_client_ip(request) -> str: + """ + 获取客户端真实IP地址 + ============================================================= + + 通过HTTP请求头获取客户端的真实IP地址，支持代理环境。 + 优先从 HTTP_X_FORWARDED_FOR 获取，其次使用 REMOTE_ADDR。 + + Args: + request: HttpRequest对象 + + Returns: + str: 客户端IP地址 + + 处理逻辑: + 1. 检查 HTTP_X_FORWARDED_FOR 头（适用于代理环境） + 2. 如果存在，取第一个IP（客户端真实IP） + 3. 否则使用 REMOTE_ADDR（直接连接IP） + + 使用示例: + >>> ip = get_client_ip(request) + >>> print(f"客户端IP: {ip}") + """ + # 获取代理转发的IP列表 + x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') + if x_forwarded_for: + # 取第一个IP（客户端真实IP） + ip = x_forwarded_for.split(',')[0] + else: + # 直接获取连接IP + ip = request.META.get('REMOTE_ADDR') + return ip + + +def get_ip_location(ip: str) -> str: + """ + 根据IP获取地理位置（使用纯真IP库或第三方API） + ============================================================= + + 调用第三方IP地理位置API，获取IP所在的国家、省份、城市信息。 + + Args: + ip: IP地址 + + Returns: + str: 地理位置信息，格式："国家 省份 城市" + + 处理逻辑: + 1. 内网IP直接返回"内网IP" + 2. 调用 ip-api.com API 获取地理位置 + 3. 解析返回的JSON数据 + 4. 返回格式化后的地理位置 + + 注意: + - ip-api.com 免费版限制45次/分钟 + - 生产环境建议使用本地IP库如纯真IP库 + """ + # 内网IP或本地IP判断 + if ip in ['127.0.0.1', 'localhost'] or ip.startswith('192.168.') or ip.startswith('10.'): + return "内网IP" + + try: + # 使用ip-api.com获取地理位置（免费，限制45次/分钟） + # 生产环境建议使用本地IP库如纯真IP库 + url = f"http://ip-api.com/json/{ip}?lang=zh-CN" + with urllib.request.urlopen(url, timeout=3) as response: + data = json.loads(response.read().decode('utf-8')) + if data.get('status') == 'success': + # 返回格式化后的地理位置 + return f"{data.get('country', '')} {data.get('regionName', '')} {data.get('city', '')}" + except Exception as e: + # 记录警告日志 + logger.warning(f"获取IP地理位置失败: {e}") + + return "未知" + + +# ============================= User-Agent解析工具 ============================= + +def parse_user_agent(user_agent: str) -> dict: + """ + 解析User-Agent，获取操作系统和浏览器信息 + ============================================================= + + 通过正则表达式匹配User-Agent字符串，识别用户的操作系统和浏览器类型。 + + Args: + user_agent: User-Agent字符串 + + Returns: + dict: 包含os和browser的字典 + { + 'os': '操作系统', + 'browser': '浏览器' + } + + 识别规则: + 操作系统: + - Windows系列: Windows 10/8/7/XP + - macOS: macOS + - Linux: Linux + - Android: Android + - iOS: iOS + + 浏览器: + - Edge: Edge + - Chrome: Chrome + - Safari: Safari + - Firefox: Firefox + - IE: IE + + 使用示例: + >>> ua = request.META.get('HTTP_USER_AGENT', '') + >>> info = parse_user_agent(ua) + >>> print(f"操作系统: {info['os']}, 浏览器: {info['browser']}") + """ + # 初始化返回结果 + result = { + 'os': '未知', + 'browser': '未知' + } + + # 空值判断 + if not user_agent: + return result + + # 转换为小写，便于匹配 + ua_lower = user_agent.lower() + + # 操作系统识别规则 + os_patterns = [ + (r'windows nt 10\.0', 'Windows 10'), + (r'windows nt 6\.1', 'Windows 7'), + (r'windows nt 6\.2', 'Windows 8'), + (r'windows nt 6\.3', 'Windows 8.1'), + (r'windows nt', 'Windows'), + (r'mac os x', 'macOS'), + (r'linux', 'Linux'), + (r'android', 'Android'), + (r'iphone', 'iOS'), + (r'ipad', 'iOS'), + ] + + # 遍历匹配操作系统 + for pattern, os_name in os_patterns: + if re.search(pattern, ua_lower): + result['os'] = os_name + break + + # 浏览器识别规则 + browser_patterns = [ + (r'edg', 'Edge'), + (r'chrome', 'Chrome'), + (r'safari', 'Safari'), + (r'firefox', 'Firefox'), + (r'msie', 'IE'), + (r'trident', 'IE'), + ] + + # 遍历匹配浏览器 + for pattern, browser_name in browser_patterns: + if re.search(pattern, ua_lower): + result['browser'] = browser_name + break + + return result + + +# ============================= 参数格式化工具 ============================= + +def format_params(params) -> str: + """ + 格式化请求参数，用于日志存储 + ============================================================= + + 将请求参数字典转换为JSON字符串，并过滤敏感信息。 + + Args: + params: 请求参数（字典或其他类型） + + Returns: + str: 格式化后的JSON字符串 + + 处理逻辑: + 1. 如果参数为空，返回空字符串 + 2. 如果是字典类型，复制并过滤敏感字段 + 3. 敏感字段值替换为 '******' + 4. 转换为JSON字符串 + 5. 如果转换失败，返回字符串表示（限制500字符） + + 敏感字段: + - password: 密码 + - old_password: 旧密码 + - new_password: 新密码 + - confirm_password: 确认密码 + + 使用示例: + >>> params = {'username': 'admin', 'password': '123456'} + >>> result = format_params(params) + >>> # 结果: {"username": "admin", "password": "******"} + """ + # 空值判断 + if not params: + return "" + + try: + # 敏感字段列表 + sensitive_fields = ['password', 'old_password', 'new_password', 'confirm_password'] + # 如果是字典类型，需要过滤敏感信息 + if isinstance(params, dict): + # 复制字典，避免修改原数据 + params_copy = params.copy() + # 遍历敏感字段，替换值 + for field in sensitive_fields: + if field in params_copy: + params_copy[field] = '******' + # 转换为JSON字符串 + return json.dumps(params_copy, ensure_ascii=False) + # 非字典类型直接转换 + return json.dumps(params, ensure_ascii=False) + except Exception: + # 转换失败时，返回字符串表示（限制500字符） + return str(params)[:500] + + +def get_method_name(request): + """ + 获取请求方法名（如：user.add） + ============================================================= + + 从Django的路由解析结果中获取视图方法名称。 + + Args: + request: HttpRequest对象 + + Returns: + str: 方法名称，获取失败返回空字符串 + + 使用场景: + - 操作日志记录，标识具体执行的方法 + - 权限验证，根据方法名判断权限 + + 注意: + - 需要在URL配置中定义了name属性才能获取到 + - 如果未定义name，返回空字符串 + """ + # 从request.resolver_match获取路由名称 + if hasattr(request, 'resolver_match') and request.resolver_match: + return request.resolver_match.view_name or '' + return '' + + +def format_result(result): + """ + 格式化返回结果 + ============================================================= + + 将返回结果转换为适合存储的字符串格式，支持JsonResponse、字典等类型。 + + Args: + result: 返回结果（JsonResponse 或其他类型） + + Returns: + str: 格式化后的结果字符串 + + 处理逻辑: + 1. 空值判断 + 2. JsonResponse类型：解析content内容 + 3. 字典类型：转换为JSON字符串 + 4. 其他类型：转换为字符串 + 5. 长度超过2000字符时截断 + + 使用场景: + - 操作日志记录，保存接口返回结果 + - 接口调试，查看返回数据 + + 使用示例: + >>> response = R.ok(data={'id': 1, 'name': '张三'}) + >>> result_str = format_result(response) + >>> # 结果: {"code": 0, "data": {"id": 1, "name": "张三"}, "msg": "操作成功", "ok": true} + """ + # 空值判断 + if not result: + return '' + + try: + # 处理 JsonResponse 对象（R.ok 或 R.failed 返回的类型） + if hasattr(result, 'content') and result.content: + # 解码响应内容 + content = result.content.decode('utf-8') + if content: + try: + # 尝试解析并格式化 JSON + data = json.loads(content) + result_str = json.dumps(data, ensure_ascii=False) + except: + # 不是 JSON 直接返回字符串 + result_str = content + + # 限制长度，避免数据库字段溢出 + if len(result_str) > 10000: + result_str = truncate_json(result_str, 10000) + return result_str + return None + + # 处理字典类型 + elif isinstance(result, dict): + result_str = json.dumps(result, ensure_ascii=False) + if len(result_str) > 10000: + result_str = truncate_json(result_str, 10000) + return result_str + + # 处理其他类型 + else: + result_str = str(result) + if len(result_str) > 10000: + result_str = truncate_json(result_str, 10000) + return result_str + + except Exception as e: + # 记录异常日志 + logger.error(f"格式化返回结果异常: {e}") + # 返回截断后的字符串 + return truncate_json(str(result), 10000) if result else '' + + +def truncate_json(json_str, max_len): + """ + 截断JSON字符串 + ============================================================= + + 将超长的字符串截断，添加省略号提示。 + + Args: + json_str: JSON字符串 + max_len: 最大长度 + + Returns: + str: 截断后的字符串 + + 处理逻辑: + 1. 空值判断，返回空字符串 + 2. 长度未超限，返回原字符串 + 3. 长度超限，截取前max_len个字符，添加省略号 + + 使用示例: + >>> long_text = 'a' * 3000 + >>> result = truncate_json(long_text, 2000) + >>> # 结果: 'a' * 2000 + '...' + """ + # 空值判断 + if not json_str: + return '' + # 长度判断 + if len(json_str) <= max_len: + return json_str + # 截断并添加省略号 + return json_str[:max_len] + '...' + + +def parse_request_body(request): + """ + 解析请求体JSON数据 + + Args: + request: HttpRequest对象 + + Returns: + tuple: (data, error) + 成功时 data 为解析后的字典，error 为 None + 失败时 data 为 None，error 为错误描述字符串 + + Examples: + >>> data, error = parse_request_body(request) + >>> if error: + ... return R.failed(error) + >>> username = data.get('username') + """ + try: + # 检查请求体是否存在 + if not hasattr(request, 'body') or request.body is None: + # 没有请求体，返回空数据但不报错 + return None, None + + # 获取请求体内容 + if not request.body: + return None, "参数不能为空" + + json_data = request.body.decode('utf-8') + if not json_data: + return None, "参数不能为空" + + # 解析JSON + data = json.loads(json_data) + return data, None + + except json.JSONDecodeError as e: + logging.error(f"JSON解析异常: {str(e)}") + return None, "参数格式错误" + + except UnicodeDecodeError as e: + logging.error(f"字符编码异常: {str(e)}") + return None, "参数编码错误" + + except Exception as e: + logging.error(f"解析请求体异常: {str(e)}") + return None, "参数处理失败" + + +def format_datetime(dt, format_str='%Y-%m-%d %H:%M:%S'): + """ + 格式化日期时间 + + Args: + dt: datetime对象 + format_str: 日期时间格式字符串，默认为 '%Y-%m-%d %H:%M:%S' + + Returns: + str: 格式化后的日期字符串，如果输入为None则返回None + """ + return dt.strftime(format_str) if dt else None diff --git a/utils/file.py b/utils/file.py new file mode 100644 index 0000000..4e14715 --- /dev/null +++ b/utils/file.py @@ -0,0 +1,342 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +""" +文件目录操作工具模块 +============================================================= + +该模块提供文件和目录操作的常用工具函数。 +主要用于处理文件上传、目录创建、路径管理等场景，确保文件系统操作的 +安全性和可靠性。包含目录创建、文件操作等基础功能。 + +主要功能： +1. 递归创建目录（mkdir）- 确保目标目录存在 +2. 文件路径处理 +3. 目录存在性检查 + +所属包： utils +所在文件： utils/file_utils.py +依赖模块： os - 操作系统接口模块 +""" + +import os + + +def mkdir(path): + """ + 递归创建目录 + ============================================================= + + 检查指定路径的目录是否存在，如果不存在则递归创建所有必需的目录。 + 用于确保文件上传等操作前目标目录已存在。 + + 参数: + path (str): 需要创建/检查的目录路径 + + 返回: + bool: True - 目录创建成功或已存在 + False - 目录创建失败（保留扩展用） + + 使用示例: + >>> # 确保上传目录存在 + >>> upload_dir = '/data/uploads/2024/01' + >>> mkdir(upload_dir) + /data/uploads/2024/01 创建成功 + True + + 注意事项: + 1. 函数会自动创建多级不存在的目录 + 2. 如果目录已存在，不会重复创建 + 3. 需要有相应的文件系统权限 + """ + + # 去除首位空格 + # 说明：消除用户输入中可能包含的前后空格，避免路径错误 + path = path.strip() + + # 去除尾部 \ 符号 + # 说明：Windows系统路径可能以\结尾，去除后保持路径格式统一 + path = path.rstrip("\\") + + # 判断路径是否存在 + # 函数：os.path.exists(path) + # 说明：检查指定路径的文件或目录是否存在 + isExists = os.path.exists(path) + + if not isExists: + # 如果不存在则创建目录 + # 函数：os.makedirs(path) + # 说明：递归创建目录，如果中间目录不存在也会一并创建 + # 例如：创建 /a/b/c 时，如果 /a 和 /a/b 都不存在，会一并创建 + os.makedirs(path) + print(path + ' 创建成功') + return True + else: + # 如果目录存在则不创建，并提示目录已存在 + # 说明：目录已存在是正常情况，不需要重复创建 + print(path + ' 目录已存在') + return False + + +# ======================= 扩展：常用文件操作工具类 ========================== + +class FileUtils: + """ + 文件操作工具类 + ============================================================= + + 提供更多常用的文件和目录操作功能，封装为类方法便于调用。 + 包含文件读写、路径处理、文件信息获取等实用功能。 + + 使用示例: + >>> # 获取文件大小 + >>> size = FileUtils.get_file_size('/path/file.txt') + >>> print(f"{size} bytes") + + >>> # 安全删除文件 + >>> if FileUtils.safe_delete('/path/file.txt'): + ... print("文件删除成功") + """ + + @staticmethod + def get_file_size(file_path): + """ + 获取文件大小 + ============================================================= + + 参数: + file_path (str): 文件路径 + + 返回: + int: 文件大小（字节），如果文件不存在返回0 + """ + if os.path.exists(file_path) and os.path.isfile(file_path): + return os.path.getsize(file_path) + return 0 + + @staticmethod + def get_file_extension(filename): + """ + 获取文件扩展名 + ============================================================= + + 参数: + filename (str): 文件名或文件路径 + + 返回: + str: 文件扩展名（小写），例如：'.jpg', '.txt' + """ + import os + return os.path.splitext(filename)[1].lower() + + @staticmethod + def get_file_name_without_ext(filename): + """ + 获取不带扩展名的文件名 + ============================================================= + + 参数: + filename (str): 文件名或文件路径 + + 返回: + str: 不带扩展名的文件名 + """ + import os + return os.path.splitext(os.path.basename(filename))[0] + + @staticmethod + def safe_delete(file_path): + """ + 安全删除文件 + ============================================================= + + 参数: + file_path (str): 文件路径 + + 返回: + bool: True - 删除成功或文件不存在 + False - 删除失败（如权限不足） + """ + try: + if os.path.exists(file_path) and os.path.isfile(file_path): + os.remove(file_path) + return True + except Exception as e: + print(f"文件删除失败: {e}") + return False + + @staticmethod + def read_file(file_path, encoding='utf-8'): + """ + 读取文件内容 + ============================================================= + + 参数: + file_path (str): 文件路径 + encoding (str): 文件编码，默认utf-8 + + 返回: + str/None: 文件内容字符串，读取失败返回None + """ + try: + with open(file_path, 'r', encoding=encoding) as f: + return f.read() + except Exception as e: + print(f"文件读取失败: {e}") + return None + + @staticmethod + def write_file(file_path, content, encoding='utf-8', mode='w'): + """ + 写入文件内容 + ============================================================= + + 参数: + file_path (str): 文件路径 + content (str): 要写入的内容 + encoding (str): 文件编码，默认utf-8 + mode (str): 写入模式，'w'覆盖写入，'a'追加写入 + + 返回: + bool: True - 写入成功，False - 写入失败 + """ + try: + # 确保目录存在 + directory = os.path.dirname(file_path) + if directory and not os.path.exists(directory): + os.makedirs(directory) + + with open(file_path, mode, encoding=encoding) as f: + f.write(content) + return True + except Exception as e: + print(f"文件写入失败: {e}") + return False + + @staticmethod + def list_files(directory, pattern=None): + """ + 列出目录下的所有文件 + ============================================================= + + 参数: + directory (str): 目录路径 + pattern (str): 文件匹配模式，如 '*.txt'，None表示所有文件 + + 返回: + list: 文件名列表 + """ + import glob + + if not os.path.exists(directory): + return [] + + if pattern: + search_path = os.path.join(directory, pattern) + return [os.path.basename(f) for f in glob.glob(search_path)] + else: + return [f for f in os.listdir(directory) + if os.path.isfile(os.path.join(directory, f))] + + @staticmethod + def get_file_info(file_path): + """ + 获取文件详细信息 + ============================================================= + + 参数: + file_path (str): 文件路径 + + 返回: + dict: 包含文件信息的字典，文件不存在返回None + """ + if not os.path.exists(file_path) or not os.path.isfile(file_path): + return None + + stat = os.stat(file_path) + return { + 'name': os.path.basename(file_path), + 'path': os.path.abspath(file_path), + 'size': stat.st_size, + 'size_format': FileUtils._format_size(stat.st_size), + 'extension': FileUtils.get_file_extension(file_path), + 'created_time': stat.st_ctime, + 'modified_time': stat.st_mtime, + 'is_hidden': os.path.basename(file_path).startswith('.') + } + + @staticmethod + def _format_size(size_bytes): + """ + 格式化文件大小显示 + ============================================================= + + 参数: + size_bytes (int): 文件大小（字节） + + 返回: + str: 格式化后的大小字符串，如 '1.5 MB' + """ + if size_bytes < 1024: + return f"{size_bytes} B" + elif size_bytes < 1024 * 1024: + return f"{size_bytes / 1024:.2f} KB" + elif size_bytes < 1024 * 1024 * 1024: + return f"{size_bytes / (1024 * 1024):.2f} MB" + else: + return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB" + + +# ======================= 便捷函数封装 ========================== + +def get_file_size(file_path): + """获取文件大小（字节）的便捷函数""" + return FileUtils.get_file_size(file_path) + + +def safe_delete(file_path): + """安全删除文件的便捷函数""" + return FileUtils.safe_delete(file_path) + + +def get_file_extension(filename): + """获取文件扩展名的便捷函数""" + return FileUtils.get_file_extension(filename) diff --git a/utils/ip2region.py b/utils/ip2region.py new file mode 100644 index 0000000..c69eb2b --- /dev/null +++ b/utils/ip2region.py @@ -0,0 +1,243 @@ +# +====================================================================== +# | 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"] diff --git a/utils/ip2region.xdb b/utils/ip2region.xdb new file mode 100644 index 0000000..6b53b1f Binary files /dev/null and b/utils/ip2region.xdb differ diff --git a/utils/ip_location.py b/utils/ip_location.py new file mode 100644 index 0000000..d2249b6 --- /dev/null +++ b/utils/ip_location.py @@ -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 diff --git a/utils/jwt.py b/utils/jwt.py new file mode 100644 index 0000000..64f6af1 --- /dev/null +++ b/utils/jwt.py @@ -0,0 +1,623 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +# +====================================================================== +# | 模块: JWT令牌处理 +# | 说明: 提供JWT令牌的生成、验证、刷新和黑名单管理功能 +# +====================================================================== + +""" +JWT令牌处理模块 +============================================================= + +该模块提供基于JWT（JSON Web Token）的令牌生成和验证功能。 +主要用于用户登录认证，生成访问令牌，并在后续请求中验证令牌的有效性。 +支持令牌过期机制，确保系统的安全性。 + +主要功能： +1. JWT令牌生成（create_token）- 用于用户登录成功后生成访问令牌 +2. JWT令牌解析验证（parse_payload）- 验证请求中的令牌有效性 +3. 统一的异常处理，返回标准格式的验证结果 +4. 可配置的令牌过期时间 + +所属包： utils +所在文件： utils/jwt.py +依赖模块： jwt - JWT处理第三方库 + datetime - 时间处理模块 + hashlib - 摘要算法模块 + os - 环境变量读取模块 + logging - 日志记录模块 +""" + +import datetime +import hashlib +import logging +import os +import time + +import jwt +from jwt import exceptions + +# 配置日志记录器 +logger = logging.getLogger(__name__) + + +# ============================================================= +# JWT密钥配置 +# ============================================================= +# 优先从环境变量读取JWT密钥 +JWT_SALT = os.environ.get('JWT_SALT', None) + +# 如果没有环境变量，使用默认密钥 +if not JWT_SALT: + JWT_SALT = "ds()udsjo@jlsdosjf)wjd_#(#)$%^&*()_+=-0987654321" + +# JWT算法配置 +JWT_ALGORITHM = 'HS256' + +# 默认令牌过期时间（分钟） +DEFAULT_TIMEOUT_MINUTES = 20 + + +def _validate_key_strength(key: str) -> None: + """ + 启动时校验 JWT 密钥强度 + + 校验规则： + 1. 密钥不能为空 + 2. 密钥长度至少 32 字节（256 位） + + 校验失败直接抛出 RuntimeError，拒绝启动应用。 + """ + if not key: + raise RuntimeError( + "JWT_SALT 未配置！请在 .env 中设置至少32字节的强随机密钥。\n" + "生成方式：python -c \"import secrets; print(secrets.token_urlsafe(48))\"" + ) + + key_bytes = len(key.encode('utf-8')) + if key_bytes < 32: + logger.warning( + f"JWT密钥长度不足：{key_bytes} 字节，建议至少 32 字节（256位）。" + ) + + logger.info(f"JWT密钥校验通过，长度：{key_bytes}字节") + + +# 启动时校验密钥强度 +_validate_key_strength(JWT_SALT) + + +# ============================================================= +# JWT核心功能函数 +# ============================================================= + +def create_token(payload, timeout=None): + """ + 生成JWT令牌 + ============================================================= + + 根据传入的载荷数据生成JWT访问令牌，自动添加过期时间。 + 使用强密钥确保令牌安全性，符合RFC 7518安全规范。 + + 参数: + payload (dict): JWT载荷数据，通常包含用户ID、用户名等信息 + timeout (int): 令牌过期时间（分钟），默认使用DEFAULT_TIMEOUT_MINUTES + + 返回: + str: 生成的JWT令牌字符串 + + 使用示例: + >>> payload = {'userId': 1001, 'username': 'admin'} + >>> token = create_token(payload, timeout=30) + >>> print(token) + eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9... + + 注意事项: + 1. 密钥长度必须至少32字节，否则会发出警告 + 2. 生产环境建议通过环境变量设置JWT_SALT + 3. 令牌过期后需要重新登录获取新令牌 + """ + try: + # 设置过期时间 + # 说明：在载荷中添加exp字段，指定令牌的过期时间点 + # 计算方式：当前UTC时间 + 指定的分钟数 + if timeout is None: + timeout = DEFAULT_TIMEOUT_MINUTES + + # 创建payload副本，避免修改原始数据 + payload_copy = payload.copy() + + # 添加过期时间（exp）和签发时间（iat） + # exp: 过期时间，超过此时间令牌无效 + # iat: 签发时间，用于令牌生命周期计算 + current_time = datetime.datetime.now(tz=datetime.timezone.utc) + payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout) + payload_copy['iat'] = current_time + + # 声明类型和加密算法 + # 说明：定义JWT头部信息，包含令牌类型和加密算法 + headers = { + "typ": "JWT", # 令牌类型 + "alg": JWT_ALGORITHM # 加密算法 + } + + # 生成JWT令牌 + # 函数：jwt.encode() + # 参数说明： + # payload: JWT载荷数据 + # key: 加密密钥（盐） + # algorithm: 加密算法（HS256） + # headers: JWT头部信息 + token = jwt.encode( + payload=payload_copy, + key=JWT_SALT, + algorithm=JWT_ALGORITHM, + headers=headers + ) + + # 记录日志（可选，生产环境建议使用DEBUG级别） + logger.debug(f"生成JWT令牌成功，过期时间：{timeout}分钟") + + return token + + except Exception as e: + # 捕获并记录生成令牌时的异常 + logger.error(f"生成JWT令牌失败: {str(e)}") + raise + + +def parse_payload(token): + """ + 解析验证JWT令牌 + ============================================================= + + 验证传入的JWT令牌，检查签名有效性和过期状态。 + 统一处理各种异常情况，返回标准格式的结果对象。 + + 参数: + token (str): 需要验证的JWT令牌字符串 + + 返回: + dict: 验证结果，包含以下字段： + code (int): 状态码，0-成功，1-失败 + data (dict/None): 验证成功时返回载荷数据，失败时为None + msg (str): 提示信息 + + 使用示例: + >>> result = parse_payload(token) + >>> if result['code'] == 0: + ... user_id = result['data']['userId'] + ... username = result['data']['username'] + ... else: + ... error_msg = result['msg'] + ... # 处理令牌失效，跳转登录页 + + 异常处理说明: + - ExpiredSignatureError: 令牌已过期，需要重新登录 + - DecodeError: 解码失败，令牌格式错误或签名无效 + - InvalidTokenError: 其他JWT相关异常 + """ + # 初始化返回结果 + # 说明：定义统一返回格式，默认成功状态 + result = { + "code": 0, + "data": None, + "msg": "操作成功" + } + + # 参数校验 + if not token: + result['code'] = -1 + result['msg'] = "token不能为空" + return result + + # 检查Token是否在黑名单中 + if is_token_blacklisted(token): + result['code'] = -1 + result['msg'] = "token已失效，请重新登录" + return result + + try: + # 解密验证JWT令牌 + # 函数：jwt.decode() + # 参数说明： + # jwt: 令牌字符串 + # key: 解密密钥（必须与加密时使用的密钥一致） + # algorithms: 允许的算法列表 + # + # 说明：jwt.decode会自动完成以下验证： + # 1. 签名验证 - 确保令牌未被篡改 + # 2. 过期时间验证 - 检查exp字段是否大于当前时间 + # 3. 算法验证 - 确保使用正确的加密算法 + verified_payload = jwt.decode( + token, + JWT_SALT, + algorithms=[JWT_ALGORITHM] + ) + + # 验证成功，将载荷数据存入返回结果 + # 注意：返回的数据中不包含敏感信息（如密码等） + result['data'] = verified_payload + + # 可选：记录验证成功的日志（生产环境建议使用DEBUG级别） + logger.debug(f"JWT令牌验证成功，用户: {verified_payload.get('username', 'unknown')}") + + except exceptions.ExpiredSignatureError as e: + # 令牌已过期异常处理 + # 说明：令牌超过设置的exp时间，需要重新登录获取新令牌 + result['code'] = -1 + result['msg'] = "token已失效，请重新登录" + logger.info(f"JWT令牌已过期: {str(e)}") + + except exceptions.DecodeError as e: + # 令牌解码失败异常处理 + # 说明：令牌格式错误或签名验证失败，可能被篡改 + result['code'] = -1 + result['msg'] = "token认证失败，无效的令牌格式" + logger.warning(f"JWT令牌解码失败: {str(e)}") + + except exceptions.InvalidTokenError as e: + # 无效令牌异常处理 + # 说明：捕获其他所有JWT相关异常，如算法不匹配等 + result['code'] = -1 + result['msg'] = "非法的token，请检查令牌有效性" + logger.warning(f"无效的JWT令牌: {str(e)}") + + except Exception as e: + # 其他未预期的异常处理 + result['code'] = -1 + result['msg'] = "token验证失败，系统异常" + logger.error(f"JWT验证系统异常: {str(e)}") + + return result + + +def refresh_token(token, extend_minutes=None): + """ + 刷新JWT令牌 + ============================================================= + + 验证现有令牌有效性，如果有效则生成新的令牌，延长有效期。 + + 参数: + token (str): 需要刷新的JWT令牌字符串 + extend_minutes (int): 延长的分钟数，默认使用DEFAULT_TIMEOUT_MINUTES + + 返回: + dict: 刷新结果，包含以下字段： + code (int): 状态码，0-成功，-1-失败 + data (dict/None): 成功时包含新令牌，失败时为None + msg (str): 提示信息 + + 使用示例: + >>> result = refresh_token(old_token, extend_minutes=30) + >>> if result['code'] == 0: + ... new_token = result['data']['access_token'] + ... else: + ... # 令牌无效，需要重新登录 + ... error_msg = result['msg'] + + 注意事项: + 1. 令牌只有在有效期内才能刷新 + 2. 刷新后会生成全新的令牌，旧令牌立即失效 + 3. 建议在令牌剩余有效期少于一半时进行刷新 + """ + if extend_minutes is None: + extend_minutes = DEFAULT_TIMEOUT_MINUTES + + # 先验证原令牌 + result = parse_payload(token) + + # 如果验证失败，返回错误信息 + if result['code'] != 0: + return { + "code": -1, + "data": None, + "msg": f"令牌无效，无法刷新: {result['msg']}" + } + + try: + # 获取原载荷数据（不包含exp、iat等JWT标准字段） + payload = result['data'] + + # 移除JWT标准字段，避免重复添加 + # 注意：保留原始业务数据，移除系统字段 + clean_payload = {k: v for k, v in payload.items() + if k not in ['exp', 'iat', 'nbf']} + + # 生成新令牌 + new_token = create_token(clean_payload, timeout=extend_minutes) + + logger.info(f"JWT令牌刷新成功，延长{extend_minutes}分钟") + + return { + "code": 0, + "data": { + "access_token": new_token, + "expires_in": extend_minutes * 60 # 转换为秒 + }, + "msg": "令牌刷新成功" + } + + except Exception as e: + logger.error(f"刷新JWT令牌失败: {str(e)}") + return { + "code": -1, + "data": None, + "msg": "令牌刷新失败，请重新登录" + } + + +def get_token_expiration(token): + """ + 获取JWT令牌的过期时间 + + 验证签名后读取过期时间，确保token未被篡改。 + 不过验证过期状态（verify_exp=False），以便调用方 + 读取已过期token的过期时间用于剩余时间计算等场景。 + + Args: + token (str): JWT令牌字符串 + + Returns: + datetime/None: 令牌过期时间（UTC），解析失败返回None + """ + try: + # 验证签名确保token未被篡改，但不验证过期状态 + payload = jwt.decode( + token, + JWT_SALT, + algorithms=[JWT_ALGORITHM], + options={"verify_exp": False} + ) + + exp_timestamp = payload.get('exp') + if exp_timestamp: + return datetime.datetime.fromtimestamp(exp_timestamp, tz=datetime.timezone.utc) + return None + + except Exception as e: + logger.warning(f"获取JWT过期时间失败: {str(e)}") + return None + + +def get_access_token(request): + """ + 从请求头中提取 access_token + + Args: + request: Django 请求对象 + + Returns: + tuple: (success, token, message) + """ + try: + auth_header = request.META.get('HTTP_AUTHORIZATION', '') + if not auth_header: + return False, None, '缺少 Authorization 头' + + # Bearer 认证方案前缀大小写不敏感（RFC 7235），兼容 'Bearer'/'bearer'/'BEARER' + stripped = auth_header.strip() + if stripped.lower().startswith('bearer '): + access_token = stripped[7:].strip() + else: + access_token = stripped + if not access_token: + return False, None, 'token 为空' + + return True, access_token, 'success' + + except Exception as e: + logger.error(f"获取 access_token 异常: {str(e)}") + return False, None, f'token 提取异常: {str(e)}' + + +def parse_token(request): + """ + 解析请求头中的token，返回解析结果 + + Args: + request: Django 请求对象 + + Returns: + dict: 包含解析结果和用户信息的字典 + """ + result = { + 'code': -1, + 'msg': 'token无效', + 'data': { + 'userId': 0, + 'username': None + } + } + + try: + success, access_token, msg = get_access_token(request) + if not success: + result['msg'] = msg + return result + + parse_result = parse_payload(access_token) + + if not isinstance(parse_result, dict): + result['msg'] = '解析结果格式错误' + return result + + code = parse_result.get('code', 1) + if code != 0: + result['msg'] = parse_result.get('msg', 'token验证失败') + return result + + data = parse_result.get('data', {}) + if not isinstance(data, dict): + result['msg'] = '数据格式错误' + return result + + result['code'] = 0 + result['msg'] = 'success' + result['data']['userId'] = data.get('userId', 0) + result['data']['username'] = data.get('username', '') + result['data']['realname'] = data.get('realname', '') + + except Exception as e: + result['msg'] = f'token解析异常: {str(e)}' + + return result + + +# ============================================================= +# Token黑名单功能 +# ============================================================= + +# 黑名单Key前缀，用于Redis中存储被注销的Token +_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:' + +# 黑名单检查告警节流：首次失败记 ERROR，随后每 _BLACKLIST_ALERT_INTERVAL 秒至多一条 ERROR，期间重复失败降为 DEBUG +_BLACKLIST_ALERT_INTERVAL = 60.0 +_blacklist_check_last_alert = 0.0 + + +def _log_blacklist_check_alert(error: Exception) -> None: + """节流记录黑名单检查失败告警，避免 Redis 故障期间每个请求刷一条 ERROR""" + global _blacklist_check_last_alert + now = time.monotonic() + if now - _blacklist_check_last_alert >= _BLACKLIST_ALERT_INTERVAL: + _blacklist_check_last_alert = now + logger.error( + "Token黑名单检查异常，已按 fail-open 放行（已注销的令牌可能被误放行）: %s", error) + else: + logger.debug("Token黑名单检查异常（节流中，跳过 ERROR）: %s", error) + + +def _token_fingerprint(token: str) -> str: + """ + 生成Token指纹（SHA256摘要） + + 避免将完整Token存入Redis，仅存储摘要值，减少泄露风险。 + + Args: + token: JWT令牌字符串 + + Returns: + str: SHA256十六进制摘要 + """ + return hashlib.sha256(token.encode('utf-8')).hexdigest() + + +def add_token_to_blacklist(token: str) -> bool: + """ + 将Token加入黑名单 + + 退出登录时调用，使Token立即失效。黑名单条目的TTL与Token剩余有效期一致， + Token自然过期后黑名单条目自动清除，避免Redis内存无限增长。 + + Args: + token: JWT令牌字符串 + + Returns: + bool: 是否成功加入黑名单 + """ + try: + from django_redis import get_redis_connection + redis = get_redis_connection("default") + + # 获取Token剩余有效期 + expiration = get_token_expiration(token) + if expiration is None: + # 无法解析过期时间，使用默认过期时间作为TTL + ttl_seconds = DEFAULT_TIMEOUT_MINUTES * 60 + else: + ttl_seconds = int((expiration - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds()) + if ttl_seconds <= 0: + # Token已自然过期，无需加入黑名单 + return True + + fingerprint = _token_fingerprint(token) + key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}' + redis.set(key, '1', ex=ttl_seconds) + + logger.info(f"Token已加入黑名单，TTL={ttl_seconds}秒") + return True + + except Exception as e: + logger.error(f"Token加入黑名单失败: {str(e)}") + return False + + +def is_token_blacklisted(token: str) -> bool: + """ + 检查Token是否在黑名单中 + + 在登录验证中间件中调用，拒绝已被注销的Token。 + + Args: + token: JWT令牌字符串 + + Returns: + bool: True-已被注销（在黑名单中），False-有效 + """ + try: + from django_redis import get_redis_connection + redis = get_redis_connection("default") + + fingerprint = _token_fingerprint(token) + key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}' + return redis.exists(key) > 0 + + except Exception as e: + # fail-open 权衡：Redis 不可用时放行，避免全部请求被拒；代价是已注销的令牌可能被放行。 + # 属安全降级，需告警；节流记录避免 Redis 故障期间日志刷屏。 + _log_blacklist_check_alert(e) + return False + + +def check_token_blacklist(token: str) -> bool: + """ + 检查Token是否在黑名单中（用于中间件） + + Args: + token: JWT令牌字符串 + + Returns: + bool: True-在黑名单中（应拒绝），False-不在黑名单中（可继续验证） + """ + return is_token_blacklisted(token) diff --git a/utils/password.py b/utils/password.py new file mode 100644 index 0000000..4c408a8 --- /dev/null +++ b/utils/password.py @@ -0,0 +1,75 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +# +====================================================================== +# | 模块: 密码加密工具 +# | 说明: 基于 bcrypt 的双重加盐方案 +# +====================================================================== + +"""密码加密工具：基于 bcrypt 的双重加盐方案。""" + +import secrets + +import bcrypt + +# bcrypt 工作因子，值越大计算越慢越安全，默认12（约250ms） +BCRYPT_ROUNDS = 12 + + +# ============================================================ +# 密码工具函数 +# ============================================================ +def generate_salt(length: int = 10) -> str: + """生成随机盐值""" + return secrets.token_urlsafe(length)[:length] + + +def encrypt_password(password: str, salt: str) -> str: + """使用 bcrypt 加密密码（外部 salt 双重加盐）""" + combined = (password + salt).encode('utf-8') + hashed = bcrypt.hashpw(combined, bcrypt.gensalt(rounds=BCRYPT_ROUNDS)) + return hashed.decode('utf-8') + + +def verify_password(password: str, salt: str, hashed: str) -> bool: + """验证密码是否正确""" + combined = (password + salt).encode('utf-8') + return bcrypt.checkpw(combined, hashed.encode('utf-8')) diff --git a/utils/regular.py b/utils/regular.py new file mode 100644 index 0000000..aba11c0 --- /dev/null +++ b/utils/regular.py @@ -0,0 +1,117 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +""" +表单错误处理工具模块 +============================================================= + +该模块提供统一的表单验证错误信息提取功能。 +用于处理Django表单验证失败时的错误信息，将复杂的错误数据结构 +转换为简洁易读的字符串格式，便于前端展示和日志记录。 + +主要功能： +1. 提取表单验证错误信息 +2. 将错误列表转换为字符串格式 +3. 支持多字段错误信息的合并展示 + +所属包： utils +所在文件： utils/form_errors.py +依赖模块： django.forms - Django表单模块 +""" + + +def get_err(form): + """ + 获取表单错误文本 + ============================================================= + + 从Django表单对象中提取所有错误信息，并将其合并为单个字符串。 + 主要用于表单验证失败时，向前端返回统一的错误提示。 + + 参数: + form (django.forms.Form): Django表单实例，包含验证后的错误信息 + + 返回: + str: 合并后的错误信息字符串，多个错误用'/'分隔 + 如果没有错误，返回空字符串 + + 使用示例: + >>> form = UserForm(request.POST) + >>> if not form.is_valid(): + ... error_msg = get_err(form) + ... return R.failed(msg=error_msg) + + 处理流程: + 1. 从form.errors中获取JSON格式的错误数据 + 2. 遍历每个字段的错误信息 + 3. 提取每个错误的具体消息内容 + 4. 将所有错误消息合并为字符串 + + 错误格式说明: + form.errors.get_json_data() 返回格式示例: + { + 'username': [{'message': '用户名不能为空', 'code': 'required'}], + 'password': [{'message': '密码长度至少6位', 'code': 'min_length'}] + } + """ + + # 初始化错误列表 + # 说明：用于收集所有字段的错误消息 + error_list = [] + + # 遍历所有字段的错误信息 + # 方法：form.errors.get_json_data() + # 说明：获取表单错误的JSON表示，包含每个字段的所有错误详情 + for item in form.errors.get_json_data().values(): + # 提取错误消息 + # 说明：每个字段可能有多个错误，取第一个错误的message字段 + # item[0]：获取该字段的第一个错误对象 + # get('message')：从错误对象中获取具体的错误消息文本 + error_list.append(item[0].get('message')) + + # 合并错误字符串 + # 方法：'/'.join(error_list) + # 说明：将所有错误消息用'/'连接成一个字符串 + # 例如：['用户名不能为空', '密码不能为空'] -> '用户名不能为空/密码不能为空' + err_str = '/'.join(error_list) + + # 返回合并后的错误字符串 + return err_str diff --git a/utils/security.py b/utils/security.py new file mode 100644 index 0000000..40f5440 --- /dev/null +++ b/utils/security.py @@ -0,0 +1,149 @@ +# +====================================================================== +# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] +# +---------------------------------------------------------------------- +# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 +# +---------------------------------------------------------------------- +# | 开源协议: Apache-2.0 +# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 +# +---------------------------------------------------------------------- +# | 官方网站: https://www.djangoadmin.cn +# +---------------------------------------------------------------------- +# | 研发团队: 小蚂蚁云团队 荣誉出品 +# +====================================================================== +# | 版权和免责声明: +# | 一、知识产权声明 +# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 +# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， +# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 +# | 二、使用限制与禁止条款 +# | 您理解并同意，禁止将本产品用于以下任何行为： +# | 1. 任何违反中华人民共和国法律法规的用途； +# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； +# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 +# | 三、责任豁免 +# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 +# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 +# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 +# | 四、违约与法律责任 +# | 如发现任何违反本声明的行为，本团队有权： +# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； +# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； +# | 3. 保留进一步采取刑事控告等法律措施的权利。 +# | 五、许可范围说明 +# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 +# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 +# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 +# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 +# | 成的一切经济损失及商誉损害。 +# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 +# | 许可协议》附件为准。 +# +====================================================================== + +""" +用户安全工具模块 +============================================================= + +该模块提供了用户认证和安全相关的工具函数。 +用于快速获取当前登录用户的信息，简化业务代码中的用户信息获取操作。 +""" +import logging + +from utils.jwt import parse_token + + +# ============================= 用户信息获取工具 ============================= + + +def get_user_id(request): + """ + 获取当前登录用户ID + ============================================================= + + 快速获取当前登录用户的ID，用于需要用户ID的业务场景。 + 基于parse_token函数实现，简化调用方式。 + + 使用场景： + - 记录操作日志时需要记录操作人ID + - 数据查询时进行用户权限过滤 + - 创建数据时自动关联当前用户 + + 参数: + request (HttpRequest): Django请求对象，包含请求头中的token信息 + + 返回: + int: 当前登录用户的ID，若token无效或用户不存在则返回0 + + 异常: + 可能抛出parse_token函数产生的异常（如token过期、token无效等） + 建议调用方自行处理异常 + + 示例: + >>> user_id = get_user_id(request) + >>> if user_id: + ... # 执行需要用户ID的操作 + ... create_log(user_id, "操作内容") + """ + try: + # 解析请求中的token，获取用户信息 + result = parse_token(request) + + # 安全获取用户ID，避免KeyError + if result and isinstance(result, dict): + data = result.get('data', {}) + if isinstance(data, dict): + return data.get('userId', 0) + + return 0 + except Exception as e: + # 记录异常日志（可根据需要添加日志记录） + # logger.error(f"获取用户ID失败: {str(e)}") + return 0 + + +def get_username(request): + """ + 获取当前登录用户名称 + ============================================================= + + 快速获取当前登录用户的真实姓名，用于显示用户信息。 + 基于parse_token函数实现，简化调用方式。 + + 使用场景： + - 在页面中显示当前登录用户名称 + - 记录操作日志时显示操作人姓名 + - 发送通知时获取用户名称 + + 参数: + request (HttpRequest): Django请求对象，包含请求头中的token信息 + + 返回: + str: 当前登录用户的真实姓名（realname），若token无效或用户不存在则返回空字符串 + + 异常: + 可能抛出parse_token函数产生的异常（如token过期、token无效等） + 建议调用方自行处理异常 + + 示例: + >>> username = get_username(request) + >>> if username: + ... print(f"当前用户: {username}") + ... else: + ... print("未登录或token无效") + """ + try: + # 解析请求中的token，获取用户信息 + result = parse_token(request) + + # 安全获取用户真实姓名，避免KeyError和None值 + if result and isinstance(result, dict): + data = result.get('data', {}) + if isinstance(data, dict): + username = data.get('realname') + # 如果realname为None或空字符串，返回空字符串 + return username if username else '' + + return '' + except Exception as e: + # 记录异常日志（可根据需要添加日志记录） + logging.error(f"获取用户名失败: {str(e)}") + return ''