From e377ae46ba7f61754cba4fdd75d25f3f5e717afb Mon Sep 17 00:00:00 2001 From: root Date: Fri, 11 Jul 2025 01:02:25 +0800 Subject: [PATCH] fix --- python_core/services/project_manager.py | 426 ------------------------ 1 file changed, 426 deletions(-) delete mode 100644 python_core/services/project_manager.py diff --git a/python_core/services/project_manager.py b/python_core/services/project_manager.py deleted file mode 100644 index 79b0103..0000000 --- a/python_core/services/project_manager.py +++ /dev/null @@ -1,426 +0,0 @@ -#!/usr/bin/env python3 -""" -Project Manager Service -项目管理服务 - -Handles project creation, loading, saving, and management. -""" - -import argparse -import json -import sys -from pathlib import Path -from typing import Dict, Any, List, Optional -from datetime import datetime -import uuid - -import sys -import os - -from ..config import settings -from ..utils import setup_logger, ensure_directory, get_unique_filename - -logger = setup_logger(__name__) - - -class ProjectManager: - """Project management service.""" - - def __init__(self): - self.projects_dir = settings.projects_dir - self.temp_dir = settings.temp_dir - - def create_project(self, name: str, description: str = "") -> Dict[str, Any]: - """ - Create a new project. - - Args: - name: Project name - description: Project description - - Returns: - Dictionary with project information - """ - try: - # Create project directory - project_path = self.projects_dir / name - ensure_directory(project_path) - - # Create project structure - (project_path / "assets").mkdir(exist_ok=True) - (project_path / "exports").mkdir(exist_ok=True) - (project_path / "temp").mkdir(exist_ok=True) - - # Create project info - project_info = { - "id": str(uuid.uuid4()), - "name": name, - "description": description, - "path": str(project_path), - "created_at": datetime.now().isoformat(), - "modified_at": datetime.now().isoformat(), - "version": "1.0", - "settings": { - "resolution": settings.max_video_resolution, - "fps": settings.default_fps, - "video_codec": settings.default_video_codec, - "audio_codec": settings.default_audio_codec - }, - "video_tracks": [], - "audio_tracks": [], - "timeline": { - "duration": 0.0, - "zoom": 1.0, - "position": 0.0 - } - } - - # Save project file - project_file = project_path / f"{name}.mixvideo" - with open(project_file, 'w', encoding='utf-8') as f: - json.dump(project_info, f, indent=2, ensure_ascii=False) - - logger.info(f"Created project: {name} at {project_path}") - - return { - "status": "success", - "project": project_info - } - - except Exception as e: - logger.error(f"Failed to create project: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - def load_project(self, project_path: str) -> Dict[str, Any]: - """ - Load an existing project. - - Args: - project_path: Path to project file or directory - - Returns: - Dictionary with project information - """ - try: - path = Path(project_path) - - # If it's a directory, look for .mixvideo file - if path.is_dir(): - mixvideo_files = list(path.glob("*.mixvideo")) - if not mixvideo_files: - raise FileNotFoundError(f"No .mixvideo file found in {path}") - project_file = mixvideo_files[0] - else: - project_file = path - - if not project_file.exists(): - raise FileNotFoundError(f"Project file not found: {project_file}") - - # Load project data - with open(project_file, 'r', encoding='utf-8') as f: - project_info = json.load(f) - - # Update modified time - project_info["modified_at"] = datetime.now().isoformat() - - logger.info(f"Loaded project: {project_info['name']}") - - return { - "status": "success", - "project": project_info - } - - except Exception as e: - logger.error(f"Failed to load project: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - def save_project(self, project_info: Dict[str, Any]) -> Dict[str, Any]: - """ - Save project to file. - - Args: - project_info: Project information dictionary - - Returns: - Dictionary with save result - """ - try: - project_path = Path(project_info["path"]) - project_name = project_info["name"] - - # Update modified time - project_info["modified_at"] = datetime.now().isoformat() - - # Save project file - project_file = project_path / f"{project_name}.mixvideo" - with open(project_file, 'w', encoding='utf-8') as f: - json.dump(project_info, f, indent=2, ensure_ascii=False) - - logger.info(f"Saved project: {project_name}") - - return { - "status": "success", - "message": f"Project saved to {project_file}" - } - - except Exception as e: - logger.error(f"Failed to save project: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - def list_projects(self) -> Dict[str, Any]: - """ - List all projects in the projects directory. - - Returns: - Dictionary with list of projects - """ - try: - projects = [] - - for project_file in self.projects_dir.glob("**/*.mixvideo"): - try: - with open(project_file, 'r', encoding='utf-8') as f: - project_info = json.load(f) - - # Add basic project info - projects.append({ - "id": project_info.get("id"), - "name": project_info.get("name"), - "description": project_info.get("description", ""), - "path": str(project_file.parent), - "created_at": project_info.get("created_at"), - "modified_at": project_info.get("modified_at"), - "file_size": project_file.stat().st_size - }) - - except Exception as e: - logger.warning(f"Failed to read project file {project_file}: {str(e)}") - continue - - # Sort by modified time (newest first) - projects.sort(key=lambda x: x.get("modified_at", ""), reverse=True) - - return { - "status": "success", - "projects": projects, - "count": len(projects) - } - - except Exception as e: - logger.error(f"Failed to list projects: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - def delete_project(self, project_path: str) -> Dict[str, Any]: - """ - Delete a project. - - Args: - project_path: Path to project directory - - Returns: - Dictionary with deletion result - """ - try: - import shutil - - path = Path(project_path) - if not path.exists(): - raise FileNotFoundError(f"Project path not found: {path}") - - if path.is_dir(): - shutil.rmtree(path) - else: - path.unlink() - - logger.info(f"Deleted project: {path}") - - return { - "status": "success", - "message": f"Project deleted: {path}" - } - - except Exception as e: - logger.error(f"Failed to delete project: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - def add_video_track(self, project_info: Dict[str, Any], file_path: str, - start_time: float = 0.0, duration: Optional[float] = None) -> Dict[str, Any]: - """ - Add a video track to the project. - - Args: - project_info: Project information - file_path: Path to video file - start_time: Start time in timeline - duration: Duration (auto-detect if None) - - Returns: - Updated project information - """ - try: - from utils import get_file_info - - file_info = get_file_info(file_path) - if "error" in file_info: - raise ValueError(f"Invalid video file: {file_info['error']}") - - track_id = str(uuid.uuid4()) - video_track = { - "id": track_id, - "name": Path(file_path).stem, - "file_path": file_path, - "start_time": start_time, - "duration": duration or file_info.get("duration", 0.0), - "file_info": file_info, - "effects": [], - "volume": 1.0, - "enabled": True - } - - project_info["video_tracks"].append(video_track) - project_info["modified_at"] = datetime.now().isoformat() - - logger.info(f"Added video track: {video_track['name']}") - - return { - "status": "success", - "project": project_info, - "track_id": track_id - } - - except Exception as e: - logger.error(f"Failed to add video track: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - def add_audio_track(self, project_info: Dict[str, Any], file_path: str, - start_time: float = 0.0, duration: Optional[float] = None) -> Dict[str, Any]: - """ - Add an audio track to the project. - - Args: - project_info: Project information - file_path: Path to audio file - start_time: Start time in timeline - duration: Duration (auto-detect if None) - - Returns: - Updated project information - """ - try: - from utils import get_file_info - - file_info = get_file_info(file_path) - if "error" in file_info: - raise ValueError(f"Invalid audio file: {file_info['error']}") - - track_id = str(uuid.uuid4()) - audio_track = { - "id": track_id, - "name": Path(file_path).stem, - "file_path": file_path, - "start_time": start_time, - "duration": duration or file_info.get("duration", 0.0), - "file_info": file_info, - "effects": [], - "volume": 1.0, - "enabled": True - } - - project_info["audio_tracks"].append(audio_track) - project_info["modified_at"] = datetime.now().isoformat() - - logger.info(f"Added audio track: {audio_track['name']}") - - return { - "status": "success", - "project": project_info, - "track_id": track_id - } - - except Exception as e: - logger.error(f"Failed to add audio track: {str(e)}") - return { - "status": "error", - "error": str(e) - } - - -def main(): - """Command line interface for project management.""" - parser = argparse.ArgumentParser(description="Project Manager") - parser.add_argument("--action", required=True, - choices=["create", "load", "save", "list", "delete", "get_info"], - help="Action to perform") - parser.add_argument("--name", help="Project name") - parser.add_argument("--path", help="Project path") - parser.add_argument("--data", help="Project data as JSON string") - parser.add_argument("--description", help="Project description") - - args = parser.parse_args() - - try: - manager = ProjectManager() - - if args.action == "create": - if not args.name: - raise ValueError("Project name is required for create action") - result = manager.create_project(args.name, args.description or "") - - elif args.action == "load": - if not args.path: - raise ValueError("Project path is required for load action") - result = manager.load_project(args.path) - - elif args.action == "save": - if not args.data: - raise ValueError("Project data is required for save action") - project_data = json.loads(args.data) - result = manager.save_project(project_data) - - elif args.action == "list": - result = manager.list_projects() - - elif args.action == "delete": - if not args.path: - raise ValueError("Project path is required for delete action") - result = manager.delete_project(args.path) - - elif args.action == "get_info": - if not args.path: - raise ValueError("Project path is required for get_info action") - result = manager.load_project(args.path) - - else: - raise ValueError(f"Unknown action: {args.action}") - - print(json.dumps(result)) - - except Exception as e: - error_result = { - "status": "error", - "error": str(e) - } - print(json.dumps(error_result)) - sys.exit(1) - - -if __name__ == "__main__": - main()