diff --git a/.gitignore b/.gitignore index 4e90c93..bb73170 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,4 @@ dist-ssr *.sln *.sw? src-tauri/binaries/ +src-tauri/resources/nodejs/ diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 0114e08..cec1630 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -1,2 +1,3 @@ pub mod auth; +pub mod nodejs; pub mod quickjs; diff --git a/src-tauri/src/commands/nodejs.rs b/src-tauri/src/commands/nodejs.rs new file mode 100644 index 0000000..896215a --- /dev/null +++ b/src-tauri/src/commands/nodejs.rs @@ -0,0 +1,120 @@ +//! Node.js 流水线命令：脚本由后端 `/api/v1/nodejs-scripts` 下发， +//! 由打包在 `resources/nodejs/node.exe` 的 Node 解释器执行。 +//! +//! 前端用法示例： +//! +//! ```ignore +//! import { invoke } from "@tauri-apps/api/core"; +//! import { listen } from "@tauri-apps/api/event"; +//! +//! const off = await listen("nodejs:event", (e) => console.log(e.payload)); +//! const result = await invoke("run_nodejs_script", { +//! scriptName: "douyin_pipeline.js", +//! params: { shareText: "...", modelConfig: {...} }, +//! }); +//! off(); +//! ``` +//! +//! 与 QuickJS 流水线对比： +//! - 入口约定：`globalThis.__nodejsMain(params)`（QuickJS 那边是 `__quickjsMain`） +//! - 事件通道：`nodejs:event`（QuickJS 那边是 `quickjs:event`），payload 形状一致 +//! - __native 只暴露 `emitProgress` / `log` 两个能力，其它 IO / 网络 +//! 直接用 Node 自带 API 即可，无需 Rust 端再封一遍桥。 + +use serde::Serialize; +use serde_json::Value; +use tauri::{AppHandle, Emitter, Window}; +use tokio::sync::mpsc; + +use crate::js_runtime::PipelineEvent; +use crate::nodejs; + +const EVENT_NAME: &str = "nodejs:event"; + +#[derive(Debug, Serialize, Clone)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum FrontendEvent { + Progress { + #[serde(rename = "scriptName")] + script_name: String, + status: String, + }, + Log { + #[serde(rename = "scriptName")] + script_name: String, + level: String, + msg: String, + #[serde(skip_serializing_if = "Option::is_none")] + fields: Option, + }, +} + +fn spawn_event_pump( + app: AppHandle, + window: Window, + script_name: String, + mut rx: mpsc::UnboundedReceiver, +) -> tokio::task::JoinHandle<()> { + let win_label = window.label().to_string(); + tokio::spawn(async move { + while let Some(ev) = rx.recv().await { + let payload = match ev { + PipelineEvent::Progress(s) => FrontendEvent::Progress { + script_name: script_name.clone(), + status: s, + }, + PipelineEvent::Log { level, msg, fields } => FrontendEvent::Log { + script_name: script_name.clone(), + level, + msg, + fields, + }, + }; + let _ = app.emit_to(&win_label, EVENT_NAME, payload); + } + }) +} + +/// 按脚本名跑：去后端拉 `scriptName`，包装 runner，spawn 子进程，回送结果。 +#[tauri::command] +pub async fn run_nodejs_script( + app: AppHandle, + window: Window, + script_name: String, + params: Value, +) -> Result { + let (tx, rx) = mpsc::unbounded_channel::(); + let pump = spawn_event_pump(app, window, script_name.clone(), rx); + + let result = nodejs::run_node_script(script_name, params, tx) + .await + .map_err(|e| e.to_string())?; + + pump.await.ok(); + Ok(result) +} + +/// 调试：直接把传入的 source 当用户脚本喂给 runner，不走后端。 +#[tauri::command] +pub async fn run_nodejs_script_source( + app: AppHandle, + window: Window, + script_source: String, + params: Value, +) -> Result { + let (tx, rx) = mpsc::unbounded_channel::(); + let pump = spawn_event_pump(app, window, "".to_string(), rx); + + let result = nodejs::run_node_script_source(script_source, params, tx) + .await + .map_err(|e| e.to_string())?; + + pump.await.ok(); + Ok(result) +} + +/// 列出后端 `scripts/nodejs` 目录下的 `.js` 文件名。 +#[tauri::command] +pub async fn list_nodejs_scripts() -> Result, String> { + nodejs::list_scripts().await.map_err(|e| e.to_string()) +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index feadf8f..82ea289 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -5,6 +5,7 @@ pub mod commands; pub mod ffmpeg; pub mod http; pub mod js_runtime; +pub mod nodejs; pub mod oss; use tauri::Manager; @@ -28,6 +29,9 @@ pub fn run() { commands::quickjs::run_quickjs_script, commands::quickjs::run_quickjs_script_source, commands::quickjs::list_quickjs_scripts, + commands::nodejs::run_nodejs_script, + commands::nodejs::run_nodejs_script_source, + commands::nodejs::list_nodejs_scripts, ]) .run(tauri::generate_context!()) .expect("error while running tauri application"); diff --git a/src-tauri/src/nodejs.rs b/src-tauri/src/nodejs.rs new file mode 100644 index 0000000..bda898a --- /dev/null +++ b/src-tauri/src/nodejs.rs @@ -0,0 +1,392 @@ +//! Bundled Node.js sidecar runner. +//! +//! 与 QuickJS 路径完全对齐： +//! - 脚本源码由后端 `GET /api/v1/nodejs-scripts?name=.js` 下发， +//! 约定响应形如 `{ ok: true, data: { source: "..." } }`； +//! - 在 `src-tauri/resources/nodejs/node.exe`（开发期）/ 同目录 +//! `resources/nodejs/node.exe`（打包后）的 Node 解释器中执行； +//! - 进度 / 日志由用户脚本通过 `globalThis.__native.emitProgress(s)` +//! / `__native.log(level, msg, fields)` 写到 stderr，Rust 这边解析后 +//! 转成 `PipelineEvent` 推给前端； +//! - 用户脚本的最终结果通过赋值 / `return` 的方式（约定函数名为 +//! `globalThis.__nodejsMain(params)`）返回，runner 把 JSON.stringify 后 +//! 的结果以 `__RESULT__` 这一行写到 stdout，Rust 抓回来反序列化。 +//! +//! 这套协议有意保持极小：跟 QuickJS 的 `globalThis.__quickjsMain` / `__native` +//! 看起来完全对称，方便用户在两种运行时之间迁移脚本。 + +use std::path::PathBuf; +use std::process::Stdio; + +use serde_json::Value; +use thiserror::Error; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::process::Command; +use tokio::sync::mpsc::UnboundedSender; +use uuid::Uuid; + +use crate::js_runtime::PipelineEvent; + +#[derive(Debug, Error)] +pub enum NodeError { + #[error("node.exe 未找到（请检查 resources/nodejs/ 是否存在）")] + NotFound, + #[error("io: {0}")] + Io(#[from] std::io::Error), + #[error("拉取脚本失败: {0}")] + FetchScript(String), + #[error("node 进程非零退出 status={status:?}, stderr={stderr}")] + NonZero { status: Option, stderr: String }, + #[error("脚本未输出 __RESULT__ 行")] + NoResult, + #[error("json: {0}")] + Json(#[from] serde_json::Error), +} + +// --------------------------------------------------------------------------- +// 定位 node.exe +// --------------------------------------------------------------------------- +fn exe_name() -> &'static str { + if cfg!(windows) { "node.exe" } else { "node" } +} + +pub fn locate_node() -> Option { + if let Ok(p) = std::env::var("AICLIENT_NODE_PATH") { + let pb = PathBuf::from(p); + if pb.exists() { + return Some(pb); + } + } + if let Ok(exe) = std::env::current_exe() { + if let Some(dir) = exe.parent() { + let c = dir.join("resources").join("nodejs").join(exe_name()); + if c.exists() { + return Some(c); + } + } + } + let dev = PathBuf::from("src-tauri/resources/nodejs").join(exe_name()); + if dev.exists() { + return Some(dev); + } + let dev2 = PathBuf::from("resources/nodejs").join(exe_name()); + if dev2.exists() { + return Some(dev2); + } + Some(PathBuf::from(exe_name())) +} + +/// 用于设置 `cwd`：让用户脚本里的 `require('xxx')` 能解析到 +/// `resources/nodejs/node_modules/` 内置依赖。 +pub fn nodejs_dir() -> Option { + locate_node().and_then(|p| p.parent().map(|x| x.to_path_buf())) +} + +// --------------------------------------------------------------------------- +// Runner 模板 +// +// HEADER 安装 __native 桥，开始 IIFE 异步函数； +// 用户脚本原样夹在中间（通常用户脚本里会赋值 globalThis.__nodejsMain）； +// FOOTER 调用 __nodejsMain(params)、捕获异常、写 __RESULT__。 +// --------------------------------------------------------------------------- +const RUNNER_HEADER: &str = r#" +// ---- aiclient auto-generated nodejs runner (do not edit) ---- +"use strict"; +(function () { + function emit(obj) { + try { + process.stderr.write("__EVT__" + JSON.stringify(obj) + "\n"); + } catch (_) {} + } + globalThis.__native = { + emitProgress(status) { + emit({ type: "progress", status: String(status) }); + }, + log(level, msg, fields) { + emit({ + type: "log", + level: String(level || "info"), + msg: String(msg), + fields: fields == null ? null : fields, + }); + }, + }; +})(); + +(async () => { + // 1) read params JSON from stdin (one-shot) + let raw = ""; + process.stdin.setEncoding("utf8"); + for await (const chunk of process.stdin) raw += chunk; + const params = raw ? JSON.parse(raw) : null; + + // 2) inline user script begins ↓↓↓ +"#; + +const RUNNER_FOOTER: &str = r#" + // ↑↑↑ inline user script ends + + // 3) dispatch to __nodejsMain(params) + if (typeof globalThis.__nodejsMain !== "function") { + process.stderr.write( + "__EVT__" + JSON.stringify({ + type: "log", + level: "error", + msg: "脚本未注册 globalThis.__nodejsMain", + }) + "\n" + ); + process.exit(1); + } + try { + const result = await globalThis.__nodejsMain(params); + process.stdout.write( + "__RESULT__" + JSON.stringify(result === undefined ? null : result) + "\n" + ); + process.exit(0); + } catch (e) { + const msg = (e && (e.stack || e.message)) || String(e); + process.stderr.write( + "__EVT__" + JSON.stringify({ type: "log", level: "error", msg }) + "\n" + ); + process.exit(2); + } +})(); +"#; + +fn build_runner(user_source: &str) -> String { + let mut s = String::with_capacity( + RUNNER_HEADER.len() + user_source.len() + RUNNER_FOOTER.len(), + ); + s.push_str(RUNNER_HEADER); + s.push_str(user_source); + s.push_str(RUNNER_FOOTER); + s +} + +// --------------------------------------------------------------------------- +// 从后端拉脚本源码 +// --------------------------------------------------------------------------- +fn api_base() -> String { + std::env::var("AICLIENT_API_BASE").unwrap_or_else(|_| "http://127.0.0.1:8001".to_string()) +} + +async fn fetch_node_script_source(script_name: &str) -> Result { + let base = api_base().trim_end_matches('/').to_string(); + let q = urlencoding::encode(script_name); + let url = format!("{base}/api/v1/nodejs-scripts?name={q}"); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(120)) + .build() + .map_err(|e| NodeError::FetchScript(e.to_string()))?; + let res = client + .get(&url) + .send() + .await + .map_err(|e| NodeError::FetchScript(e.to_string()))?; + let body = res + .text() + .await + .map_err(|e| NodeError::FetchScript(e.to_string()))?; + let v: Value = serde_json::from_str(&body) + .map_err(|e| NodeError::FetchScript(format!("响应非 JSON: {e}")))?; + if !v.get("ok").and_then(|x| x.as_bool()).unwrap_or(false) { + let msg = v + .get("message") + .and_then(|x| x.as_str()) + .unwrap_or("未知错误"); + return Err(NodeError::FetchScript(msg.to_string())); + } + let source = v + .get("data") + .and_then(|d| d.get("source")) + .and_then(|s| s.as_str()) + .ok_or_else(|| NodeError::FetchScript("响应缺少 data.source".into()))?; + Ok(source.to_string()) +} + +/// 后端 `scripts/nodejs` 目录下的 `.js` 文件名列表。 +pub async fn list_scripts() -> Result, NodeError> { + let base = api_base().trim_end_matches('/').to_string(); + let url = format!("{base}/api/v1/nodejs-scripts/list"); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .map_err(|e| NodeError::FetchScript(e.to_string()))?; + let res = client + .get(&url) + .send() + .await + .map_err(|e| NodeError::FetchScript(e.to_string()))?; + let body = res + .text() + .await + .map_err(|e| NodeError::FetchScript(e.to_string()))?; + let v: Value = serde_json::from_str(&body) + .map_err(|e| NodeError::FetchScript(format!("响应非 JSON: {e}")))?; + if !v.get("ok").and_then(|x| x.as_bool()).unwrap_or(false) { + let msg = v + .get("message") + .and_then(|x| x.as_str()) + .unwrap_or("未知错误"); + return Err(NodeError::FetchScript(msg.to_string())); + } + let names = v + .get("data") + .and_then(|d| d.get("names")) + .and_then(|n| n.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|x| x.as_str().map(|s| s.to_string())) + .collect::>() + }) + .unwrap_or_default(); + Ok(names) +} + +// --------------------------------------------------------------------------- +// 公开入口 +// --------------------------------------------------------------------------- + +/// 按脚本名运行：去后端拉源码 → 包装 runner → 启动 node → 收集结果。 +pub async fn run_node_script( + script_name: String, + params: Value, + events: UnboundedSender, +) -> Result { + let user_source = fetch_node_script_source(&script_name).await?; + run_inline_source(user_source, params, events).await +} + +/// 调试入口：直接传源码字符串，跳过后端拉取。 +pub async fn run_node_script_source( + source: String, + params: Value, + events: UnboundedSender, +) -> Result { + run_inline_source(source, params, events).await +} + +async fn run_inline_source( + user_source: String, + params: Value, + events: UnboundedSender, +) -> Result { + let node = locate_node().ok_or(NodeError::NotFound)?; + let cwd = nodejs_dir().unwrap_or_else(std::env::temp_dir); + + // 写到 OS 临时目录而非 resources/，避免污染只读资源 + let tmp_dir = std::env::temp_dir().join("aiclient-node"); + tokio::fs::create_dir_all(&tmp_dir).await?; + let script_path = tmp_dir.join(format!("script-{}.cjs", Uuid::new_v4())); + tokio::fs::write(&script_path, build_runner(&user_source)).await?; + + let mut child = Command::new(&node) + .arg(&script_path) + .current_dir(&cwd) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()?; + + // 把 params JSON 灌进 stdin + if let Some(mut stdin) = child.stdin.take() { + let json = serde_json::to_vec(¶ms)?; + stdin.write_all(&json).await?; + stdin.shutdown().await?; + } + + // stderr → 解析 __EVT__ → PipelineEvent + let stderr = child.stderr.take().expect("stderr piped"); + let events_tx = events.clone(); + let stderr_task = tokio::spawn(async move { + let mut reader = BufReader::new(stderr).lines(); + let mut tail = String::new(); + while let Ok(Some(line)) = reader.next_line().await { + if let Some(rest) = line.strip_prefix("__EVT__") { + if let Ok(v) = serde_json::from_str::(rest) { + let typ = v.get("type").and_then(|x| x.as_str()).unwrap_or(""); + match typ { + "progress" => { + let s = v + .get("status") + .and_then(|x| x.as_str()) + .unwrap_or("") + .to_string(); + let _ = events_tx.send(PipelineEvent::Progress(s)); + } + "log" => { + let level = v + .get("level") + .and_then(|x| x.as_str()) + .unwrap_or("info") + .to_string(); + let msg = v + .get("msg") + .and_then(|x| x.as_str()) + .unwrap_or("") + .to_string(); + let fields = v.get("fields").cloned().filter(|x| !x.is_null()); + let _ = events_tx.send(PipelineEvent::Log { level, msg, fields }); + } + _ => { + // 兜底：未知 __EVT__ 类型也当 log 抛出 + let _ = events_tx.send(PipelineEvent::Log { + level: "info".into(), + msg: rest.to_string(), + fields: None, + }); + } + } + } + } else { + // 纯文本 stderr（比如 node 自己打印的 warning） → 当 info log + let _ = events_tx.send(PipelineEvent::Log { + level: "warn".into(), + msg: line.clone(), + fields: None, + }); + tail.push_str(&line); + tail.push('\n'); + } + } + tail + }); + + // stdout → 找 __RESULT__ 那一行 + let stdout = child.stdout.take().expect("stdout piped"); + let events_tx2 = events.clone(); + let stdout_task = tokio::spawn(async move { + let mut reader = BufReader::new(stdout).lines(); + let mut result: Option = None; + while let Ok(Some(line)) = reader.next_line().await { + if let Some(rest) = line.strip_prefix("__RESULT__") { + result = Some(rest.to_string()); + } else if !line.is_empty() { + // 用户 console.log 的内容也透传给前端 + let _ = events_tx2.send(PipelineEvent::Log { + level: "info".into(), + msg: line, + fields: None, + }); + } + } + result + }); + + let status = child.wait().await?; + let stderr_tail = stderr_task.await.unwrap_or_default(); + let result_line = stdout_task.await.unwrap_or(None); + + let _ = tokio::fs::remove_file(&script_path).await; + + if !status.success() { + return Err(NodeError::NonZero { + status: status.code(), + stderr: stderr_tail, + }); + } + + let result_str = result_line.ok_or(NodeError::NoResult)?; + let v: Value = serde_json::from_str(&result_str)?; + Ok(v) +} diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json index 38e12d7..9546477 100644 --- a/src-tauri/tauri.conf.json +++ b/src-tauri/tauri.conf.json @@ -27,6 +27,7 @@ }, "bundle": { "resources": [ + "resources/**/*", "binaries/*" ], "active": true,