# +====================================================================== # | 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