repo_description: AutoClip: AI-powered video clipping and highlight generation · 一款智能高光提取与剪辑的二创工具
repo_topics: ai-video-editing auto-highlight llm video video-editing
code: """
  项目数据管理器 - 确保多项目数据完全隔离
  """
  import os
  import json
  import shutil
  import logging
  from pathlib import Path
  from typing import Dict, Any, List, Optional
  from datetime import datetime
  import uuid
  
  try:
      from .error_handler import FileIOError, ValidationError, ProcessingError
      from ..config import config_manager
  except ImportError:
      # 独立运行时的导入
      import sys
      from pathlib import Path
      sys.path.append(str(Path(__file__).parent.parent))
      from utils.error_handler import FileIOError, ValidationError, ProcessingError
      from config import ConfigManager
      config_manager = ConfigManager()
  
  logger = logging.getLogger(__name__)
  
  class ProjectManager:
      """项目数据管理器"""
      
      def __init__(self):
          self.config = config_manager
      
      def create_project(self, project_name: Optional[str] = None) -> str:
          """
          创建新项目
          
          Args:
              project_name: 项目名称（可选）
              
          Returns:
              项目ID
          """
          project_id = str(uuid.uuid4())
          project_name = project_name or f"project_{project_id[:8]}"
          
          # 确保项目目录结构存在
          self.config.ensure_project_directories(project_id)
          
          # 创建项目元数据
          project_metadata = {
              "project_id": project_id,
              "project_name": project_name,
              "created_at": datetime.now().isoformat(),
              "status": "created",
              "current_step": 0,
              "total_steps": 6,
              "error_message": None,
              "file_info": {
                  "video_file": None,
                  "srt_file": None,
                  "txt_file": None
              }
          }
          
          # 保存项目元数据
          self._save_project_metadata(project_id, project_metadata)
          
          logger.info(f"创建项目: {project_id} ({project_name})")
          return project_id
      
      def get_project_paths(self, project_id: str) -> Dict[str, Path]:
          """
          获取项目路径配置
          
          Args:
              project_id: 项目ID
              
          Returns:
              项目路径字典
          """
          return self.config.get_project_paths(project_id)
      
      def validate_project_exists(self, project_id: str) -> bool:
          """
          验证项目是否存在
          
          Args:
              project_id: 项目ID
              
          Returns:
              项目是否存在
          """
          paths = self.get_project_paths(project_id)
          return paths["project_base"].exists()
      
      def get_project_metadata(self, project_id: str) -> Dict[str, Any]:
          """
          获取项目元数据
          
          Args:
              project_id: 项目ID
              
          Returns:
              项目元数据
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          metadata_file = self.get_project_paths(project_id)["metadata_dir"] / "project_metadata.json"
          
          if not metadata_file.exists():
              # 如果元数据文件不存在，创建默认元数据
              default_metadata = {
                  "project_id": project_id,
                  "project_name": f"project_{project_id[:8]}",
                  "created_at": datetime.now().isoformat(),
                  "status": "unknown",
                  "current_step": 0,
                  "total_steps": 6,
                  "error_message": None,
                  "file_info": {
                      "video_file": None,
                      "srt_file": None,
                      "txt_file": None
                  }
              }
              self._save_project_metadata(project_id, default_metadata)
              return default_metadata
          
          try:
              with open(metadata_file, 'r', encoding='utf-8') as f:
                  return json.load(f)
          except Exception as e:
              raise FileIOError(f"读取项目元数据失败: {e}")
      
      def update_project_metadata(self, project_id: str, updates: Dict[str, Any]):
          """
          更新项目元数据
          
          Args:
              project_id: 项目ID
              updates: 要更新的字段
          """
          metadata = self.get_project_metadata(project_id)
          metadata.update(updates)
          metadata["updated_at"] = datetime.now().isoformat()
          
          self._save_project_metadata(project_id, metadata)
      
      def _save_project_metadata(self, project_id: str, metadata: Dict[str, Any]) -> None:
          """保存项目元数据"""
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          metadata_file = metadata_dir / "project_metadata.json"
          
          try:
              # 确保metadata目录存在
              metadata_dir.mkdir(parents=True, exist_ok=True)
              
              with open(metadata_file, 'w', encoding='utf-8') as f:
                  json.dump(metadata, f, ensure_ascii=False, indent=2)
          except Exception as e:
              raise FileIOError(f"保存项目元数据失败: {e}")
      
      def save_input_file(self, project_id: str, file_path: Path, file_type: str) -> str:
          """
          保存输入文件到项目目录
          
          Args:
              project_id: 项目ID
              file_path: 源文件路径
              file_type: 文件类型 (video, srt, txt)
              
          Returns:
              保存后的文件路径
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          if not file_path.exists():
              raise FileIOError(f"源文件不存在: {file_path}")
          
          paths = self.get_project_paths(project_id)
          input_dir = paths["input_dir"]
          
          # 确定目标文件名
          if file_type == "video":
              target_name = "input.mp4"
          elif file_type == "srt":
              target_name = "input.srt"
          elif file_type == "txt":
              target_name = "input.txt"
          else:
              raise ValidationError(f"不支持的文件类型: {file_type}")
          
          target_path = input_dir / target_name
          
          try:
              # 复制文件
              shutil.copy2(file_path, target_path)
              
              # 更新项目元数据
              metadata = self.get_project_metadata(project_id)
              metadata["file_info"][f"{file_type}_file"] = str(target_path)
              self._save_project_metadata(project_id, metadata)
              
              logger.info(f"文件已保存到项目 {project_id}: {target_path}")
              return str(target_path)
              
          except Exception as e:
              raise FileIOError(f"保存文件失败: {e}")
      
      def get_input_files(self, project_id: str) -> Dict[str, Optional[Path]]:
          """
          获取项目输入文件
          
          Args:
              project_id: 项目ID
              
          Returns:
              输入文件路径字典
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          project_base = paths["project_base"]
          input_dir = paths["input_dir"]
          
          # 检查两个可能的位置：input子目录和项目根目录
          file_names = ["input.mp4", "input.srt", "input.txt"]
          file_keys = ["video_file", "srt_file", "txt_file"]
          
          files = {}
          for key, name in zip(file_keys, file_names):
              # 优先检查input子目录
              input_path = input_dir / name
              if input_path.exists():
                  files[key] = input_path
              else:
                  # 检查项目根目录
                  root_path = project_base / name
                  if root_path.exists():
                      files[key] = root_path
                  else:
                      files[key] = None
          
          return files
      
      def validate_input_files(self, project_id: str) -> Dict[str, bool]:
          """
          验证项目输入文件
          
          Args:
              project_id: 项目ID
              
          Returns:
              文件验证结果
          """
          files = self.get_input_files(project_id)
          
          validation = {
              "has_video": files["video_file"] is not None,
              "has_srt": files["srt_file"] is not None,
              "has_txt": files["txt_file"] is not None,
              "can_process": files["video_file"] is not None and files["srt_file"] is not None
          }
          
          return validation
      
      def save_processing_result(self, project_id: str, step: int, result: Dict[str, Any]):
          """
          保存处理结果
          
          Args:
              project_id: 项目ID
              step: 处理步骤
              result: 处理结果
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          
          # 确保metadata目录存在
          metadata_dir.mkdir(parents=True, exist_ok=True)
          
          # 保存步骤结果
          step_file = metadata_dir / f"step{step}_result.json"
          
          try:
              with open(step_file, 'w', encoding='utf-8') as f:
                  json.dump(result, f, ensure_ascii=False, indent=2)
              
              # 更新项目状态
              self.update_project_metadata(project_id, {
                  "current_step": step,
                  "status": "processing" if step < 6 else "completed"
              })
              
              logger.info(f"步骤 {step} 结果已保存到项目 {project_id}")
              
          except Exception as e:
              raise FileIOError(f"保存处理结果失败: {e}")
      
      def get_processing_result(self, project_id: str, step: int) -> Optional[Dict[str, Any]]:
          """
          获取处理结果
          
          Args:
              project_id: 项目ID
              step: 处理步骤
              
          Returns:
              处理结果，如果不存在则返回None
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          step_file = metadata_dir / f"step{step}_result.json"
          
          if not step_file.exists():
              return None
          
          try:
              with open(step_file, 'r', encoding='utf-8') as f:
                  return json.load(f)
          except Exception as e:
              raise FileIOError(f"读取处理结果失败: {e}")
      
      def save_clip(self, project_id: str, clip_data: Dict[str, Any], clip_index: int):
          """
          保存视频切片信息
          
          Args:
              project_id: 项目ID
              clip_data: 切片数据
              clip_index: 切片索引
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          
          # 确保metadata目录存在
          metadata_dir.mkdir(parents=True, exist_ok=True)
          
          # 读取现有切片数据
          clips_file = metadata_dir / "clips_metadata.json"
          clips_data = []
          
          if clips_file.exists():
              try:
                  with open(clips_file, 'r', encoding='utf-8') as f:
                      clips_data = json.load(f)
              except Exception:
                  clips_data = []
          
          # 添加新切片
          clip_data["clip_index"] = clip_index
          clip_data["created_at"] = datetime.now().isoformat()
          
          # 确保不重复添加
          existing_indices = [clip["clip_index"] for clip in clips_data]
          if clip_index in existing_indices:
              # 更新现有切片
              for i, clip in enumerate(clips_data):
                  if clip["clip_index"] == clip_index:
                      clips_data[i] = clip_data
                      break
          else:
              # 添加新切片
              clips_data.append(clip_data)
          
          # 保存切片数据
          try:
              with open(clips_file, 'w', encoding='utf-8') as f:
                  json.dump(clips_data, f, ensure_ascii=False, indent=2)
              
              logger.info(f"切片 {clip_index} 已保存到项目 {project_id}")
              
          except Exception as e:
              raise FileIOError(f"保存切片数据失败: {e}")
      
      def get_clips(self, project_id: str) -> List[Dict[str, Any]]:
          """
          获取项目所有切片
          
          Args:
              project_id: 项目ID
              
          Returns:
              切片列表
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          clips_file = metadata_dir / "clips_metadata.json"
          
          if not clips_file.exists():
              return []
          
          try:
              with open(clips_file, 'r', encoding='utf-8') as f:
                  return json.load(f)
          except Exception as e:
              raise FileIOError(f"读取切片数据失败: {e}")
      
      def save_collection(self, project_id: str, collection_data: Dict[str, Any]):
          """
          保存合集信息
          
          Args:
              project_id: 项目ID
              collection_data: 合集数据
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          
          # 确保metadata目录存在
          metadata_dir.mkdir(parents=True, exist_ok=True)
          
          # 读取现有合集数据
          collections_file = metadata_dir / "collections_metadata.json"
          collections_data = []
          
          if collections_file.exists():
              try:
                  with open(collections_file, 'r', encoding='utf-8') as f:
                      collections_data = json.load(f)
              except Exception:
                  collections_data = []
          
          # 添加新合集
          collection_data["created_at"] = datetime.now().isoformat()
          collections_data.append(collection_data)
          
          # 保存合集数据
          try:
              with open(collections_file, 'w', encoding='utf-8') as f:
                  json.dump(collections_data, f, ensure_ascii=False, indent=2)
              
              logger.info(f"合集已保存到项目 {project_id}")
              
          except Exception as e:
              raise FileIOError(f"保存合集数据失败: {e}")
      
      def get_collections(self, project_id: str) -> List[Dict[str, Any]]:
          """
          获取项目所有合集
          
          Args:
              project_id: 项目ID
              
          Returns:
              合集列表
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          paths = self.get_project_paths(project_id)
          metadata_dir = paths["metadata_dir"]
          collections_file = metadata_dir / "collections_metadata.json"
          
          if not collections_file.exists():
              return []
          
          try:
              with open(collections_file, 'r', encoding='utf-8') as f:
                  return json.load(f)
          except Exception as e:
              raise FileIOError(f"读取合集数据失败: {e}")
      
      def list_projects(self) -> List[Dict[str, Any]]:
          """
          列出所有项目
          
          Returns:
              项目列表
          """
          projects = []
          uploads_dir = Path(self.config.settings.uploads_dir)
          
          if not uploads_dir.exists():
              return projects
          
          for project_dir in uploads_dir.iterdir():
              if project_dir.is_dir() and not project_dir.name.startswith('.'):
                  try:
                      metadata = self.get_project_metadata(project_dir.name)
                      projects.append(metadata)
                  except Exception as e:
                      logger.warning(f"读取项目 {project_dir.name} 元数据失败: {e}")
          
          # 按创建时间排序
          projects.sort(key=lambda x: x.get("created_at", ""), reverse=True)
          return projects
      
      def delete_project(self, project_id: str) -> bool:
          """
          删除项目
          
          Args:
              project_id: 项目ID
              
          Returns:
              是否删除成功
          """
          if not self.validate_project_exists(project_id):
              logger.warning(f"项目不存在: {project_id}")
              return False
          
          paths = self.get_project_paths(project_id)
          project_base = paths["project_base"]
          
          try:
              shutil.rmtree(project_base)
              logger.info(f"项目已删除: {project_id}")
              return True
          except Exception as e:
              logger.error(f"删除项目失败: {e}")
              return False
      
      def get_project_summary(self, project_id: str) -> Dict[str, Any]:
          """
          获取项目摘要信息
          
          Args:
              project_id: 项目ID
              
          Returns:
              项目摘要
          """
          if not self.validate_project_exists(project_id):
              raise FileIOError(f"项目不存在: {project_id}")
          
          metadata = self.get_project_metadata(project_id)
          validation = self.validate_input_files(project_id)
          clips = self.get_clips(project_id)
          collections = self.get_collections(project_id)
          
          return {
              "project_info": metadata,
              "file_validation": validation,
              "clips_count": len(clips),
              "collections_count": len(collections),
              "processing_progress": {
                  "current_step": metadata.get("current_step", 0),
                  "total_steps": metadata.get("total_steps", 6),
                  "progress_percentage": (metadata.get("current_step", 0) / metadata.get("total_steps", 6)) * 100
              }
          }
  
  # 全局项目管理器实例
  project_manager = ProjectManager()