comi-logic/mxivideo
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 集成 AI 视频生成功能到 MixVideo V2

Browse Source 
🎬 主要功能：
- ✅ 完整的 AI 视频生成模块 (Python)
- ✅ 图片转视频 API 集成 (字节跳动 Seedance)
- ✅ 云存储支持 (腾讯云 COS)
- ✅ 单张图片和批量处理模式
- ✅ 现代化 React 界面组件
- ✅ Tauri 桥接通信

🛠️ 技术实现：
- Python 模块：VideoGenerator, CloudStorage, APIClient
- Rust 命令：generate_ai_video, batch_generate_ai_videos
- React 组件：AIVideoGenerator, AIVideoPage
- 状态管理：useAIVideoStore (Zustand)
- 路由集成：/ai-video 页面

�� 新增文件：
- python_core/ai_video/ - AI 视频生成核心模块
- src/components/AIVideoGenerator.tsx - 主要 UI 组件
- src/pages/AIVideoPage.tsx - AI 视频生成页面
- src/stores/useAIVideoStore.ts - 状态管理

🎯 功能特性：
- 支持 Lite (720p) 和 Pro (1080p) 模型
- 可配置视频时长 (5秒/10秒)
- 实时进度跟踪和任务管理
- 批量处理多张图片
- 云存储自动上传下载
- 错误处理和重试机制

🔗 界面集成：
- 侧边栏导航添加 'AI 视频' 入口
- 首页快速操作卡片
- 完整的用户引导和帮助文档

这是从原始 Tkinter GUI 到现代 Web 应用的完整迁移！
This commit is contained in:
root
2025-07-10 10:43:40 +08:00
parent 289fb4f7e2
commit 96e166725b
14 changed files with 2359 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
python_core/ai_video/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
"""
AI Video Generation Module
AI 视频生成模块
This module provides functionality for generating videos from images using AI models.
"""
from .video_generator import VideoGenerator
from .cloud_storage import CloudStorage
from .api_client import APIClient
__all__ = ['VideoGenerator', 'CloudStorage', 'APIClient']

258
python_core/ai_video/api_client.py Normal file
View File

@@ -0,0 +1,258 @@
#!/usr/bin/env python3
"""
API Client Module
API 客户端模块
Handles communication with AI video generation APIs.
"""
import os
import time
from typing import Dict, Any, Optional, Callable
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import settings
from utils import setup_logger
logger = setup_logger(__name__)
class APIClient:
"""Client for AI video generation API."""
def __init__(self, api_key: str = None, base_url: str = None):
"""
Initialize API client.
Args:
api_key: API key for authentication
base_url: Base URL for API endpoints
"""
self.api_key = api_key or os.getenv('AI_VIDEO_API_KEY', '21575c22-14aa-40ca-8aa8-f00ca27a3a17')
self.base_url = base_url or os.getenv('AI_VIDEO_BASE_URL', 'https://ark.cn-beijing.volces.com/api/v3')
# Model configurations
self.models = {
'lite': {
'name': 'doubao-seedance-1-0-lite-i2v-250428',
'resolution': '720p'
},
'pro': {
'name': 'doubao-seedance-1-0-pro-250528',
'resolution': '1080p'
}
}
def submit_task(self, prompt: str, img_url: str, duration: str = '5', model_type: str = 'lite') -> Dict[str, Any]:
"""
Submit video generation task.
Args:
prompt: Text prompt for video generation
img_url: URL of the input image
duration: Video duration ('5' or '10')
model_type: Model type ('lite' or 'pro')
Returns:
Dictionary with task submission result
"""
result = {'status': False, 'data': None, 'msg': ''}
if duration not in ('5', '10'):
result['msg'] = 'Duration must be either 5 or 10'
return result
if model_type not in self.models:
result['msg'] = f'Model type must be one of: {list(self.models.keys())}'
return result
try:
import requests
model_config = self.models[model_type]
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.api_key}',
}
json_data = {
'model': model_config['name'],
'content': [
{
'type': 'text',
'text': f'{prompt} --resolution {model_config["resolution"]} --dur {duration} --camerafixed false',
},
{
'type': 'image_url',
'image_url': {
'url': img_url,
},
},
],
}
response = requests.post(
f'{self.base_url}/contents/generations/tasks',
headers=headers,
json=json_data,
timeout=30
)
# Check HTTP status code
if response.status_code != 200:
error_msg = f"API request failed, status code: {response.status_code}, response: {response.text}"
logger.error(error_msg)
result['msg'] = error_msg
return result
resp_json = response.json()
# Check if response contains id field
if 'id' not in resp_json:
error_msg = f"API response missing id field, response: {resp_json}"
logger.error(error_msg)
result['msg'] = error_msg
return result
job_id = resp_json['id']
result['status'] = True
result['data'] = job_id
result['msg'] = 'Task submitted successfully'
logger.info(f"Task submitted successfully, job ID: {job_id}")
except Exception as e:
logger.error(f"Failed to submit task: {str(e)}")
result['msg'] = str(e)
return result
def query_task_status(self, job_id: str) -> Dict[str, Any]:
"""
Query task status.
Args:
job_id: Task ID to query
Returns:
Dictionary with task status
"""
result = {'status': False, 'data': None, 'msg': ''}
try:
import requests
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.api_key}',
}
response = requests.get(
f'{self.base_url}/contents/generations/tasks/{job_id}',
headers=headers,
timeout=30
)
if response.status_code != 200:
result['msg'] = f"API request failed, status code: {response.status_code}"
return result
resp_json = response.json()
# Parse response
task_status = resp_json.get('status', 'unknown')
result['msg'] = task_status
if task_status == 'succeeded':
result['status'] = True
result['data'] = resp_json.get('content', {}).get('video_url')
elif task_status in ['failed', 'cancelled']:
result['status'] = False
result['data'] = None
else:
# Still running, pending, or queued
result['status'] = False
result['data'] = None
except Exception as e:
logger.error(f"Failed to query task status: {str(e)}")
result['msg'] = str(e)
return result
def wait_for_completion(self, job_id: str, timeout: int = 180, interval: int = 2,
progress_callback: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
"""
Wait for task completion with progress updates.
Args:
job_id: Task ID to wait for
timeout: Maximum wait time in seconds
interval: Check interval in seconds
progress_callback: Optional callback for progress updates
Returns:
Dictionary with final result
"""
result = {'status': False, 'data': None, 'msg': ''}
end_time = time.time() + timeout
wait_count = 0
if progress_callback:
progress_callback(f"开始查询任务状态，任务ID: {job_id}")
while time.time() < end_time:
status_result = self.query_task_status(job_id)
if status_result['status']:
# Task completed successfully
result['status'] = True
result['data'] = status_result['data']
result['msg'] = 'succeeded'
if progress_callback:
progress_callback("✓ 视频生成完成！")
break
elif status_result['msg'] == 'running':
wait_count += 1
elapsed = wait_count * interval
remaining = max(0, timeout - elapsed)
progress_msg = f"⏳ 任务运行中，已等待{elapsed}秒，预计剩余{remaining}秒..."
logger.info(progress_msg)
if progress_callback:
progress_callback(progress_msg)
time.sleep(interval)
elif status_result['msg'] == 'failed':
result['msg'] = '任务执行失败'
if progress_callback:
progress_callback("✗ 任务执行失败")
break
elif status_result['msg'] in ['pending', 'queued']:
wait_count += 1
elapsed = wait_count * interval
remaining = max(0, timeout - elapsed)
progress_msg = f"⏳ 任务排队中，已等待{elapsed}秒，预计剩余{remaining}秒..."
logger.info(progress_msg)
if progress_callback:
progress_callback(progress_msg)
time.sleep(interval)
else:
# Unknown status, continue waiting
wait_count += 1
logger.info(f"未知状态: {status_result['msg']}，继续等待...")
if progress_callback:
progress_callback(f"❔ 状态: {status_result['msg']}，继续等待...")
time.sleep(interval)
if not result['status'] and result['msg'] == '':
result['msg'] = '任务超时'
if progress_callback:
progress_callback(f"⏰ 任务查询超时({timeout}秒)")
return result

244
python_core/ai_video/cloud_storage.py Normal file
View File

@@ -0,0 +1,244 @@
#!/usr/bin/env python3
"""
Cloud Storage Module
云存储模块
Handles file upload and download operations with cloud storage services.
"""
import os
import time
import mimetypes
from typing import Dict, Any, Optional
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import settings
from utils import setup_logger
logger = setup_logger(__name__)
class CloudStorage:
"""Cloud storage service for handling file uploads and downloads."""
def __init__(self,
bucket_name: str = None,
secret_id: str = None,
secret_key: str = None,
region: str = None):
"""
Initialize cloud storage client.
Args:
bucket_name: COS bucket name
secret_id: COS secret ID
secret_key: COS secret key
region: COS region
"""
# Use provided credentials or fallback to environment variables
self.bucket_name = bucket_name or os.getenv('COS_BUCKET_NAME', 'sucai-1324682537')
self.secret_id = secret_id or os.getenv('COS_SECRET_ID', 'AKIDsrihIyjZOBsjimt8TsN8yvv1AMh5dB44')
self.secret_key = secret_key or os.getenv('COS_SECRET_KEY', 'CPZcxdk6W39Jd4cGY95wvupoyMd0YFqW')
self.region = region or os.getenv('COS_REGION', 'ap-shanghai')
self.client = None
self._initialize_client()
def _initialize_client(self):
"""Initialize COS client."""
try:
from qcloud_cos import CosConfig, CosS3Client
config = CosConfig(
Region=self.region,
SecretId=self.secret_id,
SecretKey=self.secret_key
)
self.client = CosS3Client(config)
logger.info("COS client initialized successfully")
except ImportError:
logger.warning("qcloud_cos not installed. Cloud storage features will be disabled.")
self.client = None
except Exception as e:
logger.error(f"Failed to initialize COS client: {str(e)}")
self.client = None
def upload_file(self, file_path: str, remove_src_file: bool = False) -> Dict[str, Any]:
"""
Upload file to cloud storage.
Args:
file_path: Local file path to upload
remove_src_file: Whether to remove source file after upload
Returns:
Dictionary with upload result
"""
result = {'status': False, 'data': '', 'msg': ''}
if not self.client:
result['msg'] = 'Cloud storage client not available'
return result
if not os.path.exists(file_path):
result['msg'] = f'File not found: {file_path}'
return result
try:
# Determine file type and generate object key
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type:
mime_type = 'application/octet-stream'
category = mime_type.split('/')[0]
file_name = os.path.basename(file_path)
suffix = file_name.split('.')[-1] if '.' in file_name else 'bin'
real_name = f'{int(time.time() * 1000)}.{suffix}'
object_key = f'mixvideo/{category}/{real_name}'
# Upload file
response = self.client.upload_file(
Bucket=self.bucket_name,
Key=object_key,
LocalFilePath=file_path,
EnableMD5=False,
progress_callback=None
)
# Generate public URL
url = f'https://{self.bucket_name}.cos.{self.region}.myqcloud.com/{object_key}'
result['status'] = True
result['data'] = url
result['msg'] = 'Upload successful'
logger.info(f"File uploaded successfully: {url}")
except Exception as e:
logger.error(f"Failed to upload file: {str(e)}")
result['msg'] = str(e)
finally:
if remove_src_file and result['status'] and os.path.exists(file_path):
try:
os.remove(file_path)
logger.info(f"Source file removed: {file_path}")
except Exception as e:
logger.warning(f"Failed to remove source file: {str(e)}")
return result
def download_file(self, url: str, save_path: str, filename: str = None) -> str:
"""
Download file from URL to local path.
Args:
url: File URL to download
save_path: Directory to save the file
filename: Optional custom filename
Returns:
Local file path if successful, None if failed
"""
try:
import requests
if not filename:
filename = f'{int(time.time() * 1000)}.mp4'
full_path = os.path.join(save_path, filename)
# Ensure save directory exists
os.makedirs(save_path, exist_ok=True)
# Download file
response = requests.get(url, stream=True)
response.raise_for_status()
with open(full_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=1024 * 1024 * 5):
if chunk:
f.write(chunk)
logger.info(f"File downloaded successfully: {full_path}")
return full_path
except Exception as e:
logger.error(f"Failed to download file: {str(e)}")
return None
def list_files(self, prefix: str = 'mixvideo/') -> Dict[str, Any]:
"""
List files in cloud storage.
Args:
prefix: Object key prefix to filter
Returns:
Dictionary with file list
"""
result = {'status': False, 'files': [], 'msg': ''}
if not self.client:
result['msg'] = 'Cloud storage client not available'
return result
try:
response = self.client.list_objects(
Bucket=self.bucket_name,
Prefix=prefix
)
files = []
if 'Contents' in response:
for obj in response['Contents']:
files.append({
'key': obj['Key'],
'size': obj['Size'],
'last_modified': obj['LastModified'],
'url': f'https://{self.bucket_name}.cos.{self.region}.myqcloud.com/{obj["Key"]}'
})
result['status'] = True
result['files'] = files
result['msg'] = f'Found {len(files)} files'
except Exception as e:
logger.error(f"Failed to list files: {str(e)}")
result['msg'] = str(e)
return result
def delete_file(self, object_key: str) -> Dict[str, Any]:
"""
Delete file from cloud storage.
Args:
object_key: Object key to delete
Returns:
Dictionary with deletion result
"""
result = {'status': False, 'msg': ''}
if not self.client:
result['msg'] = 'Cloud storage client not available'
return result
try:
self.client.delete_object(
Bucket=self.bucket_name,
Key=object_key
)
result['status'] = True
result['msg'] = 'File deleted successfully'
logger.info(f"File deleted: {object_key}")
except Exception as e:
logger.error(f"Failed to delete file: {str(e)}")
result['msg'] = str(e)
return result

363
python_core/ai_video/video_generator.py Normal file
View File

@@ -0,0 +1,363 @@
#!/usr/bin/env python3
"""
Video Generator Module
视频生成器模块
Main module for AI-powered video generation from images.
"""
import os
import glob
from typing import Dict, Any, List, Optional, Callable
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import settings
from utils import setup_logger
from .cloud_storage import CloudStorage
from .api_client import APIClient
logger = setup_logger(__name__)
class VideoGenerator:
"""AI video generator for converting images to videos."""
def __init__(self,
api_key: str = None,
cos_config: Dict[str, str] = None):
"""
Initialize video generator.
Args:
api_key: API key for video generation service
cos_config: Cloud storage configuration
"""
self.api_client = APIClient(api_key=api_key)
if cos_config:
self.cloud_storage = CloudStorage(**cos_config)
else:
self.cloud_storage = CloudStorage()
def generate_video_from_image(self,
image_path: str,
prompt: str,
duration: str = '5',
model_type: str = 'lite',
save_path: str = None,
timeout: int = 180,
interval: int = 2,
progress_callback: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
"""
Generate video from a single image.
Args:
image_path: Path to input image
prompt: Text prompt for video generation
duration: Video duration ('5' or '10')
model_type: Model type ('lite' or 'pro')
save_path: Directory to save generated video
timeout: Task timeout in seconds
interval: Status check interval in seconds
progress_callback: Optional progress callback function
Returns:
Dictionary with generation result
"""
result = {'status': False, 'video_path': '', 'video_url': '', 'msg': ''}
try:
# Check if image file exists
if not os.path.exists(image_path):
result['msg'] = f'Image file not found: {image_path}'
logger.error(result['msg'])
return result
# Step 1: Upload image to cloud storage
if progress_callback:
progress_callback("📤 正在上传图片到云存储...")
logger.info(f"Uploading image to cloud storage: {image_path}")
upload_result = self.cloud_storage.upload_file(image_path)
if not upload_result['status']:
result['msg'] = f"Failed to upload image: {upload_result['msg']}"
logger.error(result['msg'])
return result
img_url = upload_result['data']
if progress_callback:
progress_callback("✓ 图片上传成功")
logger.info(f"Image uploaded successfully: {img_url}")
# Step 2: Submit video generation task
if progress_callback:
progress_callback("🚀 正在提交视频生成任务...")
logger.info("Submitting video generation task...")
task_result = self.api_client.submit_task(prompt, img_url, duration, model_type)
if not task_result['status']:
result['msg'] = f"Failed to submit task: {task_result['msg']}"
logger.error(result['msg'])
return result
task_id = task_result['data']
if progress_callback:
progress_callback(f"✓ 任务提交成功，任务ID: {task_id}")
logger.info(f"Task submitted successfully, task ID: {task_id}")
# Step 3: Wait for task completion
if progress_callback:
progress_callback("⏳ 正在等待视频生成完成...")
logger.info("Waiting for video generation to complete...")
completion_result = self.api_client.wait_for_completion(
task_id,
timeout=timeout,
interval=interval,
progress_callback=progress_callback
)
if not completion_result['status']:
result['msg'] = f"Video generation failed: {completion_result['msg']}"
logger.error(result['msg'])
return result
video_url = completion_result['data']
result['video_url'] = video_url
logger.info(f"Video generated successfully: {video_url}")
# Step 4: Download video if save_path is provided
if save_path:
if progress_callback:
progress_callback("📥 正在下载视频到本地...")
logger.info("Downloading video to local storage...")
video_path = self.cloud_storage.download_file(video_url, save_path)
if video_path and os.path.exists(video_path):
result['status'] = True
result['video_path'] = video_path
result['msg'] = '视频生成并下载成功'
if progress_callback:
progress_callback(f"✓ 视频下载成功: {os.path.basename(video_path)}")
logger.info(f"Video downloaded successfully: {video_path}")
else:
result['msg'] = '视频下载失败'
if progress_callback:
progress_callback("✗ 视频下载失败")
logger.error(result['msg'])
else:
result['status'] = True
result['video_path'] = video_url
result['msg'] = '视频生成成功'
except Exception as e:
result['msg'] = f'处理过程中发生异常: {str(e)}'
logger.error(result['msg'])
return result
def batch_generate_videos(self,
image_folder: str,
prompts: List[str],
output_folder: str,
duration: str = '5',
model_type: str = 'lite',
timeout: int = 300,
interval: int = 3,
progress_callback: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
"""
Batch generate videos from multiple images.
Args:
image_folder: Folder containing input images
prompts: List of prompts (will cycle through them)
output_folder: Folder to save generated videos
duration: Video duration ('5' or '10')
model_type: Model type ('lite' or 'pro')
timeout: Task timeout in seconds
interval: Status check interval in seconds
progress_callback: Optional progress callback function
Returns:
Dictionary with batch processing result
"""
result = {'status': False, 'success_count': 0, 'failed_count': 0, 'results': [], 'msg': ''}
try:
if progress_callback:
progress_callback(f"开始批量处理任务...")
progress_callback(f" 图片文件夹: {image_folder}")
progress_callback(f" 视频保存目录: {output_folder}")
progress_callback(f" 视频时长: {duration}秒")
progress_callback(f" 模型类型: {model_type}")
# Find image files
image_extensions = ['*.jpg', '*.jpeg', '*.png', '*.bmp', '*.gif', '*.tiff', '*.webp']
image_files = []
for ext in image_extensions:
image_files.extend(glob.glob(os.path.join(image_folder, ext)))
image_files.extend(glob.glob(os.path.join(image_folder, ext.upper())))
if not image_files:
result['msg'] = f"No image files found in {image_folder}"
logger.error(result['msg'])
return result
if progress_callback:
progress_callback(f'目录下找到 {len(image_files)} 张图片')
if not prompts:
result['msg'] = "No prompts provided"
logger.error(result['msg'])
return result
if progress_callback:
progress_callback(f'解析到 {len(prompts)} 个提示词，将循环使用')
# Ensure output directory exists
os.makedirs(output_folder, exist_ok=True)
success_count = 0
failed_count = 0
# Process each image
for i, image_path in enumerate(image_files):
image_name = os.path.basename(image_path)
# Use cycling prompts
current_prompt = prompts[i % len(prompts)]
if progress_callback:
progress_callback(f"[{i + 1}/{len(image_files)}] 处理图片: {image_name}")
progress_callback(f" 使用提示词[{i % len(prompts) + 1}]: {current_prompt[:50]}{'...' if len(current_prompt) > 50 else ''}")
try:
# Generate video for this image
video_result = self.generate_video_from_image(
image_path=image_path,
prompt=current_prompt,
duration=duration,
model_type=model_type,
save_path=output_folder,
timeout=timeout,
interval=interval,
progress_callback=progress_callback
)
if video_result and video_result.get('status'):
success_count += 1
video_path = video_result.get('video_path', '')
if progress_callback:
progress_callback(f" ✓ 视频生成成功: {os.path.basename(video_path)}")
else:
failed_count += 1
error_msg = video_result.get('msg', '未知错误') if video_result else '处理失败'
if progress_callback:
progress_callback(f" ✗ 视频生成失败: {error_msg}")
result['results'].append({
'image_path': image_path,
'prompt': current_prompt,
'result': video_result
})
except Exception as e:
failed_count += 1
error_msg = str(e)
if progress_callback:
progress_callback(f" ✗ 处理图片时出错: {error_msg}")
result['results'].append({
'image_path': image_path,
'prompt': current_prompt,
'result': {'status': False, 'msg': error_msg}
})
# Final statistics
result['success_count'] = success_count
result['failed_count'] = failed_count
result['status'] = success_count > 0
result['msg'] = f'处理完成！成功: {success_count}, 失败: {failed_count}'
if progress_callback:
progress_callback("=" * 50)
progress_callback(f"处理完成！成功: {success_count}, 失败: {failed_count}")
if success_count > 0:
progress_callback(f"生成的视频已保存到: {output_folder}")
except Exception as e:
result['msg'] = f'批量处理过程中发生错误: {str(e)}'
logger.error(result['msg'])
return result
def main():
"""Command line interface for AI video generation."""
import argparse
import json
parser = argparse.ArgumentParser(description="AI Video Generator")
parser.add_argument("--action", required=True,
choices=["single", "batch"],
help="Action to perform")
parser.add_argument("--image", help="Image file path (for single)")
parser.add_argument("--folder", help="Image folder path (for batch)")
parser.add_argument("--prompt", help="Generation prompt")
parser.add_argument("--prompts", help="JSON string of prompts list (for batch)")
parser.add_argument("--output", help="Output directory")
parser.add_argument("--duration", default="5", choices=["5", "10"], help="Video duration")
parser.add_argument("--model", default="lite", choices=["lite", "pro"], help="Model type")
parser.add_argument("--timeout", type=int, default=180, help="Task timeout in seconds")
args = parser.parse_args()
try:
generator = VideoGenerator()
def progress_callback(message):
print(message)
if args.action == "single":
if not args.image or not args.prompt or not args.output:
raise ValueError("Single mode requires --image, --prompt, and --output")
result = generator.generate_video_from_image(
image_path=args.image,
prompt=args.prompt,
duration=args.duration,
model_type=args.model,
save_path=args.output,
timeout=args.timeout,
progress_callback=progress_callback
)
elif args.action == "batch":
if not args.folder or not args.prompts or not args.output:
raise ValueError("Batch mode requires --folder, --prompts, and --output")
prompts = json.loads(args.prompts)
result = generator.batch_generate_videos(
image_folder=args.folder,
prompts=prompts,
output_folder=args.output,
duration=args.duration,
model_type=args.model,
timeout=args.timeout,
progress_callback=progress_callback
)
print(json.dumps(result, ensure_ascii=False, indent=2))
except Exception as e:
error_result = {
"status": False,
"error": str(e)
}
print(json.dumps(error_result, ensure_ascii=False, indent=2))
sys.exit(1)
if __name__ == "__main__":
main()