""" 路由层通用工具函数 提供文件操作、题目数据获取等通用功能 """ import os from pathlib import Path from flask import jsonify, g # 尝试导入数据库操作 try: from app.models.database.operations import ( get_latest_challenge, get_challenge_by_name, update_challenge, get_challenge_record ) HAS_DB_OPERATIONS = True except ImportError: HAS_DB_OPERATIONS = False print("数据库操作模块导入失败，将使用文件方式读取数据") def get_project_root(): """获取项目根目录""" return Path(__file__).parent.parent.parent.parent def is_binary_file(filename): """ 检查文件是否为二进制文件 Args: filename: 文件名或文件路径 Returns: bool: 是否为二进制文件 """ binary_extensions = { '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.ico', '.svg', '.pdf', '.zip', '.tar', '.gz', '.bz2', '.7z', '.rar', '.exe', '.dll', '.so', '.dylib', '.bin', '.mp3', '.mp4', '.avi', '.mov', '.wav', '.ttf', '.otf', '.woff', '.woff2', '.eot' } ext = os.path.splitext(filename)[1].lower() return ext in binary_extensions def get_category_id_from_challenge(challenge_id): """ 通过 challenge_id 从任务记录或数据库获取 category_id Args: challenge_id: 题目ID Returns: str: category_id，如果找不到则返回 'web'（默认） """ if not challenge_id: return 'web' # 首先从任务记录中查找对应的 category_id（内存中的任务状态） try: from app.routes.generator.tasks import tasks_status, _tasks_lock with _tasks_lock: for task_id, task in tasks_status.items(): if task.get('challenge_id') == challenge_id: category_id = task.get('category_id', 'web') if category_id: return category_id except ImportError: pass # 如果内存中找不到，从数据库读取题目记录 if HAS_DB_OPERATIONS: try: challenge_data = get_challenge_record(int(challenge_id)) if challenge_data: # challenge_type 字段对应 category_id challenge_type = challenge_data.get('challenge_type') if challenge_type: return challenge_type except Exception as e: print(f"从数据库读取题目记录失败: {str(e)}") # 如果都找不到，返回默认值 return 'web' def get_challenge_and_output_dir(challenge_id=None): """ 获取题目数据和输出目录的公共函数 Args: challenge_id: 题目ID，如果为None则获取最新题目 Returns: tuple: (challenge_data, output_dir, category_id) 或 (None, error_response) """ # 从数据库获取题目 if not HAS_DB_OPERATIONS: return None, (jsonify({"status": "error", "message": "数据库操作不可用"}), 500) if challenge_id: challenge_data = get_challenge_record(int(challenge_id)) else: challenge_data = get_latest_challenge() if not challenge_data: return None, (jsonify({"status": "error", "message": "未找到题目数据"}), 404) # 权限检查：只能查看自己的题目或管理员可以查看所有题目 if hasattr(g, 'user') and g.user: if challenge_data.get('user_id') != g.user.id and g.user.role != 'admin': return None, (jsonify({"status": "error", "message": "您没有权限查看此题目"}), 403) else: # 未登录用户不应该能访问到这里（因为有@login_required装饰器） return None, (jsonify({"status": "error", "message": "请先登录"}), 401) # 获取 category_id（从任务记录中查找） challenge_id_int = challenge_data.get('id') category_id = get_category_id_from_challenge(challenge_id_int) if challenge_id_int else 'web' # 将 category_id 添加到 challenge_data 中，方便后续使用 challenge_data['category_id'] = category_id # 获取output_dir并处理路径（支持相对路径和绝对路径） output_dir = challenge_data.get('output_dir') project_root = get_project_root() if not output_dir: return None, (jsonify({"status": "error", "message": "输出目录未设置"}), 404) # 如果是相对路径，转换为绝对路径（相对于项目根目录） original_output_dir = output_dir if not output_dir.startswith('/') and not os.path.isabs(output_dir): # 相对路径，转换为绝对路径（相对于项目根目录） output_dir = str(project_root / output_dir) # 检查路径是否存在，如果不存在，尝试从新的路径结构中查找 if not os.path.exists(output_dir): # 尝试从旧路径中提取题目目录名 old_output_path = Path(original_output_dir) challenge_dir_name = old_output_path.name # 例如: 20251230_163149_SimpleSQL # 尝试在新的路径结构中查找：ge10/{category_id}/output/{challenge_dir_name} new_output_dir = project_root / 'ge10' / category_id / 'output' / challenge_dir_name if new_output_dir.exists(): # 找到了新路径，使用新路径并更新数据库 new_output_dir_str = str(new_output_dir.relative_to(project_root)) output_dir = str(new_output_dir) # 更新数据库中的 output_dir if HAS_DB_OPERATIONS: try: update_challenge(challenge_id_int, {'output_dir': new_output_dir_str}) print(f"已更新题目 {challenge_id_int} 的 output_dir: {original_output_dir} -> {new_output_dir_str}") except Exception as e: print(f"更新数据库 output_dir 失败: {str(e)}") else: # 新旧路径都不存在 return None, (jsonify({ "status": "error", "message": f"输出目录不存在: {output_dir}（已尝试新路径: {new_output_dir}）" }), 404) else: # 旧路径存在，检查是否是旧格式路径（ge10/output/xxx），如果是则尝试迁移到新路径 output_path = Path(output_dir) if 'ge10' in str(output_path.parts): # 检查是否是旧格式：ge10/output/xxx try: parts = output_path.parts ge10_index = None for i, part in enumerate(parts): if part == 'ge10': ge10_index = i break if ge10_index is not None and len(parts) > ge10_index + 2: # 检查是否是旧格式：ge10/output/xxx if parts[ge10_index + 1] == 'output': # 是旧格式，尝试迁移到新路径：ge10/{category_id}/output/xxx challenge_dir_name = parts[-1] new_output_dir = project_root / 'ge10' / category_id / 'output' / challenge_dir_name # 如果新路径不存在，创建并迁移 if not new_output_dir.exists(): try: import shutil new_output_dir.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(output_path), str(new_output_dir)) print(f"已迁移题目目录: {output_path} -> {new_output_dir}") # 更新数据库 new_output_dir_str = str(new_output_dir.relative_to(project_root)) if HAS_DB_OPERATIONS: update_challenge(challenge_id_int, {'output_dir': new_output_dir_str}) output_dir = str(new_output_dir) except Exception as e: print(f"迁移题目目录失败: {str(e)}") # 迁移失败，继续使用旧路径 except Exception as e: print(f"检查旧格式路径失败: {str(e)}") return challenge_data, output_dir, category_id def validate_file_path(base_dir, file_path): """ 验证文件路径是否在基础目录内（防止路径遍历攻击） Args: base_dir: 基础目录 file_path: 要验证的文件路径 Returns: bool: 是否合法 """ try: base = Path(base_dir).resolve() target = (Path(base_dir) / file_path).resolve() return target.is_relative_to(base) except (ValueError, OSError): return False