diff --git a/python_core/ai_video/jsonrpc.py b/python_core/ai_video/jsonrpc.py new file mode 100644 index 0000000..7dc2bde --- /dev/null +++ b/python_core/ai_video/jsonrpc.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +JSON-RPC Communication Module +JSON-RPC 通信模块 + +Provides standardized JSON-RPC 2.0 communication for Tauri-Python integration. +""" + +import json +import sys +import time +from typing import Any, Dict, Optional, Union + + +class JSONRPCResponse: + """JSON-RPC 2.0 Response handler""" + + def __init__(self, request_id: Optional[Union[str, int]] = None): + self.request_id = request_id + + def success(self, result: Any) -> None: + """Send successful response""" + response = { + "jsonrpc": "2.0", + "id": self.request_id, + "result": result + } + self._send_response(response) + + def error(self, code: int, message: str, data: Any = None) -> None: + """Send error response""" + error_obj = { + "code": code, + "message": message + } + if data is not None: + error_obj["data"] = data + + response = { + "jsonrpc": "2.0", + "id": self.request_id, + "error": error_obj + } + self._send_response(response) + + def notification(self, method: str, params: Any = None) -> None: + """Send notification (no response expected)""" + notification = { + "jsonrpc": "2.0", + "method": method, + "params": params + } + self._send_response(notification) + + def _send_response(self, response: Dict[str, Any]) -> None: + """Send JSON-RPC response to stdout""" + try: + json_str = json.dumps(response, ensure_ascii=False, separators=(',', ':')) + print(f"JSONRPC:{json_str}") + sys.stdout.flush() + except Exception as e: + # Fallback error response + fallback = { + "jsonrpc": "2.0", + "id": self.request_id, + "error": { + "code": -32603, + "message": "Internal error", + "data": str(e) + } + } + print(f"JSONRPC:{json.dumps(fallback)}") + sys.stdout.flush() + + +class ProgressReporter: + """Progress reporting using JSON-RPC notifications""" + + def __init__(self): + self.rpc = JSONRPCResponse() + + def report(self, step: str, progress: float, message: str, details: Dict[str, Any] = None) -> None: + """Report progress using JSON-RPC notification""" + params = { + "step": step, + "progress": progress, # 0.0 to 1.0 + "message": message, + "timestamp": time.time() + } + if details: + params["details"] = details + + self.rpc.notification("progress", params) + + def step(self, step_name: str, message: str) -> None: + """Report a step without specific progress""" + self.report(step_name, -1, message) + + def complete(self, message: str = "完成") -> None: + """Report completion""" + self.report("complete", 1.0, message) + + def error(self, message: str, error_details: Dict[str, Any] = None) -> None: + """Report error""" + self.report("error", -1, message, error_details) + + +# Error codes following JSON-RPC 2.0 specification +class JSONRPCError: + PARSE_ERROR = -32700 + INVALID_REQUEST = -32600 + METHOD_NOT_FOUND = -32601 + INVALID_PARAMS = -32602 + INTERNAL_ERROR = -32603 + + # Custom application errors (start from -32000) + FILE_NOT_FOUND = -32001 + UPLOAD_FAILED = -32002 + GENERATION_FAILED = -32003 + DOWNLOAD_FAILED = -32004 + TIMEOUT_ERROR = -32005 + + +def create_response_handler(request_id: Optional[Union[str, int]] = None) -> JSONRPCResponse: + """Create a JSON-RPC response handler""" + return JSONRPCResponse(request_id) + + +def create_progress_reporter() -> ProgressReporter: + """Create a progress reporter""" + return ProgressReporter() + + +def parse_request(request_str: str) -> Dict[str, Any]: + """Parse JSON-RPC request string""" + try: + request = json.loads(request_str) + if not isinstance(request, dict): + raise ValueError("Request must be a JSON object") + + # Validate JSON-RPC 2.0 format + if request.get("jsonrpc") != "2.0": + raise ValueError("Invalid JSON-RPC version") + + if "method" not in request: + raise ValueError("Missing method field") + + return request + except json.JSONDecodeError as e: + raise ValueError(f"Invalid JSON: {e}") + + +# Example usage functions +def example_video_generation(): + """Example of how to use JSON-RPC for video generation""" + rpc = create_response_handler("video_gen_001") + progress = create_progress_reporter() + + try: + # Report progress steps + progress.step("upload", "[1/4] 正在上传图片到云存储...") + # ... upload logic ... + + progress.step("submit", "[2/4] 正在提交视频生成任务...") + # ... submit logic ... + + progress.step("wait", "[3/4] 正在等待视频生成完成...") + # ... wait logic ... + + progress.step("download", "[4/4] 正在下载视频到本地...") + # ... download logic ... + + progress.complete("[完成] 视频生成并下载成功") + + # Send final result + result = { + "status": True, + "video_path": "/path/to/video.mp4", + "video_url": "https://example.com/video.mp4", + "message": "视频生成并下载成功" + } + rpc.success(result) + + except Exception as e: + progress.error(f"生成失败: {str(e)}") + rpc.error(JSONRPCError.GENERATION_FAILED, "Video generation failed", str(e)) + + +if __name__ == "__main__": + # Test the JSON-RPC module + example_video_generation() diff --git a/python_core/ai_video/video_generator.py b/python_core/ai_video/video_generator.py index 65dce00..e9579fe 100644 --- a/python_core/ai_video/video_generator.py +++ b/python_core/ai_video/video_generator.py @@ -28,10 +28,12 @@ except ImportError: try: from .cloud_storage import CloudStorage from .api_client import APIClient + from .jsonrpc import create_response_handler, create_progress_reporter, JSONRPCError except ImportError: # Fallback for when running as script from cloud_storage import CloudStorage from api_client import APIClient + from jsonrpc import create_response_handler, create_progress_reporter, JSONRPCError logger = setup_logger(__name__) @@ -55,7 +57,7 @@ class VideoGenerator: else: self.cloud_storage = CloudStorage() - def generate_video_from_image(self, + def generate_video_from_image(self, image_path: str, prompt: str, duration: str = '5', @@ -63,7 +65,8 @@ class VideoGenerator: save_path: str = None, timeout: int = 180, interval: int = 2, - progress_callback: Optional[Callable[[str], None]] = None) -> Dict[str, Any]: + progress_callback: Optional[Callable[[str], None]] = None, + request_id: str = None) -> Dict[str, Any]: """ Generate video from a single image. @@ -80,8 +83,12 @@ class VideoGenerator: Returns: Dictionary with generation result """ + # Initialize JSON-RPC handlers + rpc = create_response_handler(request_id) + progress = create_progress_reporter() + result = {'status': False, 'video_path': '', 'video_url': '', 'msg': ''} - + try: # Check if image file exists if not os.path.exists(image_path): @@ -108,6 +115,7 @@ class VideoGenerator: return result # Step 1: Upload image to cloud storage + progress.step("upload", "[1/4] 正在上传图片到云存储...") if progress_callback: progress_callback("[1/4] 正在上传图片到云存储...") logger.info(f"Uploading image to cloud storage: {image_path}") @@ -116,9 +124,12 @@ class VideoGenerator: if not upload_result['status']: result['msg'] = f"Failed to upload image: {upload_result['msg']}" logger.error(result['msg']) + progress.error(result['msg']) + rpc.error(JSONRPCError.UPLOAD_FAILED, "Image upload failed", result['msg']) return result - + img_url = upload_result['data'] + progress.step("upload_success", "[1/4] 图片上传成功") if progress_callback: progress_callback("[1/4] 图片上传成功") logger.info(f"Image uploaded successfully: {img_url}") @@ -374,7 +385,8 @@ def main(): model_type=args.model, save_path=args.output, timeout=args.timeout, - progress_callback=progress_callback + progress_callback=progress_callback, + request_id="cli_request" ) elif args.action == "batch": diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index a485a7a..406cdcc 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -21,6 +21,7 @@ pub struct BatchAIVideoRequest { } use std::process::Command; use tauri_plugin_shell::ShellExt; +use serde_json; #[derive(Debug, Serialize, Deserialize)] pub struct VideoProcessRequest { @@ -114,7 +115,8 @@ async fn execute_python_command(app: tauri::AppHandle, args: &[String]) -> Resul let mut stdout_lines = Vec::new(); let mut stderr_lines = Vec::new(); - let mut json_output = String::new(); + let mut final_result: Option = None; + let mut progress_messages = Vec::new(); // Collect output from the process while let Some(event) = rx.recv().await { @@ -124,9 +126,27 @@ async fn execute_python_command(app: tauri::AppHandle, args: &[String]) -> Resul println!("Python stdout: {}", line); stdout_lines.push(line.to_string()); - // Try to extract JSON from the line - if line.trim().starts_with('{') && line.trim().ends_with('}') { - json_output = line.trim().to_string(); + // Parse JSON-RPC messages + if line.starts_with("JSONRPC:") { + let json_str = &line[8..]; // Remove "JSONRPC:" prefix + if let Ok(json_value) = serde_json::from_str::(json_str) { + if let Some(method) = json_value.get("method") { + if method == "progress" { + progress_messages.push(json_value); + println!("Progress: {}", json_str); + } + } else if json_value.get("result").is_some() { + // This is a final result + final_result = Some(json_str.to_string()); + println!("Final result: {}", json_str); + } + } + } else if line.trim().starts_with('{') && line.trim().ends_with('}') { + // Fallback: try to parse as direct JSON result + if let Ok(_) = serde_json::from_str::(line.trim()) { + final_result = Some(line.trim().to_string()); + println!("Direct JSON result: {}", line.trim()); + } } } CommandEvent::Stderr(data) => { @@ -157,19 +177,30 @@ async fn execute_python_command(app: tauri::AppHandle, args: &[String]) -> Resul // If no termination event was captured, assume success if we have output if exit_code.is_none() { - success = !stdout_lines.is_empty() || !json_output.is_empty(); + success = !stdout_lines.is_empty() || final_result.is_some(); exit_code = Some(if success { 0 } else { 1 }); } println!("Python script exit code: {:?}", exit_code); if success { - // Return JSON output if found, otherwise return full stdout - if !json_output.is_empty() { - println!("Extracted JSON: {}", json_output); - Ok(json_output) + // Return final result if found, otherwise return full stdout + if let Some(result) = final_result { + println!("Extracted JSON-RPC result: {}", result); + Ok(result) } else { - Ok(stdout_lines.join("\n")) + // Fallback: try to find JSON in the combined stdout + let full_output = stdout_lines.join("\n"); + if let Some(json_start) = full_output.find('{') { + if let Some(json_end) = full_output.rfind('}') { + if json_end > json_start { + let json_str = &full_output[json_start..=json_end]; + println!("Extracted JSON from full output: {}", json_str); + return Ok(json_str.to_string()); + } + } + } + Ok(full_output) } } else { let error_msg = format!(