diff --git a/python_core/cli/commands/template.py b/python_core/cli/commands/template.py index b51e5a3..e8738c7 100644 --- a/python_core/cli/commands/template.py +++ b/python_core/cli/commands/template.py @@ -12,7 +12,7 @@ from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskPr from rich.live import Live import json -from python_core.services.template_manager import TemplateManager, TemplateInfo +from python_core.services.template_manager_cloud import TemplateManagerCloud, TemplateInfo from python_core.utils.logger import logger console = Console() @@ -42,7 +42,7 @@ def batch_import( raise typer.Exit(1) # 创建模板管理器 - manager = TemplateManager() + manager = TemplateManagerCloud() # 进度回调函数 progress_messages = [] @@ -112,7 +112,7 @@ def list_templates( console.print(f"📋 [bold blue]模板列表[/bold blue]") # 创建模板管理器 - manager = TemplateManager() + manager = TemplateManagerCloud() # 获取模板列表 templates = manager.get_templates() @@ -211,7 +211,7 @@ def get_template( console.print(f"模板ID: {template_id}") # 创建模板管理器 - manager = TemplateManager() + manager = TemplateManagerCloud() # 获取模板 template = manager.get_template(template_id) @@ -301,7 +301,7 @@ def get_template_detail( console.print(f"模板ID: {template_id}") # 创建模板管理器 - manager = TemplateManager() + manager = TemplateManagerCloud() # 获取模板详细信息 detail = manager.get_template_detail(template_id) @@ -357,7 +357,7 @@ def delete_template( console.print(f"模板ID: {template_id}") # 创建模板管理器 - manager = TemplateManager() + manager = TemplateManagerCloud() # 获取模板信息 template = manager.get_template(template_id) @@ -397,7 +397,7 @@ def show_stats(): console.print(f"📊 [bold blue]模板统计信息[/bold blue]") # 创建模板管理器 - manager = TemplateManager() + manager = TemplateManagerCloud() # 获取所有模板 templates = manager.get_templates() diff --git a/python_core/database/template.py b/python_core/database/template.py index 557a2ef..8715bdc 100644 --- a/python_core/database/template.py +++ b/python_core/database/template.py @@ -1,22 +1,502 @@ # 模板表 -from python_core.kv import kv + +import uuid +import json +from typing import Dict, List, Any, Optional +from datetime import datetime from .db import Db from .types import TemplateInfo +from python_core.utils.logger import setup_logger + +logger = setup_logger(__name__) + +class TemplateTable(Db): + """ + 模板表类 + 基于Db类实现的模板管理功能 + """ -class TemplateDb(Db): - def __init__(self): - self.key = "template" - pass - - def list(self)->list[TemplateInfo]: - ... - - def get(self, id: str)->TemplateInfo: - ... - - def create(self, info: TemplateInfo): - ... - - def remove(self, id: str): - ... \ No newline at end of file + # 使用固定的数据库键 + super().__init__("mixvideo") + self.table_name = "templates" + + # 初始化模板表 + self._init_template_table() + + def _init_template_table(self): + """初始化模板表""" + try: + # 检查模板表是否存在,不存在则创建 + if not self._table_exists(self.table_name): + schema = { + "name": "string", + "description": "string", + "thumbnail_path": "string", + "draft_content_path": "string", + "resources_path": "string", + "canvas_config": "json", + "duration": "integer", + "material_count": "integer", + "track_count": "integer", + "tags": "json", + "is_cloud": "boolean", + "user_id": "string", + "created_at": "datetime", + "updated_at": "datetime" + } + + self.create_table(self.table_name, schema) + logger.info("Template table initialized") + else: + logger.info("Template table already exists") + + except Exception as e: + logger.error(f"Failed to initialize template table: {e}") + raise e + + def create_template(self, template_info: TemplateInfo) -> str: + """ + 创建模板 + + Args: + template_info: 模板信息 + + Returns: + 模板ID + """ + try: + # 检查模板名称是否已存在(同一用户下) + existing_template = self.get_template_by_name(template_info.name, template_info.user_id) + if existing_template: + raise ValueError(f"Template name '{template_info.name}' already exists for this user") + + # 准备模板数据 + template_data = { + "name": template_info.name, + "description": template_info.description, + "thumbnail_path": template_info.thumbnail_path, + "draft_content_path": template_info.draft_content_path, + "resources_path": template_info.resources_path, + "canvas_config": template_info.canvas_config, + "duration": template_info.duration, + "material_count": template_info.material_count, + "track_count": template_info.track_count, + "tags": template_info.tags, + "is_cloud": template_info.is_cloud, + "user_id": template_info.user_id, + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat() + } + + # 插入模板记录 + template_id = self.insert(self.table_name, template_data) + + logger.info(f"Created template: {template_info.name} (ID: {template_id})") + return template_id + + except Exception as e: + logger.error(f"Failed to create template '{template_info.name}': {e}") + raise e + + def get_template_by_id(self, template_id: str) -> Optional[TemplateInfo]: + """ + 根据ID获取模板 + + Args: + template_id: 模板ID + + Returns: + 模板信息,如果不存在返回None + """ + try: + record = self.get(self.table_name, template_id) + if record: + return self._record_to_template_info(record) + return None + + except Exception as e: + logger.error(f"Failed to get template by ID '{template_id}': {e}") + return None + + def get_template_by_name(self, name: str, user_id: str = None) -> Optional[TemplateInfo]: + """ + 根据名称获取模板 + + Args: + name: 模板名称 + user_id: 用户ID(可选,用于过滤用户模板) + + Returns: + 模板信息,如果不存在返回None + """ + try: + # 获取所有同名模板 + templates = self.find_by_field(self.table_name, "name", name) + + for record in templates: + template_info = self._record_to_template_info(record) + # 如果指定了用户ID,则过滤用户模板 + if user_id is None or template_info.user_id == user_id: + return template_info + + return None + + except Exception as e: + logger.error(f"Failed to get template by name '{name}': {e}") + return None + + def update_template(self, template_id: str, updates: Dict[str, Any]) -> bool: + """ + 更新模板信息 + + Args: + template_id: 模板ID + updates: 要更新的字段 + + Returns: + 更新成功返回True + """ + try: + # 获取现有模板信息 + existing_template = self.get_template_by_id(template_id) + if not existing_template: + logger.warning(f"Template not found: {template_id}") + return False + + # 准备更新数据 + updated_data = existing_template.__dict__.copy() + + # 移除不应该直接更新的字段 + protected_fields = ["id", "created_at"] + for field in protected_fields: + if field in updates: + del updates[field] + + # 应用更新 + updated_data.update(updates) + updated_data["updated_at"] = datetime.now().isoformat() + + # 移除元数据字段,只保留数据字段 + data_fields = {k: v for k, v in updated_data.items() + if k not in ["id", "created_at", "updated_at"]} + + # 执行更新 + result = self.update(self.table_name, template_id, data_fields) + + if result: + logger.info(f"Updated template: {template_id}") + + return result + + except Exception as e: + logger.error(f"Failed to update template '{template_id}': {e}") + return False + + def delete_template(self, template_id: str) -> bool: + """ + 删除模板 + + Args: + template_id: 模板ID + + Returns: + 删除成功返回True + """ + try: + result = self.delete(self.table_name, template_id) + + if result: + logger.info(f"Deleted template: {template_id}") + + return result + + except Exception as e: + logger.error(f"Failed to delete template '{template_id}': {e}") + return False + + def get_templates_by_user(self, user_id: str, include_cloud: bool = True, limit: int = 100) -> List[TemplateInfo]: + """ + 获取用户的模板列表 + + Args: + user_id: 用户ID + include_cloud: 是否包含云端公共模板 + limit: 最大返回数量 + + Returns: + 模板列表 + """ + try: + all_records = self.find_all(self.table_name, limit * 2) # 获取更多记录以便过滤 + + templates = [] + for record in all_records: + if len(templates) >= limit: + break + + template_info = self._record_to_template_info(record) + + # 过滤条件:用户自己的模板 或 云端公共模板 + if (template_info.user_id == user_id or + (include_cloud and template_info.is_cloud)): + templates.append(template_info) + + return templates + + except Exception as e: + logger.error(f"Failed to get templates for user '{user_id}': {e}") + return [] + + def search_templates(self, query: str, user_id: str = None, include_cloud: bool = True, limit: int = 50) -> List[TemplateInfo]: + """ + 搜索模板 + + Args: + query: 搜索关键词(匹配名称、描述、标签) + user_id: 用户ID(可选,用于过滤用户模板) + include_cloud: 是否包含云端公共模板 + limit: 最大返回数量 + + Returns: + 匹配的模板列表 + """ + try: + all_templates = self.get_templates_by_user(user_id or "", include_cloud, limit * 2) + + matching_templates = [] + query_lower = query.lower() + + for template in all_templates: + if len(matching_templates) >= limit: + break + + # 检查名称、描述、标签是否匹配 + name_match = query_lower in template.name.lower() + desc_match = query_lower in template.description.lower() + tags_match = any(query_lower in tag.lower() for tag in template.tags) + + if name_match or desc_match or tags_match: + matching_templates.append(template) + + return matching_templates + + except Exception as e: + logger.error(f"Failed to search templates with query '{query}': {e}") + return [] + + def get_templates_by_tag(self, tag: str, user_id: str = None, include_cloud: bool = True, limit: int = 50) -> List[TemplateInfo]: + """ + 根据标签获取模板 + + Args: + tag: 标签名称 + user_id: 用户ID(可选) + include_cloud: 是否包含云端公共模板 + limit: 最大返回数量 + + Returns: + 匹配的模板列表 + """ + try: + all_templates = self.get_templates_by_user(user_id or "", include_cloud, limit * 2) + + matching_templates = [] + tag_lower = tag.lower() + + for template in all_templates: + if len(matching_templates) >= limit: + break + + # 检查标签是否匹配 + if any(tag_lower == t.lower() for t in template.tags): + matching_templates.append(template) + + return matching_templates + + except Exception as e: + logger.error(f"Failed to get templates by tag '{tag}': {e}") + return [] + + def get_cloud_templates(self, limit: int = 100) -> List[TemplateInfo]: + """ + 获取云端公共模板 + + Args: + limit: 最大返回数量 + + Returns: + 云端模板列表 + """ + try: + cloud_templates = self.find_by_field(self.table_name, "is_cloud", True) + + templates = [] + for record in cloud_templates[:limit]: + template_info = self._record_to_template_info(record) + templates.append(template_info) + + return templates + + except Exception as e: + logger.error(f"Failed to get cloud templates: {e}") + return [] + + def get_template_count(self, user_id: str = None, include_cloud: bool = True) -> int: + """ + 获取模板数量 + + Args: + user_id: 用户ID(可选) + include_cloud: 是否包含云端公共模板 + + Returns: + 模板数量 + """ + try: + if user_id is None: + return self.count(self.table_name) + else: + templates = self.get_templates_by_user(user_id, include_cloud) + return len(templates) + + except Exception as e: + logger.error(f"Failed to get template count: {e}") + return 0 + + def batch_import_templates(self, templates: List[TemplateInfo]) -> Dict[str, Any]: + """ + 批量导入模板 + + Args: + templates: 模板列表 + + Returns: + 导入结果统计 + """ + try: + success_count = 0 + failed_count = 0 + failed_templates = [] + + for template in templates: + try: + # 检查是否已存在(根据unique id属性) + existing = self.get_template_by_id(template.id) + if existing: + logger.warning(f"Template already exists, skipping: {template.name}") + continue + + # 创建模板 + self.create_template(template) + success_count += 1 + + except Exception as e: + logger.error(f"Failed to import template '{template.name}': {e}") + failed_count += 1 + failed_templates.append({ + "name": template.name, + "error": str(e) + }) + + result = { + "total": len(templates), + "success": success_count, + "failed": failed_count, + "failed_templates": failed_templates + } + + logger.info(f"Batch import completed: {success_count} success, {failed_count} failed") + return result + + except Exception as e: + logger.error(f"Failed to batch import templates: {e}") + return { + "total": len(templates), + "success": 0, + "failed": len(templates), + "error": str(e) + } + + def get_popular_tags(self, user_id: str = None, limit: int = 20) -> List[Dict[str, Any]]: + """ + 获取热门标签 + + Args: + user_id: 用户ID(可选) + limit: 最大返回数量 + + Returns: + 标签列表,包含标签名称和使用次数 + """ + try: + templates = self.get_templates_by_user(user_id or "", include_cloud=True) + + # 统计标签使用次数 + tag_counts = {} + for template in templates: + for tag in template.tags: + tag_counts[tag] = tag_counts.get(tag, 0) + 1 + + # 按使用次数排序 + sorted_tags = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True) + + # 返回前N个标签 + popular_tags = [] + for tag, count in sorted_tags[:limit]: + popular_tags.append({ + "tag": tag, + "count": count + }) + + return popular_tags + + except Exception as e: + logger.error(f"Failed to get popular tags: {e}") + return [] + + # 辅助方法 + def _record_to_template_info(self, record: Dict[str, Any]) -> TemplateInfo: + """将数据库记录转换为TemplateInfo对象""" + data = record["data"] + return TemplateInfo( + id=record["id"], + name=data["name"], + description=data["description"], + thumbnail_path=data["thumbnail_path"], + draft_content_path=data["draft_content_path"], + resources_path=data["resources_path"], + canvas_config=data["canvas_config"], + duration=data["duration"], + material_count=data["material_count"], + track_count=data["track_count"], + tags=data["tags"], + is_cloud=data["is_cloud"], + user_id=data["user_id"], + created_at=data.get("created_at", record.get("created_at", "")), + updated_at=data.get("updated_at", record.get("updated_at", "")) + ) + + def _template_info_to_dict(self, template_info: TemplateInfo) -> Dict[str, Any]: + """将TemplateInfo对象转换为字典""" + return { + "id": template_info.id, + "name": template_info.name, + "description": template_info.description, + "thumbnail_path": template_info.thumbnail_path, + "draft_content_path": template_info.draft_content_path, + "resources_path": template_info.resources_path, + "canvas_config": template_info.canvas_config, + "duration": template_info.duration, + "material_count": template_info.material_count, + "track_count": template_info.track_count, + "tags": template_info.tags, + "is_cloud": template_info.is_cloud, + "user_id": template_info.user_id, + "created_at": template_info.created_at, + "updated_at": template_info.updated_at + } + + +# 创建全局模板表实例 +template_table = TemplateTable() \ No newline at end of file diff --git a/python_core/database/user.py b/python_core/database/user.py index 1c13722..4bbbed5 100644 --- a/python_core/database/user.py +++ b/python_core/database/user.py @@ -17,7 +17,7 @@ class UserTable(Db): def __init__(self): # 使用固定的数据库键 - super().__init__("user_system") + super().__init__("mixvideo") self.table_name = "users" # 初始化用户表 diff --git a/python_core/services/template_manager_cloud.py b/python_core/services/template_manager_cloud.py new file mode 100644 index 0000000..91e1852 --- /dev/null +++ b/python_core/services/template_manager_cloud.py @@ -0,0 +1,552 @@ +""" +Template Management Service (Cloud Version) +基于TemplateTable实现的云端模板管理服务 +""" + +import os +import json +import shutil +import uuid +from pathlib import Path +from typing import Dict, List, Any, Optional +from dataclasses import asdict +from datetime import datetime + +from ..utils.logger import setup_logger +from ..config import settings +from python_core.database.types import TemplateInfo, MaterialInfo +from python_core.database.template import template_table + +logger = setup_logger(__name__) + +class TemplateManagerCloud: + """Cloud-based template management service using TemplateTable""" + + def __init__(self, user_id: str = "default"): + self.user_id = user_id + self.template_table = template_table + self.templates_dir = settings.temp_dir / "templates" + self.templates_dir.mkdir(parents=True, exist_ok=True) + + logger.info(f"TemplateManagerCloud initialized for user: {user_id}") + + def _determine_material_type(self, file_path: str) -> str: + """根据文件扩展名确定素材类型""" + file_extension = Path(file_path).suffix.lower().lstrip('.') + + # 视频格式 + video_extensions = { + 'mp4', 'avi', 'mov', 'mkv', 'wmv', 'flv', 'webm', 'm4v', + '3gp', 'ogv', 'ts', 'mts', 'm2ts', 'vob', 'asf', 'rm', 'rmvb' + } + + # 音频格式 + audio_extensions = { + 'mp3', 'wav', 'aac', 'flac', 'ogg', 'wma', 'm4a', 'opus', + 'aiff', 'au', 'ra', 'ac3', 'dts', 'ape' + } + + # 图片格式 + image_extensions = { + 'jpg', 'jpeg', 'png', 'gif', 'bmp', 'tiff', 'tga', 'webp', + 'svg', 'ico', 'psd', 'ai', 'eps', 'raw', 'cr2', 'nef' + } + + # 文本格式 + text_extensions = { + 'txt', 'srt', 'ass', 'ssa', 'vtt', 'sub', 'idx', 'sup' + } + + # 特效格式 + effect_extensions = { + 'fx', 'effect', 'json', 'xml', 'preset' + } + + if file_extension in video_extensions: + return 'video' + elif file_extension in audio_extensions: + return 'audio' + elif file_extension in image_extensions: + return 'image' + elif file_extension in text_extensions: + return 'text' + elif file_extension in effect_extensions: + return 'effect' + else: + return 'video' # 默认为视频类型 + + def batch_import_templates(self, source_folder: str, progress_callback=None) -> Dict[str, Any]: + """ + 批量导入模板到云端存储 + + Args: + source_folder: 包含模板子文件夹的源文件夹 + progress_callback: 进度回调函数 + + Returns: + 导入结果,包含成功/失败计数和详情 + """ + result = { + 'status': True, + 'msg': '', + 'imported_count': 0, + 'failed_count': 0, + 'imported_templates': [], + 'failed_templates': [] + } + + try: + source_path = Path(source_folder) + if not source_path.exists(): + result['status'] = False + result['msg'] = f"Source folder does not exist: {source_folder}" + return result + + # 获取所有子文件夹 + template_folders = [d for d in source_path.iterdir() if d.is_dir()] + total_templates = len(template_folders) + + if total_templates == 0: + result['msg'] = "No template folders found" + return result + + logger.info(f"Found {total_templates} template folders to import") + + for i, template_path in enumerate(template_folders): + try: + if progress_callback: + progress_callback(i, total_templates, f"Processing {template_path.name}") + + # 处理单个模板 + template_info = self._process_single_template(template_path) + if template_info: + # 保存到云端存储 + template_id = self.template_table.create_template(template_info) + + result['imported_count'] += 1 + result['imported_templates'].append({ + 'id': template_id, + 'name': template_info.name, + 'path': str(template_path) + }) + + logger.info(f"Successfully imported template: {template_info.name}") + + except Exception as e: + logger.error(f"Failed to import template {template_path.name}: {e}") + result['failed_count'] += 1 + result['failed_templates'].append({ + 'name': template_path.name, + 'path': str(template_path), + 'error': str(e) + }) + + if progress_callback: + progress_callback(total_templates, total_templates, "Import completed") + + result['msg'] = f"Import completed: {result['imported_count']} success, {result['failed_count']} failed" + logger.info(result['msg']) + + except Exception as e: + logger.error(f"Batch import failed: {e}") + result['status'] = False + result['msg'] = f"Batch import failed: {str(e)}" + + return result + + def _process_single_template(self, template_path: Path) -> Optional[TemplateInfo]: + """ + 处理单个模板文件夹 + + Args: + template_path: 模板文件夹路径 + + Returns: + TemplateInfo对象,如果处理失败返回None + """ + try: + template_name = template_path.name + + # 查找draft_content.json文件 + draft_file = template_path / "draft_content.json" + if not draft_file.exists(): + logger.warning(f"No draft_content.json found in {template_path}") + return None + + # 读取draft content + with open(draft_file, 'r', encoding='utf-8') as f: + draft_content = json.load(f) + + # 生成唯一ID + template_id = str(uuid.uuid4()) + + # 创建模板目录 + template_dir = self.templates_dir / template_id + template_dir.mkdir(parents=True, exist_ok=True) + + # 创建资源目录 + resources_dir = template_dir / "resources" + resources_dir.mkdir(exist_ok=True) + + # 处理素材并复制资源 + materials = self._process_materials(draft_content, template_path, resources_dir) + + # 更新draft content中的路径 + updated_draft = self._update_draft_paths(draft_content, materials) + + # 保存更新后的draft content + draft_target = template_dir / "draft_content.json" + with open(draft_target, 'w', encoding='utf-8') as f: + json.dump(updated_draft, f, ensure_ascii=False, indent=2) + + # 创建模板信息 + template_info = TemplateInfo( + id=template_id, + name=template_name, + description=f"Imported template: {template_name}", + thumbnail_path="", # TODO: 生成缩略图 + draft_content_path=str(draft_target), + resources_path=str(resources_dir), + created_at=datetime.now().isoformat(), + updated_at=datetime.now().isoformat(), + canvas_config=draft_content.get('canvas_config', {}), + duration=draft_content.get('duration', 0), + material_count=len(materials), + track_count=len(draft_content.get('tracks', [])), + tags=[], + is_cloud=False, # 用户导入的模板默认为本地 + user_id=self.user_id + ) + + logger.info(f"Successfully processed template: {template_name} -> {template_id}") + return template_info + + except Exception as e: + logger.error(f"Failed to process template {template_path}: {e}") + return None + + def _process_materials(self, draft_content: Dict, template_path: Path, resources_dir: Path) -> List[MaterialInfo]: + """处理模板中的素材文件""" + materials = [] + + try: + # 从draft content中提取素材信息 + tracks = draft_content.get('tracks', []) + + for track in tracks: + segments = track.get('segments', []) + for segment in segments: + material_path = segment.get('material_path', '') + if material_path: + # 构建完整路径 + source_file = template_path / material_path + if source_file.exists(): + # 生成新的文件名 + material_id = str(uuid.uuid4()) + file_extension = source_file.suffix + new_filename = f"{material_id}{file_extension}" + target_file = resources_dir / new_filename + + # 复制文件 + shutil.copy2(source_file, target_file) + + # 创建素材信息 + material_info = MaterialInfo( + id=material_id, + material_id=material_id, + name=source_file.name, + type=self._determine_material_type(str(source_file)), + original_path=str(source_file), + relative_path=new_filename, + duration=segment.get('duration'), + size=source_file.stat().st_size if source_file.exists() else None, + is_cloud=False, + user_id=self.user_id + ) + + materials.append(material_info) + + # 更新segment中的路径 + segment['material_path'] = new_filename + segment['material_id'] = material_id + + except Exception as e: + logger.error(f"Failed to process materials: {e}") + + return materials + + def _update_draft_paths(self, draft_content: Dict, materials: List[MaterialInfo]) -> Dict: + """更新draft content中的路径为相对路径""" + # 创建路径映射 + path_mapping = {} + for material in materials: + path_mapping[material.original_path] = material.relative_path + + # 这里可以根据需要更新draft_content中的路径 + # 当前实现已在_process_materials中更新了路径 + + return draft_content + + def get_templates(self, include_cloud: bool = True) -> List[TemplateInfo]: + """ + 获取模板列表 + + Args: + include_cloud: 是否包含云端公共模板 + + Returns: + 模板列表 + """ + try: + templates = self.template_table.get_templates_by_user( + self.user_id, + include_cloud=include_cloud + ) + logger.info(f"Retrieved {len(templates)} templates for user {self.user_id}") + return templates + + except Exception as e: + logger.error(f"Failed to get templates: {e}") + return [] + + def get_template(self, template_id: str) -> Optional[TemplateInfo]: + """ + 获取特定模板 + + Args: + template_id: 模板ID + + Returns: + 模板信息,如果不存在返回None + """ + try: + template = self.template_table.get_template_by_id(template_id) + if template: + logger.info(f"Retrieved template: {template.name}") + else: + logger.warning(f"Template not found: {template_id}") + return template + + except Exception as e: + logger.error(f"Failed to get template {template_id}: {e}") + return None + + def delete_template(self, template_id: str) -> bool: + """ + 删除模板 + + Args: + template_id: 模板ID + + Returns: + 删除成功返回True + """ + try: + # 获取模板信息以进行权限检查 + template = self.template_table.get_template_by_id(template_id) + if not template: + logger.warning(f"Template not found for deletion: {template_id}") + return False + + # 检查权限(只能删除自己的模板) + if template.user_id != self.user_id: + logger.warning(f"Permission denied: user {self.user_id} cannot delete template {template_id}") + return False + + # 删除本地文件 + template_dir = self.templates_dir / template_id + if template_dir.exists(): + shutil.rmtree(template_dir) + logger.info(f"Deleted local template directory: {template_dir}") + + # 从云端存储删除 + result = self.template_table.delete_template(template_id) + if result: + logger.info(f"Successfully deleted template: {template_id}") + + return result + + except Exception as e: + logger.error(f"Failed to delete template {template_id}: {e}") + return False + + def search_templates(self, query: str, filters: Optional[Dict[str, Any]] = None) -> List[TemplateInfo]: + """ + 搜索模板 + + Args: + query: 搜索关键词 + filters: 搜索过滤器(暂未实现) + + Returns: + 匹配的模板列表 + """ + try: + templates = self.template_table.search_templates( + query, + user_id=self.user_id, + include_cloud=True + ) + logger.info(f"Search '{query}' found {len(templates)} templates") + return templates + + except Exception as e: + logger.error(f"Failed to search templates with query '{query}': {e}") + return [] + + def get_templates_by_tag(self, tag: str) -> List[TemplateInfo]: + """ + 根据标签获取模板 + + Args: + tag: 标签名称 + + Returns: + 匹配的模板列表 + """ + try: + templates = self.template_table.get_templates_by_tag( + tag, + user_id=self.user_id, + include_cloud=True + ) + logger.info(f"Tag '{tag}' found {len(templates)} templates") + return templates + + except Exception as e: + logger.error(f"Failed to get templates by tag '{tag}': {e}") + return [] + + def update_template(self, template_id: str, updates: Dict[str, Any]) -> bool: + """ + 更新模板信息 + + Args: + template_id: 模板ID + updates: 要更新的字段 + + Returns: + 更新成功返回True + """ + try: + # 检查权限 + template = self.template_table.get_template_by_id(template_id) + if not template or template.user_id != self.user_id: + logger.warning(f"Permission denied or template not found: {template_id}") + return False + + result = self.template_table.update_template(template_id, updates) + if result: + logger.info(f"Successfully updated template: {template_id}") + + return result + + except Exception as e: + logger.error(f"Failed to update template {template_id}: {e}") + return False + + def get_template_count(self, include_cloud: bool = True) -> int: + """ + 获取模板数量 + + Args: + include_cloud: 是否包含云端公共模板 + + Returns: + 模板数量 + """ + try: + count = self.template_table.get_template_count( + user_id=self.user_id, + include_cloud=include_cloud + ) + logger.info(f"Template count for user {self.user_id}: {count}") + return count + + except Exception as e: + logger.error(f"Failed to get template count: {e}") + return 0 + + def get_popular_tags(self, limit: int = 20) -> List[Dict[str, Any]]: + """ + 获取热门标签 + + Args: + limit: 最大返回数量 + + Returns: + 标签列表,包含标签名称和使用次数 + """ + try: + tags = self.template_table.get_popular_tags( + user_id=self.user_id, + limit=limit + ) + logger.info(f"Retrieved {len(tags)} popular tags") + return tags + + except Exception as e: + logger.error(f"Failed to get popular tags: {e}") + return [] + + def create_template_from_draft(self, name: str, description: str, draft_content: Dict[str, Any], + tags: List[str] = None, is_cloud: bool = False) -> Optional[str]: + """ + 从draft内容创建模板 + + Args: + name: 模板名称 + description: 模板描述 + draft_content: draft内容 + tags: 标签列表 + is_cloud: 是否为云端公共模板 + + Returns: + 模板ID,如果创建失败返回None + """ + try: + template_id = str(uuid.uuid4()) + + # 创建模板目录 + template_dir = self.templates_dir / template_id + template_dir.mkdir(parents=True, exist_ok=True) + + # 保存draft content + draft_target = template_dir / "draft_content.json" + with open(draft_target, 'w', encoding='utf-8') as f: + json.dump(draft_content, f, ensure_ascii=False, indent=2) + + # 创建模板信息 + template_info = TemplateInfo( + id=template_id, + name=name, + description=description, + thumbnail_path="", + draft_content_path=str(draft_target), + resources_path=str(template_dir / "resources"), + created_at=datetime.now().isoformat(), + updated_at=datetime.now().isoformat(), + canvas_config=draft_content.get('canvas_config', {}), + duration=draft_content.get('duration', 0), + material_count=len(draft_content.get('materials', [])), + track_count=len(draft_content.get('tracks', [])), + tags=tags or [], + is_cloud=is_cloud, + user_id=self.user_id + ) + + # 保存到云端存储 + created_id = self.template_table.create_template(template_info) + logger.info(f"Successfully created template from draft: {name} -> {created_id}") + + return created_id + + except Exception as e: + logger.error(f"Failed to create template from draft: {e}") + return None + + +# 创建全局实例工厂函数 +def create_template_manager_cloud(user_id: str = "default") -> TemplateManagerCloud: + """创建TemplateManagerCloud实例""" + return TemplateManagerCloud(user_id=user_id) diff --git a/text2video.json b/text2video.json new file mode 100644 index 0000000..be96acb --- /dev/null +++ b/text2video.json @@ -0,0 +1,930 @@ +{ + "openapi": "3.1.0", + "info": { + "title": "Text2Video - 图片生成 API", + "description": "基于 Midjourney 的图片生成服务", + "version": "1.0.0" + }, + "paths": { + "/api/mj/sync/image": { + "post": { + "tags": [ + "midjourney生图" + ], + "summary": "同步生成图片接口", + "description": "同步生成图片接口 - 提交任务并轮询结果\n\nArgs:\n prompt (str): 图片生成提示词\n max_wait_time (int): 最大等待时间,默认120秒\n poll_interval (int): 轮询间隔,默认2秒\n\nReturns:\n dict: 包含status, msg, data的响应字典", + "operationId": "generate_image_sync_api_mj_sync_image_post", + "parameters": [ + { + "name": "prompt", + "in": "query", + "required": true, + "schema": { + "type": "string", + "title": "Prompt" + } + }, + { + "name": "max_wait_time", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "default": 120, + "title": "Max Wait Time" + } + }, + { + "name": "poll_interval", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "default": 2, + "title": "Poll Interval" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/mj/generate-image": { + "post": { + "tags": [ + "midjourney生图" + ], + "summary": "生成图片", + "description": "生成图片接口", + "operationId": "generate_image_api_api_mj_generate_image_post", + "requestBody": { + "content": { + "application/x-www-form-urlencoded": { + "schema": { + "$ref": "#/components/schemas/Body_generate_image_api_api_mj_generate_image_post" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/mj/sync/img/describe": { + "post": { + "tags": [ + "midjourney生图" + ], + "summary": "获取图像描述", + "description": "获取图像描述接口", + "operationId": "describe_image_api_api_mj_sync_img_describe_post", + "requestBody": { + "content": { + "application/x-www-form-urlencoded": { + "schema": { + "$ref": "#/components/schemas/Body_describe_image_api_api_mj_sync_img_describe_post" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/mj/health": { + "get": { + "tags": [ + "midjourney生图" + ], + "summary": "健康检查", + "description": "健康检查接口", + "operationId": "health_check_api_mj_health_get", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + } + } + } + }, + "/api/file/upload": { + "post": { + "tags": [ + "文件操作" + ], + "summary": "上传文件到COS", + "description": "上传文件到腾讯云COS\n\n- **file**: 要上传的文件\n\n返回上传后的文件URL", + "operationId": "upload_file_api_file_upload_post", + "requestBody": { + "content": { + "multipart/form-data": { + "schema": { + "$ref": "#/components/schemas/Body_upload_file_api_file_upload_post" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/FileUploadResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/jm/generate-video": { + "post": { + "tags": [ + "极梦视频生成api" + ], + "summary": "生成视频", + "description": "生成视频接口", + "operationId": "generate_video_api_api_jm_generate_video_post", + "requestBody": { + "content": { + "application/x-www-form-urlencoded": { + "schema": { + "$ref": "#/components/schemas/Body_generate_video_api_api_jm_generate_video_post" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/jm/health": { + "get": { + "tags": [ + "极梦视频生成api" + ], + "summary": "健康检查", + "description": "健康检查接口", + "operationId": "health_check_api_jm_health_get", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + } + } + } + }, + "/api/prompt/default": { + "get": { + "tags": [ + "处理提示词" + ], + "summary": "获取示例提示词", + "operationId": "get_sample_prompt_api_prompt_default_get", + "parameters": [ + { + "name": "task_type", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Task Type" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/task/test": { + "post": { + "tags": [ + "任务管理api" + ], + "summary": "测试 Modal 连接", + "description": "测试 Modal 连接是否正常", + "operationId": "test_modal_connection_api_task_test_post", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + } + } + } + }, + "/api/task/create/task": { + "post": { + "tags": [ + "任务管理api" + ], + "summary": "异步提交任务(不阻塞)", + "description": "异步提交任务到 Modal 进行处理(不阻塞,立即返回任务ID)\n\n- **task_type**: 任务类型 (tea/chop/lady/vlog)\n- **prompt**: 生成提示词\n- **img_url**: 可选的参考图片URL,如果提供会先进行图片描述\n- **ar**: 生成图片的长宽比 默认为9:16\n\n立即返回 Modal 任务ID,任务在后台异步执行", + "operationId": "submit_task_api_task_create_task_post", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TaskRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/task/{task_id}": { + "get": { + "tags": [ + "任务管理api" + ], + "summary": "同步查询任务结果(阻塞等待)", + "description": "同步查询任务的执行状态和结果(阻塞等待模式)\n\n- **task_id**: 任务ID\n\n**阻塞等待任务完成**,直到任务执行完成才返回最终结果\n适用于需要等待任务完成的场景", + "operationId": "query_task_result_sync_api_task__task_id__get", + "parameters": [ + { + "name": "task_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "title": "Task Id" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/task/status/{task_id}": { + "get": { + "tags": [ + "任务管理api" + ], + "summary": "异步查询任务状态(立即返回)", + "description": "异步查询任务的状态信息(非阻塞,立即返回当前状态)\n\n- **task_id**: 任务ID\n\n**立即返回当前状态**,不等待任务完成\n适用于需要快速检查任务状态的场景", + "operationId": "get_task_status_async_api_task_status__task_id__get", + "parameters": [ + { + "name": "task_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "title": "Task Id" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/mj/video/async/submit": { + "post": { + "tags": [ + "midjourney视频生成api" + ], + "summary": "异步提交生成视频任务", + "operationId": "submit_video_task_api_mj_video_async_submit_post", + "parameters": [ + { + "name": "prompt", + "in": "query", + "required": true, + "schema": { + "type": "string", + "title": "Prompt" + } + }, + { + "name": "img_url", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Img Url" + } + } + ], + "requestBody": { + "content": { + "multipart/form-data": { + "schema": { + "$ref": "#/components/schemas/Body_submit_video_task_api_mj_video_async_submit_post" + } + } + } + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/mj/video/async/task/status": { + "post": { + "tags": [ + "midjourney视频生成api" + ], + "summary": "异步查询生成任务进度", + "operationId": "async_query_task_status_api_mj_video_async_task_status_post", + "parameters": [ + { + "name": "task_id", + "in": "query", + "required": true, + "schema": { + "type": "string", + "title": "Task Id" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/api/mj/video/sync/gen": { + "post": { + "tags": [ + "midjourney视频生成api" + ], + "summary": "同步生成视频", + "operationId": "sync_submit_task_api_mj_video_sync_gen_post", + "parameters": [ + { + "name": "prompt", + "in": "query", + "required": true, + "schema": { + "type": "string", + "title": "Prompt" + } + }, + { + "name": "img_url", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Img Url" + } + } + ], + "requestBody": { + "content": { + "multipart/form-data": { + "schema": { + "$ref": "#/components/schemas/Body_sync_submit_task_api_mj_video_sync_gen_post" + } + } + } + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/": { + "get": { + "summary": "根路径", + "description": "根路径,返回API信息", + "operationId": "root__get", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": {} + } + } + } + } + } + } + }, + "components": { + "schemas": { + "Body_describe_image_api_api_mj_sync_img_describe_post": { + "properties": { + "image_url": { + "type": "string", + "title": "Image Url", + "description": "图片URL地址" + }, + "max_wait_time": { + "type": "integer", + "title": "Max Wait Time", + "description": "最大等待时间(秒)", + "default": 120 + }, + "poll_interval": { + "type": "integer", + "title": "Poll Interval", + "description": "轮询间隔(秒)", + "default": 2 + } + }, + "type": "object", + "required": [ + "image_url" + ], + "title": "Body_describe_image_api_api_mj_sync_img_describe_post" + }, + "Body_generate_image_api_api_mj_generate_image_post": { + "properties": { + "prompt": { + "type": "string", + "title": "Prompt", + "description": "图片生成提示词" + }, + "max_wait_time": { + "type": "integer", + "title": "Max Wait Time", + "description": "最大等待时间(秒)", + "default": 120 + }, + "poll_interval": { + "type": "integer", + "title": "Poll Interval", + "description": "轮询间隔(秒)", + "default": 2 + } + }, + "type": "object", + "required": [ + "prompt" + ], + "title": "Body_generate_image_api_api_mj_generate_image_post" + }, + "Body_generate_video_api_api_jm_generate_video_post": { + "properties": { + "prompt": { + "type": "string", + "title": "Prompt", + "description": "视频生成提示词" + }, + "img_url": { + "type": "string", + "title": "Img Url", + "description": "图片URL地址" + }, + "duration": { + "type": "string", + "title": "Duration", + "description": "视频时长(秒)", + "default": "5" + }, + "max_wait_time": { + "type": "integer", + "title": "Max Wait Time", + "description": "最大等待时间(秒)", + "default": 300 + }, + "poll_interval": { + "type": "integer", + "title": "Poll Interval", + "description": "轮询间隔(秒)", + "default": 5 + } + }, + "type": "object", + "required": [ + "prompt", + "img_url" + ], + "title": "Body_generate_video_api_api_jm_generate_video_post" + }, + "Body_submit_video_task_api_mj_video_async_submit_post": { + "properties": { + "img_file": { + "anyOf": [ + { + "type": "string", + "format": "binary" + }, + { + "type": "null" + } + ], + "title": "Img File" + } + }, + "type": "object", + "title": "Body_submit_video_task_api_mj_video_async_submit_post" + }, + "Body_sync_submit_task_api_mj_video_sync_gen_post": { + "properties": { + "img_file": { + "anyOf": [ + { + "type": "string", + "format": "binary" + }, + { + "type": "null" + } + ], + "title": "Img File" + } + }, + "type": "object", + "title": "Body_sync_submit_task_api_mj_video_sync_gen_post" + }, + "Body_upload_file_api_file_upload_post": { + "properties": { + "file": { + "type": "string", + "format": "binary", + "title": "File" + } + }, + "type": "object", + "required": [ + "file" + ], + "title": "Body_upload_file_api_file_upload_post" + }, + "FileUploadResponse": { + "properties": { + "status": { + "type": "boolean", + "title": "Status", + "description": "上传状态" + }, + "msg": { + "type": "string", + "title": "Msg", + "description": "响应消息" + }, + "data": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data", + "description": "文件URL" + }, + "filename": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Filename", + "description": "原始文件名" + } + }, + "type": "object", + "required": [ + "status", + "msg" + ], + "title": "FileUploadResponse", + "description": "文件上传响应模型" + }, + "HTTPValidationError": { + "properties": { + "detail": { + "items": { + "$ref": "#/components/schemas/ValidationError" + }, + "type": "array", + "title": "Detail" + } + }, + "type": "object", + "title": "HTTPValidationError" + }, + "TaskRequest": { + "properties": { + "task_type": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Task Type", + "description": "任务类型如: tea, chop, lady, vlog", + "default": "" + }, + "prompt": { + "type": "string", + "title": "Prompt", + "description": "生图的提示词" + }, + "img_url": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Img Url", + "description": "参考图", + "default": "" + }, + "ar": { + "type": "string", + "title": "Ar", + "description": "生成图片,视频的分辨率", + "default": "9:16" + } + }, + "type": "object", + "required": [ + "prompt" + ], + "title": "TaskRequest" + }, + "ValidationError": { + "properties": { + "loc": { + "items": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "integer" + } + ] + }, + "type": "array", + "title": "Location" + }, + "msg": { + "type": "string", + "title": "Message" + }, + "type": { + "type": "string", + "title": "Error Type" + } + }, + "type": "object", + "required": [ + "loc", + "msg", + "type" + ], + "title": "ValidationError" + } + } + } +} \ No newline at end of file