fix
This commit is contained in:
root
@@ -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()
|
||||
Reference in New Issue
Block a user