wehub-resource-sync
262 lines
10 KiB
Python
262 lines
10 KiB
Python
"""
|
|
路径管理器
|
|
专门管理项目相关的路径操作
|
|
"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PathManager:
|
|
"""路径管理器，负责项目路径的统一管理"""
|
|
|
|
def __init__(self, project_id: str, base_dir: str = "data/projects"):
|
|
self.project_id = project_id
|
|
# 使用绝对路径
|
|
project_root = Path(__file__).parent.parent.parent
|
|
self.base_dir = project_root / base_dir
|
|
self.project_dir = self.base_dir / project_id
|
|
|
|
# 定义项目目录结构
|
|
self.directory_structure = {
|
|
"project_dir": self.project_dir,
|
|
"metadata_dir": self.project_dir / "metadata",
|
|
"raw_dir": self.project_dir / "raw",
|
|
"outputs_dir": self.project_dir / "outputs",
|
|
"logs_dir": self.project_dir / "logs",
|
|
"backups_dir": self.project_dir / "backups",
|
|
"temp_dir": self.project_dir / "temp"
|
|
}
|
|
|
|
# 确保目录结构存在
|
|
self.ensure_directories()
|
|
|
|
def ensure_directories(self):
|
|
"""确保所有必要的目录存在"""
|
|
for dir_name, dir_path in self.directory_structure.items():
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
logger.debug(f"确保目录存在: {dir_name} -> {dir_path}")
|
|
|
|
def get_project_paths(self) -> Dict[str, Path]:
|
|
"""获取项目相关路径"""
|
|
return self.directory_structure.copy()
|
|
|
|
def get_step_paths(self, step_name: str) -> Dict[str, Path]:
|
|
"""
|
|
获取步骤相关的路径
|
|
|
|
Args:
|
|
step_name: 步骤名称
|
|
|
|
Returns:
|
|
步骤相关路径
|
|
"""
|
|
metadata_dir = self.directory_structure["metadata_dir"]
|
|
|
|
return {
|
|
"input_path": metadata_dir / f"{step_name}_input.json",
|
|
"output_path": metadata_dir / f"{step_name}_output.json",
|
|
"intermediate_dir": metadata_dir / f"{step_name}_intermediate",
|
|
"log_path": self.directory_structure["logs_dir"] / f"{step_name}.log"
|
|
}
|
|
|
|
def get_step_input_path(self, step_name: str) -> Path:
|
|
"""获取步骤输入文件路径"""
|
|
return self.get_step_paths(step_name)["input_path"]
|
|
|
|
def get_step_output_path(self, step_name: str) -> Path:
|
|
"""获取步骤输出文件路径"""
|
|
return self.get_step_paths(step_name)["output_path"]
|
|
|
|
def get_step_intermediate_dir(self, step_name: str) -> Path:
|
|
"""获取步骤中间文件目录"""
|
|
return self.get_step_paths(step_name)["intermediate_dir"]
|
|
|
|
def get_step_log_path(self, step_name: str) -> Path:
|
|
"""获取步骤日志文件路径"""
|
|
return self.get_step_paths(step_name)["log_path"]
|
|
|
|
def get_backup_path(self, filename: str) -> Path:
|
|
"""获取备份文件路径"""
|
|
return self.directory_structure["backups_dir"] / filename
|
|
|
|
def get_temp_path(self, filename: str) -> Path:
|
|
"""获取临时文件路径"""
|
|
return self.directory_structure["temp_dir"] / filename
|
|
|
|
def get_config_path(self) -> Path:
|
|
"""获取配置文件路径"""
|
|
return self.project_dir / "config.yaml"
|
|
|
|
def get_srt_path(self) -> Path:
|
|
"""获取SRT文件路径"""
|
|
# 尝试从项目配置中获取SRT文件名
|
|
try:
|
|
from .config_manager import ProjectConfigManager
|
|
config_manager = ProjectConfigManager(self.project_id)
|
|
project_config = config_manager.get_project_config()
|
|
|
|
if project_config and "processing_config" in project_config:
|
|
srt_file = project_config["processing_config"].get("srt_file")
|
|
if srt_file:
|
|
return self.directory_structure["raw_dir"] / srt_file
|
|
|
|
# 如果配置中没有，尝试查找raw目录中的SRT文件
|
|
raw_dir = self.directory_structure["raw_dir"]
|
|
srt_files = list(raw_dir.glob("*.srt"))
|
|
if srt_files:
|
|
return srt_files[0]
|
|
|
|
return raw_dir / "transcript.srt"
|
|
except Exception as e:
|
|
logger.warning(f"获取SRT路径失败: {e}")
|
|
return self.directory_structure["raw_dir"] / "transcript.srt"
|
|
|
|
def get_video_path(self) -> Path:
|
|
"""获取视频文件路径"""
|
|
# 尝试从项目配置中获取视频文件名
|
|
try:
|
|
from .config_manager import ProjectConfigManager
|
|
config_manager = ProjectConfigManager(self.project_id)
|
|
project_config = config_manager.get_project_config()
|
|
|
|
if project_config and "processing_config" in project_config:
|
|
video_file = project_config["processing_config"].get("video_file")
|
|
if video_file:
|
|
return self.directory_structure["raw_dir"] / video_file
|
|
|
|
# 如果配置中没有，尝试查找raw目录中的视频文件
|
|
raw_dir = self.directory_structure["raw_dir"]
|
|
video_extensions = [".mp4", ".avi", ".mov", ".mkv", ".flv"]
|
|
for ext in video_extensions:
|
|
video_files = list(raw_dir.glob(f"*{ext}"))
|
|
if video_files:
|
|
return video_files[0]
|
|
|
|
return None
|
|
except Exception as e:
|
|
logger.warning(f"获取视频路径失败: {e}")
|
|
return None
|
|
|
|
if project_config and "srt_file" in project_config:
|
|
srt_filename = project_config["srt_file"]
|
|
return self.directory_structure["raw_dir"] / srt_filename
|
|
except Exception as e:
|
|
logger.warning(f"无法从项目配置获取SRT文件名: {e}")
|
|
|
|
# 回退到默认文件名
|
|
return self.directory_structure["raw_dir"] / "transcript.srt"
|
|
|
|
def get_prompt_dir(self) -> Path:
|
|
"""获取prompt目录路径"""
|
|
# 使用绝对路径指向项目根目录的prompt文件夹
|
|
project_root = Path(__file__).parent.parent.parent
|
|
return project_root / "prompt"
|
|
|
|
def create_step_directories(self, step_name: str):
|
|
"""为特定步骤创建必要的目录"""
|
|
step_paths = self.get_step_paths(step_name)
|
|
|
|
# 创建中间文件目录
|
|
step_paths["intermediate_dir"].mkdir(parents=True, exist_ok=True)
|
|
|
|
# 确保日志目录存在
|
|
step_paths["log_path"].parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
logger.debug(f"为步骤 {step_name} 创建目录结构")
|
|
|
|
def cleanup_step_files(self, step_name: str, keep_output: bool = True):
|
|
"""
|
|
清理步骤的临时文件
|
|
|
|
Args:
|
|
step_name: 步骤名称
|
|
keep_output: 是否保留输出文件
|
|
"""
|
|
step_paths = self.get_step_paths(step_name)
|
|
|
|
# 清理中间文件目录
|
|
if step_paths["intermediate_dir"].exists():
|
|
import shutil
|
|
shutil.rmtree(step_paths["intermediate_dir"])
|
|
logger.info(f"已清理步骤 {step_name} 的中间文件")
|
|
|
|
# 清理输入文件（可选）
|
|
if step_paths["input_path"].exists():
|
|
step_paths["input_path"].unlink()
|
|
logger.debug(f"已清理步骤 {step_name} 的输入文件")
|
|
|
|
# 清理输出文件（可选）
|
|
if not keep_output and step_paths["output_path"].exists():
|
|
step_paths["output_path"].unlink()
|
|
logger.info(f"已清理步骤 {step_name} 的输出文件")
|
|
|
|
def get_directory_size(self, dir_path: Path) -> int:
|
|
"""获取目录大小（字节）"""
|
|
total_size = 0
|
|
try:
|
|
for file_path in dir_path.rglob("*"):
|
|
if file_path.is_file():
|
|
total_size += file_path.stat().st_size
|
|
except Exception as e:
|
|
logger.warning(f"计算目录大小时出错: {dir_path}, 错误: {e}")
|
|
|
|
return total_size
|
|
|
|
def get_project_size_info(self) -> Dict[str, Any]:
|
|
"""获取项目大小信息"""
|
|
size_info = {}
|
|
|
|
for dir_name, dir_path in self.directory_structure.items():
|
|
if dir_path.exists():
|
|
size_info[dir_name] = {
|
|
"path": str(dir_path),
|
|
"size_bytes": self.get_directory_size(dir_path),
|
|
"file_count": len(list(dir_path.rglob("*"))) if dir_path.is_dir() else 0
|
|
}
|
|
else:
|
|
size_info[dir_name] = {
|
|
"path": str(dir_path),
|
|
"size_bytes": 0,
|
|
"file_count": 0,
|
|
"exists": False
|
|
}
|
|
|
|
return size_info
|
|
|
|
def validate_paths(self) -> List[str]:
|
|
"""验证路径的有效性"""
|
|
errors = []
|
|
|
|
# 检查项目目录是否可写
|
|
if not self.project_dir.exists():
|
|
try:
|
|
self.project_dir.mkdir(parents=True, exist_ok=True)
|
|
except Exception as e:
|
|
errors.append(f"无法创建项目目录: {self.project_dir}, 错误: {e}")
|
|
|
|
# 检查各个子目录
|
|
for dir_name, dir_path in self.directory_structure.items():
|
|
if not dir_path.exists():
|
|
try:
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
except Exception as e:
|
|
errors.append(f"无法创建目录 {dir_name}: {dir_path}, 错误: {e}")
|
|
elif not dir_path.is_dir():
|
|
errors.append(f"路径不是目录: {dir_name} -> {dir_path}")
|
|
|
|
return errors
|
|
|
|
def get_relative_path(self, absolute_path: Path) -> str:
|
|
"""获取相对于项目目录的路径"""
|
|
try:
|
|
return str(absolute_path.relative_to(self.project_dir))
|
|
except ValueError:
|
|
return str(absolute_path)
|
|
|
|
def get_absolute_path(self, relative_path: str) -> Path:
|
|
"""获取绝对路径"""
|
|
return self.project_dir / relative_path |