This commit is contained in:
949036910@qq.com
2026-05-28 20:59:42 +08:00
parent d71a654cf1
commit fa467b65b6
6 changed files with 116 additions and 20 deletions

View File

@@ -23,8 +23,9 @@
use serde::Serialize; use serde::Serialize;
use serde_json::Value; use serde_json::Value;
use std::collections::HashMap;
use tauri::{AppHandle, Emitter, Window}; use tauri::{AppHandle, Emitter, Window};
use tokio::sync::mpsc; use tokio::sync::{mpsc, watch, Mutex};
use crate::app_config::AppConfig; use crate::app_config::AppConfig;
use crate::js_runtime::PipelineEvent; use crate::js_runtime::PipelineEvent;
@@ -32,6 +33,11 @@ use crate::nodejs;
const EVENT_NAME: &str = "nodejs:event"; const EVENT_NAME: &str = "nodejs:event";
#[derive(Default)]
pub struct NodeTaskRegistry {
inner: Mutex<HashMap<String, watch::Sender<bool>>>,
}
#[derive(Debug, Serialize, Clone)] #[derive(Debug, Serialize, Clone)]
#[serde(tag = "type", rename_all = "camelCase")] #[serde(tag = "type", rename_all = "camelCase")]
pub enum FrontendEvent { pub enum FrontendEvent {
@@ -102,16 +108,33 @@ pub async fn run_nodejs_script(
app: AppHandle, app: AppHandle,
window: Window, window: Window,
app_config: tauri::State<'_, AppConfig>, app_config: tauri::State<'_, AppConfig>,
registry: tauri::State<'_, NodeTaskRegistry>,
script_name: String, script_name: String,
params: Value, params: Value,
) -> Result<Value, String> { ) -> Result<Value, String> {
let (cancel_tx, cancel_rx) = watch::channel(false);
{
let mut map = registry.inner.lock().await;
if map.contains_key(&script_name) {
return Err(format!("脚本正在运行: {script_name}"));
}
map.insert(script_name.clone(), cancel_tx);
}
let (tx, rx) = mpsc::unbounded_channel::<PipelineEvent>(); let (tx, rx) = mpsc::unbounded_channel::<PipelineEvent>();
let pump = spawn_event_pump(app, window, script_name.clone(), rx); let pump = spawn_event_pump(app, window, script_name.clone(), rx);
let config_env = app_config.env_for_node().await; let config_env = app_config.env_for_node().await;
let result = nodejs::run_node_script(script_name, params, tx, config_env) let result = nodejs::run_node_script(script_name.clone(), params, tx, config_env, cancel_rx)
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string());
{
let mut map = registry.inner.lock().await;
map.remove(&script_name);
}
let result = result?;
pump.await.ok(); pump.await.ok();
Ok(result) Ok(result)
@@ -123,16 +146,34 @@ pub async fn run_nodejs_script_source(
app: AppHandle, app: AppHandle,
window: Window, window: Window,
app_config: tauri::State<'_, AppConfig>, app_config: tauri::State<'_, AppConfig>,
registry: tauri::State<'_, NodeTaskRegistry>,
script_source: String, script_source: String,
params: Value, params: Value,
) -> Result<Value, String> { ) -> Result<Value, String> {
let script_key = "<debug-source>".to_string();
let (cancel_tx, cancel_rx) = watch::channel(false);
{
let mut map = registry.inner.lock().await;
if map.contains_key(&script_key) {
return Err("调试脚本正在运行".to_string());
}
map.insert(script_key.clone(), cancel_tx);
}
let (tx, rx) = mpsc::unbounded_channel::<PipelineEvent>(); let (tx, rx) = mpsc::unbounded_channel::<PipelineEvent>();
let pump = spawn_event_pump(app, window, "<debug-source>".to_string(), rx); let pump = spawn_event_pump(app, window, script_key.clone(), rx);
let config_env = app_config.env_for_node().await; let config_env = app_config.env_for_node().await;
let result = nodejs::run_node_script_source(script_source, params, tx, config_env) let result = nodejs::run_node_script_source(script_source, params, tx, config_env, cancel_rx)
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string());
{
let mut map = registry.inner.lock().await;
map.remove(&script_key);
}
let result = result?;
pump.await.ok(); pump.await.ok();
Ok(result) Ok(result)
@@ -143,3 +184,19 @@ pub async fn run_nodejs_script_source(
pub async fn list_nodejs_scripts() -> Result<Vec<String>, String> { pub async fn list_nodejs_scripts() -> Result<Vec<String>, String> {
nodejs::list_scripts().await.map_err(|e| e.to_string()) nodejs::list_scripts().await.map_err(|e| e.to_string())
} }
#[tauri::command]
pub async fn stop_nodejs_script(
registry: tauri::State<'_, NodeTaskRegistry>,
script_name: String,
) -> Result<bool, String> {
let sender = {
let map = registry.inner.lock().await;
map.get(&script_name).cloned()
};
if let Some(tx) = sender {
tx.send(true).map_err(|e| e.to_string())?;
return Ok(true);
}
Ok(false)
}

View File

@@ -3,6 +3,7 @@
use serde::Deserialize; use serde::Deserialize;
use serde_json::{json, Value}; use serde_json::{json, Value};
use tauri::{AppHandle, Manager, State, Window}; use tauri::{AppHandle, Manager, State, Window};
use tokio::sync::watch;
use crate::app_config::AppConfig; use crate::app_config::AppConfig;
use crate::publish_db::PlatformAccountRecord; use crate::publish_db::PlatformAccountRecord;
@@ -64,6 +65,7 @@ async fn run_publish_script(
params, params,
tx, tx,
config_env, config_env,
watch::channel(false).1,
) )
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;

View File

@@ -44,6 +44,7 @@ pub fn run() {
.manage(audio_store::AudioStore::new()) .manage(audio_store::AudioStore::new())
.manage(video_store::VideoStore::new()) .manage(video_store::VideoStore::new())
.manage(publish_store::PublishStore::new()) .manage(publish_store::PublishStore::new())
.manage(commands::nodejs::NodeTaskRegistry::default())
.setup(|app| { .setup(|app| {
let handle = app.handle().clone(); let handle = app.handle().clone();
@@ -129,6 +130,7 @@ pub fn run() {
commands::nodejs::run_nodejs_script, commands::nodejs::run_nodejs_script,
commands::nodejs::run_nodejs_script_source, commands::nodejs::run_nodejs_script_source,
commands::nodejs::list_nodejs_scripts, commands::nodejs::list_nodejs_scripts,
commands::nodejs::stop_nodejs_script,
commands::fs_util::read_local_file_base64, commands::fs_util::read_local_file_base64,
commands::fs_util::get_media_duration_seconds, commands::fs_util::get_media_duration_seconds,
commands::fs_util::copy_local_file, commands::fs_util::copy_local_file,

View File

@@ -28,6 +28,7 @@ use thiserror::Error;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command; use tokio::process::Command;
use tokio::sync::mpsc::UnboundedSender; use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::watch;
use uuid::Uuid; use uuid::Uuid;
use crate::js_runtime::PipelineEvent; use crate::js_runtime::PipelineEvent;
@@ -66,6 +67,8 @@ pub enum NodeError {
NoResult, NoResult,
#[error("json: {0}")] #[error("json: {0}")]
Json(#[from] serde_json::Error), Json(#[from] serde_json::Error),
#[error("任务已取消")]
Cancelled,
} }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -346,11 +349,12 @@ pub async fn run_node_script(
params: Value, params: Value,
events: UnboundedSender<PipelineEvent>, events: UnboundedSender<PipelineEvent>,
config_env: HashMap<String, String>, config_env: HashMap<String, String>,
mut cancel_rx: watch::Receiver<bool>,
) -> Result<Value, NodeError> { ) -> Result<Value, NodeError> {
let entry = normalize_script_name(&script_name) let entry = normalize_script_name(&script_name)
.ok_or_else(|| NodeError::FetchScript(format!("无效的脚本名称: {script_name}")))?; .ok_or_else(|| NodeError::FetchScript(format!("无效的脚本名称: {script_name}")))?;
let modules = resolve_script_bundle(&entry).await?; let modules = resolve_script_bundle(&entry).await?;
run_script_bundle(&entry, modules, params, events, config_env).await run_script_bundle(&entry, modules, params, events, config_env, &mut cancel_rx).await
} }
/// 调试入口:直接传源码;其中 `require('*.js')` 仍从服务端拉取依赖。 /// 调试入口:直接传源码;其中 `require('*.js')` 仍从服务端拉取依赖。
@@ -359,9 +363,18 @@ pub async fn run_node_script_source(
params: Value, params: Value,
events: UnboundedSender<PipelineEvent>, events: UnboundedSender<PipelineEvent>,
config_env: HashMap<String, String>, config_env: HashMap<String, String>,
mut cancel_rx: watch::Receiver<bool>,
) -> Result<Value, NodeError> { ) -> Result<Value, NodeError> {
let modules = resolve_script_bundle_for_inline(&source).await?; let modules = resolve_script_bundle_for_inline(&source).await?;
run_script_bundle(INLINE_ENTRY, modules, params, events, config_env).await run_script_bundle(
INLINE_ENTRY,
modules,
params,
events,
config_env,
&mut cancel_rx,
)
.await
} }
async fn run_script_bundle( async fn run_script_bundle(
@@ -370,6 +383,7 @@ async fn run_script_bundle(
params: Value, params: Value,
events: UnboundedSender<PipelineEvent>, events: UnboundedSender<PipelineEvent>,
config_env: HashMap<String, String>, config_env: HashMap<String, String>,
cancel_rx: &mut watch::Receiver<bool>,
) -> Result<Value, NodeError> { ) -> Result<Value, NodeError> {
let node = locate_node().ok_or(NodeError::NotFound)?; let node = locate_node().ok_or(NodeError::NotFound)?;
let node_modules_cwd = nodejs_dir().unwrap_or_else(std::env::temp_dir); let node_modules_cwd = nodejs_dir().unwrap_or_else(std::env::temp_dir);
@@ -378,6 +392,7 @@ async fn run_script_bundle(
let mut cmd = Command::new(&node); let mut cmd = Command::new(&node);
cmd.arg(&script_path) cmd.arg(&script_path)
.kill_on_drop(true)
.current_dir(&bundle_dir) .current_dir(&bundle_dir)
.env( .env(
"NODE_PATH", "NODE_PATH",
@@ -491,18 +506,26 @@ async fn run_script_bundle(
result result
}); });
// 与 stdout/stderr 读取并行等待避免子进程在管道未排空时退出Windows libuv // 与取消信号竞争等待;取消时立即杀掉子进程
let (status, stderr_tail, result_line) = tokio::join!( let (status, cancelled) = tokio::select! {
child.wait(), status = child.wait() => (status?, false),
stderr_task, _ = wait_for_cancel(cancel_rx) => {
stdout_task, let _ = child.start_kill();
); let status = child.wait().await?;
let status = status?; (status, true)
}
};
let (stderr_tail, result_line) = tokio::join!(stderr_task, stdout_task);
let stderr_tail = stderr_tail.unwrap_or_default(); let stderr_tail = stderr_tail.unwrap_or_default();
let result_line = result_line.unwrap_or(None); let result_line = result_line.unwrap_or(None);
let _ = tokio::fs::remove_dir_all(&bundle_dir).await; let _ = tokio::fs::remove_dir_all(&bundle_dir).await;
if cancelled {
return Err(NodeError::Cancelled);
}
if !status.success() { if !status.success() {
return Err(NodeError::NonZero { return Err(NodeError::NonZero {
status: status.code(), status: status.code(),
@@ -514,3 +537,14 @@ async fn run_script_bundle(
let v: Value = serde_json::from_str(&result_str)?; let v: Value = serde_json::from_str(&result_str)?;
Ok(v) Ok(v)
} }
async fn wait_for_cancel(cancel_rx: &mut watch::Receiver<bool>) {
if *cancel_rx.borrow() {
return;
}
while cancel_rx.changed().await.is_ok() {
if *cancel_rx.borrow() {
return;
}
}
}

View File

@@ -1,6 +1,7 @@
<script setup> <script setup>
import { ref } from "vue"; import { ref } from "vue";
import { storeToRefs } from "pinia"; import { storeToRefs } from "pinia";
import { invoke } from "@tauri-apps/api/core";
import { useWorkflowStore, executeVideoRewrite } from "../../stores/workflow.js"; import { useWorkflowStore, executeVideoRewrite } from "../../stores/workflow.js";
const workflow = useWorkflowStore(); const workflow = useWorkflowStore();
@@ -24,6 +25,7 @@ function onCancel() {
feedback.value = { severity: "", message: "" }; feedback.value = { severity: "", message: "" };
if (videoRewriting.value) { if (videoRewriting.value) {
workflow.videoRewriteAbortRequested = true; workflow.videoRewriteAbortRequested = true;
invoke("stop_nodejs_script", { scriptName: "douyin_pipeline.js" }).catch(() => {});
workflow.videoRewriting = false; workflow.videoRewriting = false;
workflow.rewriteProgress = ""; workflow.rewriteProgress = "";
} }
@@ -31,11 +33,7 @@ function onCancel() {
} }
function onVisibleChange(visible) { function onVisibleChange(visible) {
if (visible) { if (!visible) onCancel();
workflow.videoLinkModalVisible = true;
return;
}
onCancel();
} }
async function onRewrite() { async function onRewrite() {

View File

@@ -856,6 +856,9 @@ export async function executeVideoRewrite(store) {
}; };
} catch (err) { } catch (err) {
const msg = err instanceof Error ? err.message : String(err || "未知错误"); const msg = err instanceof Error ? err.message : String(err || "未知错误");
if (store.videoRewriteAbortRequested || String(msg).includes("任务已取消")) {
return { ok: false, cancelled: true, message: "已取消仿写任务" };
}
return { ok: false, message: `仿写失败: ${msg}` }; return { ok: false, message: `仿写失败: ${msg}` };
} finally { } finally {
if (typeof unlisten === "function") { if (typeof unlisten === "function") {