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

feat: 更通用的 带进度条的 python通信方式

Browse Source
This commit is contained in:
root
2025-07-10 22:01:59 +08:00
parent f9e67bfc3d
commit bda9eb1a6b
6 changed files with 448 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

242
docs/python_executor_progress.md Normal file
View File

@@ -0,0 +1,242 @@
# Python Executor with Progress Support
通用的Python执行器，支持实时进度回调和JSON-RPC通信。
## 功能特性
- ✅ 支持多种Python命令 (python, python3, py)
- ✅ UTF-8编码处理
- ✅ 并发stdout/stderr读取
- ✅ JSON-RPC消息解析
- ✅ 实时进度回调
- ✅ Tauri事件发射
- ✅ 超时处理
- ✅ 错误处理
## 使用方法
### 1. 基本用法（无进度）
```rust
use crate::python_executor::execute_python_command;
#[tauri::command]
pub async fn my_command(app: AppHandle) -> Result<String, String> {
let args = vec![
"-m".to_string(),
"my_python_module".to_string(),
"--action".to_string(),
"do_something".to_string(),
];
execute_python_command(app, &args, None).await
}
```
### 2. 使用Tauri事件进度回调
```rust
use crate::python_executor::execute_python_with_events;
#[tauri::command]
pub async fn my_command_with_progress(app: AppHandle) -> Result<String, String> {
let args = vec![
"-m".to_string(),
"my_python_module".to_string(),
"--action".to_string(),
"long_running_task".to_string(),
];
// 进度将通过 "my-task-progress" 事件发送到前端
execute_python_with_events(app, &args, None, "my-task-progress").await
}
```
### 3. 使用函数回调
```rust
use crate::python_executor::{execute_python_with_callback, PythonProgress};
#[tauri::command]
pub async fn my_command_with_callback(app: AppHandle) -> Result<String, String> {
let args = vec![
"-m".to_string(),
"my_python_module".to_string(),
"--action".to_string(),
"process_data".to_string(),
];
// 自定义进度处理
let progress_callback = |progress: PythonProgress| {
println!("Progress: {}% - {}", progress.progress, progress.message);
// 可以在这里添加自定义逻辑，如数据库记录、日志等
};
execute_python_with_callback(app, &args, None, progress_callback).await
}
```
### 4. 高级用法（自定义回调）
```rust
use crate::python_executor::{execute_python_command_with_progress, ProgressCallback, PythonProgress};
struct CustomProgressHandler {
task_id: String,
}
impl ProgressCallback for CustomProgressHandler {
fn on_progress(&self, progress: PythonProgress) {
// 自定义进度处理逻辑
println!("Task {}: {}% - {}", self.task_id, progress.progress, progress.message);
// 可以发送到数据库、文件、其他服务等
if progress.step == "complete" {
println!("Task {} completed!", self.task_id);
}
}
}
#[tauri::command]
pub async fn my_advanced_command(app: AppHandle, task_id: String) -> Result<String, String> {
let args = vec![
"-m".to_string(),
"my_python_module".to_string(),
"--task-id".to_string(),
task_id.clone(),
];
let progress_handler = CustomProgressHandler { task_id };
execute_python_command_with_progress(app, &args, None, Some(progress_handler)).await
}
```
## Python端实现
Python脚本需要使用JSON-RPC协议发送进度信息：
```python
from python_core.utils.jsonrpc import create_response_handler, create_progress_reporter
def my_long_running_task():
rpc = create_response_handler("task_id")
progress = create_progress_reporter()
try:
# 报告开始
progress.step("start", "开始处理任务...")
# 处理步骤1
progress.report("step1", 25.0, "正在处理步骤1...")
# ... 实际处理逻辑 ...
# 处理步骤2
progress.report("step2", 50.0, "正在处理步骤2...")
# ... 实际处理逻辑 ...
# 处理步骤3
progress.report("step3", 75.0, "正在处理步骤3...")
# ... 实际处理逻辑 ...
# 完成
progress.complete("任务完成！")
# 发送最终结果
result = {
"status": True,
"message": "任务成功完成",
"data": {...}
}
rpc.success(result)
except Exception as e:
progress.error(f"任务失败: {str(e)}")
rpc.error(-32603, "任务执行失败", str(e))
if __name__ == "__main__":
my_long_running_task()
```
## 前端使用
```typescript
import { listen } from '@tauri-apps/api/event'
// 监听进度事件
const unlisten = await listen('my-task-progress', (event) => {
const progress = event.payload as {
step: string
progress: number
message: string
details?: any
timestamp: number
}
console.log(`Progress: ${progress.progress}% - ${progress.message}`)
// 更新UI
updateProgressBar(progress.progress)
updateStatusMessage(progress.message)
})
// 执行命令
try {
const result = await invoke('my_command_with_progress')
console.log('Task completed:', result)
} finally {
unlisten() // 清理监听器
}
```
## 进度数据结构
```rust
pub struct PythonProgress {
pub step: String, // 当前步骤名称
pub progress: f64, // 进度百分比 (0-100, -1表示不确定)
pub message: String, // 进度消息
pub details: Option<serde_json::Value>, // 额外详情
pub timestamp: f64, // 时间戳
}
```
## 配置选项
```rust
pub struct PythonExecutorConfig {
pub timeout_seconds: u64, // 超时时间（秒）
pub working_directory: Option<PathBuf>, // 工作目录
}
// 默认配置：10分钟超时，当前目录
let config = PythonExecutorConfig::default();
// 自定义配置
let config = PythonExecutorConfig {
timeout_seconds: 1800, // 30分钟
working_directory: Some("/path/to/workdir".into()),
};
```
## 最佳实践
1. **选择合适的回调方式**：
- 简单任务：使用 `execute_python_command`
- 需要前端进度显示：使用 `execute_python_with_events`
- 需要自定义处理：使用 `execute_python_with_callback`
2. **Python端进度报告**：
- 使用有意义的步骤名称
- 提供清晰的进度消息
- 合理设置进度百分比
- 在关键节点报告进度
3. **错误处理**：
- Python端使用try-catch包装
- 通过JSON-RPC发送错误信息
- Rust端处理超时和进程错误
4. **性能考虑**：
- 不要过于频繁地报告进度
- 避免在进度回调中执行耗时操作
- 合理设置超时时间

68
src-tauri/src/commands/template.rs
View File

@@ -1,22 +1,16 @@
use serde::{Deserialize, Serialize};
use crate::python_executor::execute_python_command;
use tauri::{AppHandle, Emitter};
use crate::python_executor::{execute_python_command, execute_python_with_events};
use tauri::AppHandle;
#[derive(Debug, Deserialize)]
pub struct BatchImportRequest {
pub source_folder: String,
}
#[derive(Debug, Serialize, Clone)]
pub struct ImportProgress {
pub step: String,
pub progress: f64,
pub message: String,
pub current_template: Option<String>,
pub total_templates: Option<i32>,
pub processed_templates: Option<i32>,
pub timestamp: f64,
}
// Re-export PythonProgress as ImportProgress for backward compatibility
// Note: This is used by frontend TypeScript definitions
#[allow(unused_imports)]
pub use crate::python_executor::PythonProgress as ImportProgress;
#[derive(Debug, Serialize)]
pub struct TemplateInfo {
@@ -65,29 +59,8 @@ pub async fn batch_import_templates_with_progress(
request.source_folder,
];
// Create a progress callback that emits events to the frontend
let app_clone = app.clone();
let progress_callback = move |progress: ImportProgress| {
let _ = app_clone.emit("template-import-progress", &progress);
};
// Execute with progress monitoring
execute_python_command_with_progress(app, &args, None, progress_callback).await
}
async fn execute_python_command_with_progress<F>(
app: AppHandle,
args: &[String],
config: Option<crate::python_executor::PythonExecutorConfig>,
_progress_callback: F,
) -> Result<String, String>
where
F: Fn(ImportProgress) + Send + 'static,
{
// For now, we'll use the existing execute_python_command
// and parse progress from JSON-RPC notifications
// TODO: Enhance python_executor to support progress callbacks
execute_python_command(app, args, config).await
// Execute with progress monitoring via Tauri events
execute_python_with_events(app, &args, None, "template-import-progress").await
}
#[tauri::command]
@@ -129,3 +102,28 @@ pub async fn delete_template(app: tauri::AppHandle, template_id: String) -> Resu
execute_python_command(app, &args, None).await
}
// Example: AI Video generation with progress
#[allow(dead_code)]
#[tauri::command]
pub async fn generate_video_with_progress(
app: AppHandle,
image_path: String,
prompt: String,
output_path: String,
) -> Result<String, String> {
let args = vec![
"-m".to_string(),
"python_core.ai_video.video_generator".to_string(),
"--image_path".to_string(),
image_path,
"--prompt".to_string(),
prompt,
"--output_path".to_string(),
output_path,
];
// Execute with progress monitoring via Tauri events
// Frontend can listen to "video-generation-progress" event
execute_python_with_events(app, &args, None, "video-generation-progress").await
}

170
src-tauri/src/python_executor.rs
View File

@@ -4,6 +4,68 @@ use std::time::{Duration, Instant};
use std::thread;
use std::sync::{Arc, Mutex};
use std::sync::mpsc;
use serde::{Deserialize, Serialize};
use tauri::{AppHandle, Emitter};
/// Generic progress information from Python processes
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct PythonProgress {
pub step: String,
pub progress: f64, // -1 for indeterminate, 0-100 for percentage
pub message: String,
pub details: Option<serde_json::Value>,
pub timestamp: f64,
}
/// Progress callback trait for handling Python process progress
pub trait ProgressCallback: Send + Sync {
fn on_progress(&self, progress: PythonProgress);
}
/// Simple function-based progress callback
pub struct FunctionCallback<F>
where
F: Fn(PythonProgress) + Send + Sync,
{
callback: F,
}
impl<F> FunctionCallback<F>
where
F: Fn(PythonProgress) + Send + Sync,
{
#[allow(dead_code)]
pub fn new(callback: F) -> Self {
Self { callback }
}
}
impl<F> ProgressCallback for FunctionCallback<F>
where
F: Fn(PythonProgress) + Send + Sync,
{
fn on_progress(&self, progress: PythonProgress) {
(self.callback)(progress);
}
}
/// Tauri event-based progress callback
pub struct TauriEventCallback {
app: AppHandle,
event_name: String,
}
impl TauriEventCallback {
pub fn new(app: AppHandle, event_name: String) -> Self {
Self { app, event_name }
}
}
impl ProgressCallback for TauriEventCallback {
fn on_progress(&self, progress: PythonProgress) {
let _ = self.app.emit(&self.event_name, &progress);
}
}
/// Configuration for Python command execution
pub struct PythonExecutorConfig {
@@ -20,8 +82,29 @@ impl Default for PythonExecutorConfig {
}
}
/// Execute a Python command with the given arguments
///
/// Execute a Python command with progress callback support
///
/// This function handles:
/// - Multiple Python command attempts (python, python3, py)
/// - UTF-8 encoding setup
/// - Concurrent stdout/stderr reading
/// - JSON-RPC message parsing
/// - Progress notifications via callback
/// - Timeout handling
pub async fn execute_python_command_with_progress<P>(
app: tauri::AppHandle,
args: &[String],
config: Option<PythonExecutorConfig>,
progress_callback: Option<P>,
) -> Result<String, String>
where
P: ProgressCallback + 'static,
{
execute_python_internal(app, args, config, progress_callback).await
}
/// Execute a Python command with the given arguments (legacy interface)
///
/// This function handles:
/// - Multiple Python command attempts (python, python3, py)
/// - UTF-8 encoding setup
@@ -30,10 +113,49 @@ impl Default for PythonExecutorConfig {
/// - Timeout handling
/// - Detailed error reporting
pub async fn execute_python_command(
_app: tauri::AppHandle,
app: tauri::AppHandle,
args: &[String],
config: Option<PythonExecutorConfig>
) -> Result<String, String> {
execute_python_internal(app, args, config, None::<FunctionCallback<fn(PythonProgress)>>).await
}
/// Execute Python command with a simple function callback
#[allow(dead_code)]
pub async fn execute_python_with_callback<F>(
app: tauri::AppHandle,
args: &[String],
config: Option<PythonExecutorConfig>,
callback: F,
) -> Result<String, String>
where
F: Fn(PythonProgress) + Send + Sync + 'static,
{
let progress_callback = FunctionCallback::new(callback);
execute_python_command_with_progress(app, args, config, Some(progress_callback)).await
}
/// Execute Python command with Tauri event emission
pub async fn execute_python_with_events(
app: tauri::AppHandle,
args: &[String],
config: Option<PythonExecutorConfig>,
event_name: &str,
) -> Result<String, String> {
let progress_callback = TauriEventCallback::new(app.clone(), event_name.to_string());
execute_python_command_with_progress(app, args, config, Some(progress_callback)).await
}
/// Internal implementation for Python command execution
async fn execute_python_internal<P>(
_app: tauri::AppHandle,
args: &[String],
config: Option<PythonExecutorConfig>,
progress_callback: Option<P>,
) -> Result<String, String>
where
P: ProgressCallback + 'static,
{
let config = config.unwrap_or_default();
println!("Executing Python command with args: {:?}", args);
@@ -109,9 +231,13 @@ pub async fn execute_python_command(
let error_messages = Arc::new(Mutex::new(Vec::new()));
let final_result = Arc::new(Mutex::new(None::<String>));
// Wrap progress callback in Arc for thread sharing
let progress_callback = Arc::new(progress_callback);
// Spawn thread for reading stdout
let progress_messages_clone = Arc::clone(&progress_messages);
let final_result_clone = Arc::clone(&final_result);
let progress_callback_clone = Arc::clone(&progress_callback);
thread::spawn(move || {
let mut reader = BufReader::new(stdout);
let mut buffer = Vec::new();
@@ -137,9 +263,45 @@ pub async fn execute_python_command(
if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(json_str) {
if let Some(method) = json_value.get("method") {
if method == "progress" {
// Store progress message
if let Ok(mut messages) = progress_messages_clone.lock() {
messages.push(json_str.to_string());
}
// Call progress callback if available
if let Some(ref callback) = *progress_callback_clone {
if let Some(params) = json_value.get("params") {
// Try to parse as PythonProgress
if let Ok(progress) = serde_json::from_value::<PythonProgress>(params.clone()) {
callback.on_progress(progress);
} else {
// Fallback: create basic progress from available data
let step = params.get("step")
.and_then(|v| v.as_str())
.unwrap_or("unknown")
.to_string();
let message = params.get("message")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let progress_val = params.get("progress")
.and_then(|v| v.as_f64())
.unwrap_or(-1.0);
let progress = PythonProgress {
step,
progress: progress_val,
message,
details: Some(params.clone()),
timestamp: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs_f64(),
};
callback.on_progress(progress);
}
}
}
println!("Progress: {}", json_str);
}
} else if json_value.get("result").is_some() || json_value.get("error").is_some() {
@@ -148,10 +310,8 @@ pub async fn execute_python_command(
// Extract the result or error content from JSON-RPC
if let Some(result_content) = json_value.get("result") {
*result = Some(result_content.to_string());
println!("JSON-RPC result extracted: {}", result_content);
} else if let Some(error_content) = json_value.get("error") {
*result = Some(format!("{{\"status\":false,\"error\":{}}}", error_content));
println!("JSON-RPC error extracted: {}", error_content);
}
};
}

38
src-tauri/src/test_template.rs
View File

@@ -1,38 +0,0 @@
use crate::python_executor::execute_python_command;
#[tokio::test]
async fn test_get_templates() {
let app = tauri::test::mock_app();
let args = vec![
"-m".to_string(),
"python_core.services.template_manager".to_string(),
"--action".to_string(),
"get_templates".to_string(),
];
match execute_python_command(app, &args, None).await {
Ok(result) => {
println!("Result: {}", result);
// Parse the result to verify it's valid JSON
match serde_json::from_str::<serde_json::Value>(&result) {
Ok(json) => {
println!("Parsed JSON successfully");
if let Some(status) = json.get("status") {
println!("Status: {}", status);
}
if let Some(templates) = json.get("templates") {
if let Some(array) = templates.as_array() {
println!("Found {} templates", array.len());
}
}
}
Err(e) => {
println!("Failed to parse JSON: {}", e);
}
}
}
Err(e) => {
println!("Error: {}", e);
}
}
}

33
src/pages/TemplateManagePage.tsx
View File

@@ -33,11 +33,9 @@ interface ImportResult {
interface ImportProgress {
step: string
progress: number // 0-100
progress: number // -1 for indeterminate, 0-100 for percentage
message: string
currentTemplate?: string
totalTemplates?: number
processedTemplates?: number
details?: any // Additional details from Python
timestamp: number
}
@@ -66,9 +64,11 @@ const TemplateManagePage: React.FC = () => {
setTemplates(result.templates || [])
} else {
console.error('Failed to load templates:', result.msg)
setTemplates([])
}
} catch (error) {
console.error('Error loading templates:', error)
setTemplates([])
} finally {
setLoading(false)
}
@@ -93,25 +93,8 @@ const TemplateManagePage: React.FC = () => {
setImportLogs(prev => [...prev, `[${new Date().toLocaleTimeString()}] ${progress.message}`])
}
// Simulate initial progress
onProgress({
step: "scanning",
progress: 10,
message: "正在扫描模板文件夹...",
timestamp: Date.now()
})
// For now, use the regular import and simulate progress
// TODO: Switch to batchImportTemplatesWithProgress when backend is ready
const result = await TemplateService.batchImportTemplates(folderResult)
// Simulate completion progress
onProgress({
step: "complete",
progress: 100,
message: "模板导入完成",
timestamp: Date.now()
})
// Use the new progress-enabled import
const result = await TemplateService.batchImportTemplatesWithProgress(folderResult, onProgress)
setImportResult(result)
if (result.status && result.imported_count > 0) {
@@ -426,9 +409,9 @@ const TemplateManagePage: React.FC = () => {
></div>
</div>
<p className="text-sm text-gray-600 mt-2">{importProgress.message}</p>
{importProgress.processedTemplates !== undefined && importProgress.totalTemplates !== undefined && (
{importProgress.details?.processed_templates !== undefined && importProgress.details?.total_templates !== undefined && (
<p className="text-xs text-gray-500 mt-1">
已处理: {importProgress.processedTemplates} / {importProgress.totalTemplates}
已处理: {importProgress.details.processed_templates} / {importProgress.details.total_templates}
</p>
)}
</div>

30
test_template_loading.py
View File

@@ -1,30 +0,0 @@
#!/usr/bin/env python3
"""
Test script to verify template loading functionality
"""
import sys
import json
from python_core.services.template_manager import TemplateManager
from python_core.utils.jsonrpc import create_response_handler
def test_get_templates():
"""Test getting templates"""
rpc = create_response_handler("test")
try:
manager = TemplateManager()
templates = manager.get_templates()
result = {
"status": True,
"templates": [template.__dict__ for template in templates]
}
rpc.success(result)
except Exception as e:
rpc.error(-32603, f"Failed to get templates: {str(e)}")
if __name__ == "__main__":
test_get_templates()