diff --git a/Cargo.lock b/Cargo.lock index fa5467f..efade96 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -15,13 +15,18 @@ dependencies = [ name = "ai" version = "0.1.0" dependencies = [ + "anyhow", "chrono", "common", + "regex", "reqwest", "serde", "serde_json", "serde_yaml", + "thiserror", "tokio", + "tracing", + "tracing-subscriber", ] [[package]] @@ -322,8 +327,11 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" name = "common" version = "0.1.0" dependencies = [ + "anyhow", + "reqwest", "serde", "serde_json", + "serde_yaml", "thiserror", ] @@ -1374,6 +1382,18 @@ dependencies = [ "bitflags", ] +[[package]] +name = "regex" +version = "1.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + [[package]] name = "regex-automata" version = "0.4.14" @@ -1602,6 +1622,9 @@ dependencies = [ "sqlx", "thiserror", "tokio", + "tracing", + "tracing-subscriber", + "uuid", "wechat", ] diff --git a/ai/Cargo.toml b/ai/Cargo.toml index ea945b2..d73c3b5 100644 --- a/ai/Cargo.toml +++ b/ai/Cargo.toml @@ -15,3 +15,11 @@ serde_json = "1.0" # JSON 处理 serde_yaml = "0.9" # YAML 解析（工具规范文件） # ── 时间 ── chrono = "0.4" +# ── 错误处理 ── +anyhow = "1" +thiserror = "2" +# ── 日志管理 ── +tracing = "0.1" +tracing-subscriber = "0.3" +# ── 正则 ── +regex = "1" diff --git a/ai/src/core.rs b/ai/src/core.rs index e91d736..5332e9c 100644 --- a/ai/src/core.rs +++ b/ai/src/core.rs @@ -1,4 +1,4 @@ -use crate::toolkit::{call_tools, get_tool}; +use crate::toolkit::{call_tools, get_tool, tool_notice_text}; use common::model::ai::{ ChatCompletionRequestBody, ChatCompletionRequestExtra, ChatCompletionRequestMessage, ChatCompletionRequestThinking, ChatCompletionResponseBody, @@ -9,11 +9,16 @@ pub async fn send_message( sender: tokio::sync::mpsc::Sender, messages: Vec, channel: ChannelMessage, -) -> Result<(), reqwest::Error> { +) -> anyhow::Result<()> { let result = request_llm(messages).await; match result { Ok(result) => { - for choice in result.unwrap().choices { + let Some(result) = result else { + eprintln!("LLM 请求没有返回有效响应体"); + return Ok(()); + }; + + for choice in result.choices { let messsage_content = choice.message.content; let messsage_reasoning = choice.message.reasoning_content.unwrap_or_default(); let tool_calls = choice.message.tool_calls.clone(); @@ -32,10 +37,29 @@ pub async fn send_message( match choice.message.tool_calls { Some(tool_calls) => { - println!("调用了{}个工具", tool_calls.len()); - let queue_message = call_tools(channel.clone(), tool_calls).await.unwrap(); - if let Some(queue_message) = queue_message { - sender.send(queue_message).await.unwrap(); + if tool_calls.is_empty() { + eprintln!("模型返回 tool_calls 字段，但列表为空"); + continue; + } + + println!("模型请求调用{}个工具", tool_calls.len()); + if let Some(text) = tool_notice_text(&tool_calls) { + sender + .send(QueueMessage::Notice(channel.clone(), text)) + .await + .unwrap(); + } + + match call_tools(channel.clone(), tool_calls).await { + Ok(Some(queue_message)) => { + sender.send(queue_message).await.unwrap(); + } + Ok(None) => { + eprintln!("工具调用没有生成后续队列消息"); + } + Err(error) => { + eprintln!("工具调用失败: {error}"); + } } } None => { @@ -54,11 +78,13 @@ pub async fn send_message( pub async fn request_llm( messages: Vec, -) -> Result, reqwest::Error> { +) -> anyhow::Result> { let client = reqwest::Client::new(); + let api_key = + std::env::var("ROUTER_API_KEY").map_err(|_| anyhow::anyhow!("必须设置ROUTER_API_KEY"))?; let body = ChatCompletionRequestBody { model: "deepseek-v4-flash".to_string(), - messages: messages, + messages, stream: false, tools: get_tool(), extra_body: Some(ChatCompletionRequestExtra { @@ -69,7 +95,7 @@ pub async fn request_llm( }; let resp = client .post("http://100.64.52.162:9807/v1/chat/completions") - .bearer_auth(std::env::var("ROUTER_API_KEY").unwrap()) + .bearer_auth(api_key) .json(&body) .send() .await; @@ -80,8 +106,13 @@ pub async fn request_llm( let text = resp.text().await.unwrap_or_default(); // println!("response:\n{}", text); if status.is_success() { - let result = serde_json::from_str::(&text).unwrap(); - return Ok(Some(result)); + match serde_json::from_str::(&text) { + Ok(result) => return Ok(Some(result)), + Err(error) => { + eprintln!("LLM 响应解析失败: {error}"); + eprintln!("body: {text}"); + } + } } else { eprintln!("HTTP 请求失败"); eprintln!("status: {}", status); diff --git a/ai/src/lib.rs b/ai/src/lib.rs index 147a36b..37e8e7b 100644 --- a/ai/src/lib.rs +++ b/ai/src/lib.rs @@ -1,2 +1,4 @@ pub mod core; -pub mod toolkit; \ No newline at end of file +pub mod toolkit; +pub mod tools; +pub mod workflow; diff --git a/ai/src/toolkit.rs b/ai/src/toolkit.rs index 70d73ac..d800587 100644 --- a/ai/src/toolkit.rs +++ b/ai/src/toolkit.rs @@ -1,34 +1,113 @@ -use chrono::Local; -use serde_json::error; +use crate::{ + tools::date_time::get_system_date_time, + tools::web::create_web_page, + workflow::core::{call_flow, load_tools_define}, +}; use common::{ - model::ai::ChatCompletionToolCall, + model::ai::{ChatCompletionFunction, ChatCompletionToolCall}, queue::{ChannelMessage, QueueMessage}, }; +use serde_json::Value; pub async fn call_tools( channel: ChannelMessage, tools: Vec, -) -> Result, error::Error> { - if let Some(tool) = tools.first() { - if let Some(funcation) = tool.function.as_ref() { - if funcation.name == "query_date_time" { - let date_time = Local::now(); - let formatted = date_time.format("%Y-%m-%d %H:%M:%S").to_string(); - return Ok(Some(QueueMessage::Tool( - channel, - tool.id.clone(), - funcation.name.clone(), - format!("现在时间:{}", formatted), - ))); - } - } +) -> anyhow::Result> { + let tool_path = tool_path(); + + if tools.is_empty() { + eprintln!("模型返回了空工具调用列表"); + return Ok(None); } - Ok(None) + + if tools.len() > 1 { + eprintln!("模型一次返回了{}个工具调用，当前仅处理第一个", tools.len()); + } + + let Some(tool) = tools.first() else { + return Ok(None); + }; + + let (tool_name, tool_result) = match tool.function.as_ref() { + Some(function) => { + println!( + "开始调用工具: {}, arguments={}", + function.name, function.arguments + ); + ( + function.name.clone(), + execute_tool(&tool_path, function).await, + ) + } + None => { + eprintln!("工具调用缺少 function 字段: id={}", tool.id); + ( + "unknown".to_string(), + "工具调用失败: tool call 缺少 function 字段".to_string(), + ) + } + }; + + let tool_result = if tool_result.trim().is_empty() { + format!("工具调用失败: {tool_name} 返回空结果") + } else { + tool_result + }; + println!( + "工具调用完成: {}, result_len={}", + tool_name, + tool_result.len() + ); + + Ok(Some(QueueMessage::Tool( + channel, + tool.id.clone(), + tool_name, + tool_result, + ))) +} + +async fn execute_tool(tool_path: &str, function: &ChatCompletionFunction) -> String { + match function.name.as_str() { + "query_date_time" => match get_system_date_time() { + Ok(value) => value, + Err(error) => format!("工具调用失败: {error}"), + }, + "query_capabilities" => match load_tools_define(tool_path.to_string()).await { + Ok(value) => value, + Err(error) => format!("工具调用失败: 加载工具定义失败: {error}"), + }, + "create_web_page" => create_web_page(&function.arguments).await, + "call_capability" => call_flow(tool_path.to_string(), function.arguments.clone()).await, + other => call_named_capability(tool_path, other, &function.arguments).await, + } +} + +async fn call_named_capability(tool_path: &str, name: &str, arguments: &str) -> String { + let prompt: Value = match serde_json::from_str(arguments) { + Ok(value) => value, + Err(error) => return format!("工具调用失败: 解析 {name} 参数失败: {error}"), + }; + let args = serde_json::json!({ + "name": name, + "prompt": prompt, + }); + call_flow(tool_path.to_string(), args.to_string()).await +} + +pub fn tool_notice_text(tools: &[ChatCompletionToolCall]) -> Option { + let function = tools.first()?.function.as_ref()?; + let args: Value = serde_json::from_str(&function.arguments).ok()?; + args.get("text") + .and_then(Value::as_str) + .map(str::trim) + .filter(|text| !text.is_empty()) + .map(ToString::to_string) } pub fn get_tool() -> Vec { - return vec![ + vec![ serde_json::json!({ "type": "function", "function": { @@ -37,31 +116,87 @@ pub fn get_tool() -> Vec { "parameters": {"type": "object", "properties": {}, "required": []} } }), - // serde_json::json!({ - // "type": "function", - // "function": { - // "name": "query_capabilities", - // "description": "查询所有可用工具的列表、参数格式和用法说明", - // "parameters": {"type": "object", "properties": {}, "required": []} - // } - // }), - // serde_json::json!({ - // "type": "function", - // "function": { - // "name": "call_capability", - // "description": "调用指定的工具。先通过 query_capabilities 获取工具名和参数格式", - // "parameters": { - // "type": "object", - // "properties": { - // "name": {"type": "string", "description": "工具名称"}, - // "prompt": { - // "type": "object", - // "description": "工具参数对象，如 {\"location\":\"北京\"}" - // } - // }, - // "required": ["name", "prompt"] - // } - // } - // }), - ]; + serde_json::json!({ + "type": "function", + "function": { + "name": "query_capabilities", + "description": "查询所有可用工具的列表、参数格式和用法说明", + "parameters": {"type": "object", "properties": {}, "required": []} + } + }), + serde_json::json!({ + "type": "function", + "function": { + "name": "create_web_page", + "description": "创建一个可通过链接访问的动态内容页面，用于把较长或结构复杂的内容交给用户查看。邮件提醒请使用 page_type=mail，工具结果会返回 /mail/{slug} 链接。", + "parameters": { + "type": "object", + "properties": { + "page_type": { + "type": "string", + "description": "页面类型，只能包含字母、数字、下划线或连字符。普通内容可用 page，邮件内容用 mail。" + }, + "title": { + "type": "string", + "description": "页面标题，例如邮件主题或内容摘要标题。" + }, + "summary": { + "type": "string", + "description": "简短摘要，会显示在页面标题下方，也可用于最终回复用户。" + }, + "content": { + "type": "string", + "description": "页面正文。可以放较长内容，系统会按纯文本安全渲染并保留换行。" + }, + "source_label": { + "type": "string", + "description": "内容来源，例如发件人、系统名或服务名。" + }, + "metadata": { + "type": "object", + "description": "附加结构化信息，例如 {\"from\":\"xxx\",\"subject\":\"xxx\",\"received_at\":\"2026-07-06 09:00\"}。" + }, + "text": { + "type": "string", + "description": "可选。调用工具前发送给用户的提醒消息，用于说明接下来会创建详情页面；为空则不发送。" + } + }, + "required": ["title", "content"] + } + } + }), + serde_json::json!({ + "type": "function", + "function": { + "name": "call_capability", + "description": "调用指定的工具。先通过 query_capabilities 获取工具名和参数格式", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "工具名称"}, + "prompt": { + "type": "object", + "description": "工具参数对象，如 {\"location\":\"北京\"}" + }, + "text": { + "type": "string", + "description": "可选。调用工具前发送给用户的提醒消息，用于说明接下来会做什么；为空则不发送" + } + }, + "required": ["name", "prompt"] + } + } + }), + ] +} + +fn tool_path() -> String { + std::env::var("IA_TOOLS_PATH").unwrap_or_else(|_| { + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .join("tools") + .display() + .to_string() + }) } diff --git a/ai/src/tools/date_time.rs b/ai/src/tools/date_time.rs new file mode 100644 index 0000000..149efb3 --- /dev/null +++ b/ai/src/tools/date_time.rs @@ -0,0 +1,7 @@ +use chrono::Local; + +pub fn get_system_date_time() -> anyhow::Result { + let date_time = Local::now(); + let formatted = date_time.format("%Y-%m-%d %H:%M:%S").to_string(); + Ok(format!("现在时间:{}", formatted)) +} diff --git a/ai/src/tools/http.rs b/ai/src/tools/http.rs new file mode 100644 index 0000000..9d65af0 --- /dev/null +++ b/ai/src/tools/http.rs @@ -0,0 +1,381 @@ +use std::collections::HashMap; +use std::process::Stdio; + +use anyhow::{Context, Result, anyhow}; +use serde::Serialize; +use tokio::io::AsyncWriteExt; +use tokio::process::Command; + +/// 整体超时默认值（秒） +const DEFAULT_TIMEOUT: u64 = 30; + +// ────────────────────────────── 请求基座 ────────────────────────────── + +/// HTTP 请求基座：封装 curl + jq 的执行细节，供各类 HTTP 工具复用 +/// +/// 所有参数以 `Vec` 直接传给 `curl` 子进程，**不经过 shell**， +/// 因此 URL / Header / Body 中包含空格、引号、`$`、`;` 等特殊字符也不会被解释， +/// 天然规避命令注入。 +/// +/// 采用 builder 风格链式配置，后续具体工具基于此基座实现，例如： +/// ```ignore +/// // 一个「查天气」工具，复用基座 +/// pub async fn get_weather(city: &str, token: &str) -> Result { +/// let resp = HttpRequest::new(format!("https://api.example.com/weather?q={city}")) +/// .header("Authorization", format!("Bearer {token}")) +/// .jq(".current.temp_c") +/// .execute() +/// .await?; +/// Ok(resp.json()?.as_f64().unwrap_or(0.0)) +/// } +/// ``` +pub struct HttpRequest { + /// 请求地址 + pub url: String, + /// HTTP 方法 + pub method: String, + /// 请求头 + pub headers: HashMap, + /// 请求体 + pub body: Option, + /// jq 过滤表达式（为空则不过滤） + pub jq: Option, + /// 整体超时（秒） + pub timeout: u64, +} + +impl HttpRequest { + /// 以指定 URL 创建一个 GET 请求 + pub fn new(url: impl Into) -> Self { + Self { + url: url.into(), + method: "GET".to_string(), + headers: HashMap::new(), + body: None, + jq: None, + timeout: DEFAULT_TIMEOUT, + } + } + + /// 设置 HTTP 方法（自动转大写，便于容忍小写输入） + pub fn method(mut self, method: impl Into) -> Self { + self.method = method.into().to_uppercase(); + self + } + + /// 追加一个请求头（同名键会覆盖） + pub fn header(mut self, key: impl Into, value: impl Into) -> Self { + self.headers.insert(key.into(), value.into()); + self + } + + /// 设置原始请求体（不自动设置 Content-Type） + pub fn body(mut self, body: impl Into) -> Self { + self.body = Some(body.into()); + self + } + + /// 设置 JSON 请求体，并自动加上 `Content-Type: application/json` + pub fn json_body(mut self, value: &serde_json::Value) -> Self { + self.body = Some(serde_json::to_string(value).unwrap_or_default()); + self.headers + .insert("Content-Type".into(), "application/json".into()); + self + } + + /// 设置 jq 过滤表达式，对 JSON 响应体做提取/加工 + pub fn jq(mut self, filter: impl Into) -> Self { + self.jq = Some(filter.into()); + self + } + + /// 设置整体超时（秒） + pub fn timeout(mut self, secs: u64) -> Self { + self.timeout = secs; + self + } + + /// 执行请求：先跑 curl，再用 jq（可选）过滤响应体 + pub async fn execute(&self) -> Result { + let (status, body) = self.run_curl().await?; + + // 可选：用 jq 过滤响应体。jq 失败不致命——保留原文并附带错误 + let (body, jq_error) = match self.jq.as_deref() { + Some(filter) if !filter.trim().is_empty() => run_jq(filter, &body).await?, + _ => (body, None), + }; + + Ok(HttpResponse { + status, + body, + jq_error, + }) + } + + /// 调用 curl 并解析出「状态码 + 响应体」 + /// + /// 利用 `curl -w '\n%{http_code}'` 在响应体末尾追加一行状态码， + /// 随后以最后一个换行为界拆分：之前是 body，之后是 status。 + /// + /// 注意：HTTP 4xx/5xx 不是错误，会原样返回状态码与响应体； + /// 只有 curl 自身失败（连接超时、DNS 解析失败等）才返回 `Err`。 + async fn run_curl(&self) -> Result<(u16, String)> { + let method = self.method.to_uppercase(); + + // 组装 curl 参数 + let mut args: Vec = vec![ + "-s".into(), // 静默进度条 + "-S".into(), // 出错时仍打印错误到 stderr + "-L".into(), // 跟随重定向 + "--compressed".into(), // 自动解压 gzip/br/zstd + "--connect-timeout".into(), + "10".into(), + "--max-time".into(), + self.timeout.to_string(), + "-X".into(), + method, + // 末尾追加一行状态码，用换行与响应体分隔 + "-w".into(), + "\n%{http_code}".into(), + ]; + for (k, v) in &self.headers { + args.push("-H".into()); + args.push(format!("{k}: {v}")); + } + if let Some(body) = &self.body { + // --data-raw 不会把开头 @ 当作文件读取，比 -d 更安全 + args.push("--data-raw".into()); + args.push(body.clone()); + } + args.push(self.url.clone()); + + // 执行 curl + let output = Command::new("curl") + .args(&args) + .output() + .await + .with_context(|| format!("执行 curl 失败: {}", self.url))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + return Err(anyhow!("curl 请求失败: {stderr}")); + } + + // 拆分响应体与状态码（最后一行即状态码） + let stdout = String::from_utf8_lossy(&output.stdout).into_owned(); + let (body, status) = match stdout.rfind('\n') { + Some(i) => { + let status: u16 = stdout[i + 1..] + .trim() + .parse() + .context("解析 HTTP 状态码失败")?; + (stdout[..i].to_string(), status) + } + None => (String::new(), 0u16), + }; + + Ok((status, body)) + } +} + +// ────────────────────────────── 响应 ────────────────────────────── + +/// HTTP 请求结果 +#[derive(Debug, Serialize)] +pub struct HttpResponse { + /// HTTP 状态码；传输层失败时为 0 + pub status: u16, + /// 响应体（若指定了 jq 则为过滤后的内容） + pub body: String, + /// jq 执行失败时的错误信息（非致命，原响应体仍会保留在 body 中） + #[serde(skip_serializing_if = "Option::is_none")] + pub jq_error: Option, +} + +impl HttpResponse { + /// 是否为 2xx 成功响应 + pub fn is_success(&self) -> bool { + (200..300).contains(&self.status) + } + + /// 将响应体解析为 JSON（若指定了 jq，则解析过滤后的结果） + pub fn json(&self) -> Result { + serde_json::from_str(&self.body).context("响应体不是合法 JSON") + } +} + +// ────────────────────────────── jq 过滤 ────────────────────────────── + +/// 用 jq 对 `body` 做过滤 +/// +/// 成功返回 `(过滤结果, None)`；失败（非 JSON 或表达式错误）则保留原文， +/// 返回 `(原文, Some(错误信息))`，避免因 jq 问题丢失原始响应。 +async fn run_jq(filter: &str, body: &str) -> Result<(String, Option)> { + let mut child = Command::new("jq") + .arg(filter) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .context("启动 jq 失败")?; + + if let Some(mut stdin) = child.stdin.take() { + // 忽略写入错误：响应体为空时 jq 会自行报错 + let _ = stdin.write_all(body.as_bytes()).await; + } + + let output = child.wait_with_output().await.context("等待 jq 结束失败")?; + + if output.status.success() { + Ok((String::from_utf8_lossy(&output.stdout).into_owned(), None)) + } else { + let err = String::from_utf8_lossy(&output.stderr).trim().to_string(); + Ok((body.to_string(), Some(err))) + } +} + +// ────────────────────────────── 测试 ────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{Read, Write}; + use std::net::{TcpListener, TcpStream}; + + /// 读取一条完整的 HTTP 请求（含 body） + fn read_request(stream: &mut TcpStream) -> String { + let mut buf = Vec::new(); + let mut tmp = [0u8; 4096]; + loop { + let n = stream.read(&mut tmp).unwrap_or(0); + if n == 0 { + break; + } + buf.extend_from_slice(&tmp[..n]); + let s = String::from_utf8_lossy(&buf); + if let Some(split) = s.find("\r\n\r\n") { + let headers = &s[..split]; + let body_start = split + 4; + let clen = headers + .lines() + .find_map(|l| { + let l = l.to_lowercase(); + l.strip_prefix("content-length: ") + .and_then(|v| v.parse::().ok()) + }) + .unwrap_or(0); + if buf.len() >= body_start + clen { + break; + } + } + if buf.len() > 65536 { + break; + } + } + String::from_utf8_lossy(&buf).into_owned() + } + + /// 启动一个一次性本地 HTTP 服务：把完整原始请求传给 `respond`，回写其返回值 + fn spawn_server(respond: F) -> String + where + F: FnOnce(&str) -> (u16, String) + Send + 'static, + { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + if let Ok((mut stream, _)) = listener.accept() { + let req = read_request(&mut stream); + let (status, resp_body) = respond(&req); + let reason = match status { + 200 => "OK", + 404 => "Not Found", + _ => "OK", + }; + let resp = format!( + "HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}", + resp_body.len() + ); + let _ = stream.write_all(resp.as_bytes()); + let _ = stream.flush(); + } + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn builder_get_with_jq() { + let url = spawn_server(|_| (200, r#"{"origin":"1.2.3.4"}"#.to_string())); + let resp = HttpRequest::new(url).jq(".origin").execute().await.unwrap(); + assert!(resp.is_success()); + assert!(resp.jq_error.is_none(), "jq 不应报错: {:?}", resp.jq_error); + assert_eq!(resp.body.trim(), "\"1.2.3.4\""); + } + + #[tokio::test] + async fn builder_post_body_roundtrip() { + // 服务把收到的请求体原样回显到 .echo + let url = spawn_server(|req| { + let body = req.split("\r\n\r\n").nth(1).unwrap_or(""); + (200, format!("{{\"echo\":{}}}", body)) + }); + let resp = HttpRequest::new(url) + .method("post") // 小写也应被接受 + .header("Content-Type", "application/json") + .body(r#"{"hello":"world"}"#) + .jq(".echo.hello") + .execute() + .await + .unwrap(); + assert_eq!(resp.body.trim(), "\"world\""); + } + + #[tokio::test] + async fn json_body_sets_content_type() { + // 服务回显是否带 application/json 头，以及请求体内容 + let url = spawn_server(|req| { + let has_ct = req + .lines() + .any(|l| l.eq_ignore_ascii_case("content-type: application/json")); + let body = req.split("\r\n\r\n").nth(1).unwrap_or(""); + (200, format!("{{\"ct\":{},\"echo\":{}}}", has_ct, body)) + }); + let resp = HttpRequest::new(url) + .method("POST") + .json_body(&serde_json::json!({"hello": "world"})) + .jq(".ct") + .execute() + .await + .unwrap(); + assert_eq!(resp.body.trim(), "true"); + } + + #[tokio::test] + async fn bad_jq_keeps_body() { + let url = spawn_server(|_| (200, r#"{"data":"hello"}"#.to_string())); + let resp = HttpRequest::new(url).jq(".[").execute().await.unwrap(); // 非法表达式 + assert!(resp.jq_error.is_some(), "应附带 jq 错误"); + assert!( + resp.body.contains("hello"), + "应保留原始响应体: {}", + resp.body + ); + } + + #[tokio::test] + async fn http_404_is_result_not_error() { + let url = spawn_server(|_| (404, r#"{"err":"nope"}"#.to_string())); + let resp = HttpRequest::new(url).execute().await.unwrap(); + assert_eq!(resp.status, 404); + assert!(!resp.is_success()); + assert!(resp.body.contains("nope")); + } + + #[tokio::test] + async fn response_json_parses() { + let url = spawn_server(|_| (200, r#"{"a":1,"b":[2,3]}"#.to_string())); + let resp = HttpRequest::new(url).execute().await.unwrap(); + let v = resp.json().unwrap(); + assert_eq!(v["a"], 1); + assert_eq!(v["b"][1], 3); + } +} diff --git a/ai/src/tools/mod.rs b/ai/src/tools/mod.rs new file mode 100644 index 0000000..cc320f9 --- /dev/null +++ b/ai/src/tools/mod.rs @@ -0,0 +1,3 @@ +pub mod date_time; +pub mod http; +pub mod web; diff --git a/ai/src/tools/web.rs b/ai/src/tools/web.rs new file mode 100644 index 0000000..fd7a649 --- /dev/null +++ b/ai/src/tools/web.rs @@ -0,0 +1,95 @@ +use anyhow::{Context, anyhow}; +use serde::Deserialize; +use serde_json::{Value, json}; + +#[derive(Debug, Deserialize)] +struct CreateWebPageArgs { + page_type: Option, + title: String, + summary: Option, + content: String, + source_label: Option, + metadata: Option, + #[allow(dead_code)] + text: Option, +} + +#[derive(Debug, Deserialize)] +struct CreateWebPageResponse { + success: bool, + title: String, + page_type: String, + summary: Option, + path: String, + url: String, +} + +pub async fn create_web_page(arguments: &str) -> String { + match create_web_page_inner(arguments).await { + Ok(result) => result, + Err(error) => format!("工具调用失败: 创建 web 页面失败: {error}"), + } +} + +async fn create_web_page_inner(arguments: &str) -> anyhow::Result { + let args: CreateWebPageArgs = + serde_json::from_str(arguments).context("解析 create_web_page 参数失败")?; + if args.title.trim().is_empty() { + return Err(anyhow!("title 不能为空")); + } + if args.content.trim().is_empty() { + return Err(anyhow!("content 不能为空")); + } + + let payload = json!({ + "page_type": args.page_type, + "title": args.title, + "summary": args.summary, + "content": args.content, + "source_label": args.source_label, + "metadata": args.metadata, + }); + let base_url = web_base_url(); + let endpoint = format!("{base_url}/web/pages"); + let client = reqwest::Client::new(); + let response = client + .post(&endpoint) + .json(&payload) + .send() + .await + .with_context(|| format!("请求 web 页面接口失败: {endpoint}"))?; + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + + if !status.is_success() { + return Err(anyhow!("接口返回失败: status={status}, body={body}")); + } + + let result: CreateWebPageResponse = + serde_json::from_str(&body).context("解析 web 页面接口响应失败")?; + if !result.success { + return Err(anyhow!("接口未成功创建页面: {body}")); + } + + let summary = result + .summary + .as_deref() + .filter(|summary| !summary.trim().is_empty()) + .map(|summary| format!("\n摘要：{}", summary.trim())) + .unwrap_or_default(); + + Ok(format!( + "web 页面已创建：{}\n路径：{}\n类型：{}\n标题：{}{}", + result.url, result.path, result.page_type, result.title, summary + )) +} + +fn web_base_url() -> String { + std::env::var("IA_WEB_BASE_URL") + .or_else(|_| std::env::var("WEB_BASE_URL")) + .or_else(|_| std::env::var("HOST").map(|host| format!("http://{host}"))) + .unwrap_or_else(|_| "http://127.0.0.1:9003".to_string()) + .trim() + .trim_end_matches('/') + .replace("0.0.0.0", "127.0.0.1") +} diff --git a/ai/src/workflow/core.rs b/ai/src/workflow/core.rs new file mode 100644 index 0000000..acc9b76 --- /dev/null +++ b/ai/src/workflow/core.rs @@ -0,0 +1,73 @@ +use serde_json::Value; +use tokio::fs; + +use crate::workflow::flow::start_flow; +use crate::workflow::types::FlowConfig; + +pub async fn call_flow(path: String, args: String) -> String { + let all_flow = match load_configs(path).await { + Ok(flows) => flows, + Err(err) => return format!("工具调用失败:{:?}", err), + }; + let args_json: Value = match serde_json::from_str(&args) { + Ok(json) => json, + Err(err) => return format!("工具调用失败:{:?}", err), + }; + let flow_name = args_json["name"].as_str().unwrap_or_default(); + if let Some(config) = all_flow.iter().find(|e| flow_name == e.name) { + let flow_result = start_flow(config, args_json["prompt"].clone()).await; + match flow_result { + Ok(value) => value, + Err(err) => format!("工具调用失败:{:?}", err), + } + } else { + "工具不存在，请检查工具名称".to_string() + } +} + +pub async fn load_tools_define(path: String) -> anyhow::Result { + let configs = load_configs(path).await?; + Ok(serde_json::to_string(&configs).unwrap_or_default()) +} + +pub async fn load_configs(path: String) -> anyhow::Result> { + let mut configs: Vec = vec![]; + let config_files = scan_config_dir(path).await?; + if !config_files.is_empty() { + for file in config_files { + let config = load_yaml(file).await?; + configs.push(config); + } + } + Ok(configs) +} + +pub async fn load_yaml(path: String) -> anyhow::Result { + let content: String = fs::read_to_string(path).await?; + let config = serde_yaml::from_str(&content)?; + Ok(config) +} + +pub async fn scan_config_dir(path: String) -> anyhow::Result> { + let mut dir = fs::read_dir(path).await?; + let mut files: Vec = vec![]; + + while let Some(entry) = dir.next_entry().await? { + let path = entry.path(); + let abs = fs::canonicalize(&path).await?; + files.push(abs.display().to_string()); + } + + Ok(files) +} + +#[tokio::test] +async fn list_file() { + println!(); + + let tools = load_tools_define("../tools".to_string()).await.unwrap(); + + println!("{}", tools); + + println!(); +} diff --git a/ai/src/workflow/flow.rs b/ai/src/workflow/flow.rs new file mode 100644 index 0000000..ab7c070 --- /dev/null +++ b/ai/src/workflow/flow.rs @@ -0,0 +1,510 @@ +use std::str::FromStr; +use std::{ + collections::HashMap, + time::{Duration, Instant}, +}; + +use reqwest::{ + Client, Method, + header::{HeaderMap, HeaderName, HeaderValue}, +}; +use serde_json::Value; +use thiserror::Error; +use tracing::{debug, error, info, warn}; + +use crate::workflow::resolve::resolve_extract; +use crate::workflow::types::StepResult; +use crate::workflow::{ + template::render_template, + types::{FlowConfig, FlowStep}, +}; + +#[derive(Debug, Error)] +pub enum FlowError { + #[error("{name}字段解析失败")] + Abort { name: String }, + #[error("跳转到第{index}步")] + Jump { index: usize }, + #[error("执行任务出现错误：{message}")] + Other { message: String }, +} +const DEFAULT_TIMEOUT: u64 = 30; + +pub async fn start_flow(config: &FlowConfig, args: Value) -> Result { + info!( + target: "ias::flow", + flow = %config.name, + param_count = config.params.len(), + step_count = config.steps.len(), + "flow 开始" + ); + let arg_keys = args + .as_object() + .map(|values| values.keys().cloned().collect::>()) + .unwrap_or_default(); + debug!(target: "ias::flow", flow = %config.name, ?arg_keys, "flow 入参键"); + + let mut params: HashMap = HashMap::new(); + for param in config.params.clone() { + let name = param.name; + if !name.is_empty() { + if let Some(value) = args.get(name.clone()) { + params.insert(name, value_to_string(value)); + } else if param.required { + if let Some(default) = param.default { + params.insert(name, default); + } else { + warn!( + target: "ias::flow", + flow = %config.name, + param = %name, + "flow 缺少必填参数" + ); + return Ok(format!("参数{}为必传", name)); + } + } else if let Some(default) = param.default { + params.insert(name, default); + } + } + } + let param_keys = params.keys().cloned().collect::>(); + debug!(target: "ias::flow", flow = %config.name, ?param_keys, "flow 参数归一化完成"); + + let mut flow_result: String = "工具错误，没有正确处理输出".to_string(); + let mut step_results: HashMap = HashMap::new(); + let mut cache: HashMap = HashMap::new(); + let mut index = 0; + let steps = config.steps.clone(); + + while index < steps.len() { + if let Some(now_step) = steps.get(index) { + info!(target: "ias::flow", flow = %config.name, step_index = index, "flow 步骤开始"); + let result = match deal_steps(index, now_step, ¶ms, &step_results, &mut cache).await + { + Ok(result) => result, + Err(e) => match e { + FlowError::Jump { index: jump } => { + warn!( + target: "ias::flow", + flow = %config.name, + step_index = index, + jump_to = jump, + "flow 步骤触发跳转" + ); + if jump >= steps.len() { + error!( + target: "ias::flow", + flow = %config.name, + step_index = index, + jump_to = jump, + step_count = steps.len(), + "flow 跳转目标不存在" + ); + return get_other_error(format!("跳转步骤{}不存在", jump)); + } + index = jump; + continue; + } + FlowError::Abort { name } => { + warn!( + target: "ias::flow", + flow = %config.name, + step_index = index, + param = %name, + "flow 步骤中止" + ); + return get_other_error(format!("参数{}为必传", name)); + } + FlowError::Other { message } => { + error!( + target: "ias::flow", + flow = %config.name, + step_index = index, + error = %message, + "flow 步骤执行失败" + ); + return get_other_error(message); + } + }, + }; + info!( + target: "ias::flow", + flow = %config.name, + step_index = index, + raw_len = result.raw.len(), + var_count = result.var.len(), + "flow 步骤完成" + ); + flow_result = result.raw.clone(); + step_results.insert(format!("step{}", index), result); + if !now_step.result.is_empty() { + flow_result = match render_template(&now_step.result, ¶ms, &step_results) { + Ok(result) => { + info!( + target: "ias::flow", + flow = %config.name, + step_index = index, + result_len = result.len(), + "flow 命中结果模板，结束" + ); + result + } + Err(e) => { + error!( + target: "ias::flow", + flow = %config.name, + step_index = index, + error = %e, + "flow 结果模板渲染失败" + ); + return get_other_error(e.to_string()); + } + }; + return Ok(flow_result); + } + if let Some(next) = now_step.next { + if next >= steps.len() { + error!( + target: "ias::flow", + flow = %config.name, + step_index = index, + next_step = next, + step_count = steps.len(), + "flow next 目标不存在" + ); + return get_other_error(format!("跳转步骤{}不存在", next)); + } + info!( + target: "ias::flow", + flow = %config.name, + step_index = index, + next_step = next, + "flow 使用显式下一步" + ); + index = next; + } else { + index += 1; + } + }; + } + info!( + target: "ias::flow", + flow = %config.name, + result_len = flow_result.len(), + "flow 完成" + ); + Ok(flow_result) +} + +async fn deal_steps( + step_index: usize, + step: &FlowStep, + params: &HashMap, + step_results: &HashMap, + cache: &mut HashMap, +) -> Result { + let template_step = deal_template(step_index, step, params, step_results).await?; + deal_requst(step_index, &template_step, cache).await +} + +async fn deal_template( + step_index: usize, + step: &FlowStep, + params: &HashMap, + step_results: &HashMap, +) -> Result { + let json = match serde_json::to_string(step) { + Ok(json) => Ok(json), + Err(e) => { + error!(target: "ias::flow", step_index, error = %e, "flow 步骤序列化失败"); + get_other_error(e.to_string()) + } + }?; + let template_json = match render_template(&json, params, step_results) { + Ok(json) => Ok(json), + Err(e) => { + error!(target: "ias::flow", step_index, error = %e, "flow 步骤模板渲染失败"); + get_other_error(e.to_string()) + } + }?; + match serde_json::from_str(&template_json) { + Ok(flow) => Ok(flow), + Err(e) => { + error!(target: "ias::flow", step_index, error = %e, "flow 步骤模板解析失败"); + get_other_error(e.to_string()) + } + } +} + +async fn deal_requst( + step_index: usize, + step: &FlowStep, + cache: &mut HashMap, +) -> Result { + let url = step.url.clone(); + if url.is_empty() || !url.starts_with("http") { + warn!(target: "ias::flow", step_index, url = %url, "flow 步骤 URL 无效"); + return Ok(StepResult { + var: HashMap::new(), + raw: "请输入正确的url".to_string(), + }); + } + let method = deal_method(step.method.clone()); + let method_name = method.as_str().to_string(); + let should_send_body = method != Method::GET; + let cache_key = step.cache.unwrap_or(false).then(|| build_cache_key(step)); + if let Some(key) = &cache_key + && let Some(result) = cache.get(key) + { + info!( + target: "ias::flow", + step_index, + cache_key_len = key.len(), + raw_len = result.raw.len(), + var_count = result.var.len(), + "flow 缓存命中" + ); + return Ok(result.clone()); + } + let client = Client::new(); + let mut request = client.request(method, url.clone()); + request = request.query(&step.query.clone()); + let headers = match deal_headers(step.headers.clone()) { + Ok(headers) => headers, + Err(e) => { + error!(target: "ias::flow", step_index, error = %e, "flow 请求头解析失败"); + return Err(e); + } + }; + request = request.headers(headers); + if should_send_body && let Some(body) = step.body.clone() { + request = request.body(body); + } + + let timeout = step.timeout.unwrap_or(DEFAULT_TIMEOUT); + request = request.timeout(Duration::from_secs(timeout)); + let query_count = step.query.as_ref().map_or(0, Vec::len); + let header_count = step.headers.as_ref().map_or(0, HashMap::len); + let body_len = step.body.as_ref().map_or(0, String::len); + info!( + target: "ias::flow", + step_index, + method = %method_name, + url = %url, + timeout_secs = timeout, + query_count, + header_count, + body_len, + "flow HTTP 请求开始" + ); + + let started_at = Instant::now(); + let raw = match request.send().await { + Ok(response) => { + let status = response.status(); + match response.text().await { + Ok(raw) => { + info!( + target: "ias::flow", + step_index, + status = %status, + elapsed_ms = started_at.elapsed().as_millis(), + response_len = raw.len(), + "flow HTTP 响应完成" + ); + raw + } + Err(e) => { + error!( + target: "ias::flow", + step_index, + status = %status, + elapsed_ms = started_at.elapsed().as_millis(), + error = %e, + "flow HTTP 响应读取失败" + ); + return get_other_error(e.to_string()); + } + } + } + Err(e) => { + error!( + target: "ias::flow", + step_index, + elapsed_ms = started_at.elapsed().as_millis(), + error = %e, + "flow HTTP 请求失败" + ); + return get_other_error(e.to_string()); + } + }; + let vars = deal_var(step_index, step, &raw)?; + let result = StepResult { var: vars, raw }; + if let Some(key) = cache_key { + debug!( + target: "ias::flow", + step_index, + cache_key_len = key.len(), + "flow 写入缓存" + ); + cache.insert(key, result.clone()); + } + Ok(result) +} + +fn deal_var( + step_index: usize, + step: &FlowStep, + raw: &str, +) -> Result, FlowError> { + let mut vars: HashMap = HashMap::new(); + if !step.var.is_empty() { + debug!( + target: "ias::flow", + step_index, + var_count = step.var.len(), + response_len = raw.len(), + "flow 开始提取变量" + ); + for var in step.var.clone() { + let mut value = String::new(); + let name = var.name; + if let Some(expr) = &var.extract { + match resolve_extract(raw, expr) { + Ok(Some(v)) => { + debug!( + target: "ias::flow", + step_index, + var = %name, + value_len = v.len(), + "flow 变量提取成功" + ); + value = v; + } + Ok(None) | Err(_) => { + warn!( + target: "ias::flow", + step_index, + var = %name, + extract = %expr, + "flow 变量提取为空或失败" + ); + if let Some(abort) = var.abort + && abort + { + warn!( + target: "ias::flow", + step_index, + var = %name, + "flow 变量提取失败，触发中止" + ); + return Err(FlowError::Abort { name }); + } + if let Some(jump) = var.jump { + warn!( + target: "ias::flow", + step_index, + var = %name, + jump_to = jump, + "flow 变量提取失败，触发跳转" + ); + return Err(FlowError::Jump { index: jump }); + } + if let Some(default) = &var.default { + debug!( + target: "ias::flow", + step_index, + var = %name, + default_len = default.len(), + "flow 使用变量默认值" + ); + value = default.clone(); + } + } + } + } + if !value.is_empty() { + debug!( + target: "ias::flow", + step_index, + var = %name, + value_len = value.len(), + "flow 写入变量" + ); + vars.insert(name, value); + } else { + debug!( + target: "ias::flow", + step_index, + var = %name, + "flow 变量为空，跳过写入" + ); + } + } + } + Ok(vars) +} + +fn deal_method(origin: Option) -> Method { + let method_str = match origin { + Some(method) => method.to_uppercase(), + None => "GET".to_string(), + }; + match method_str.as_str() { + "POST" => Method::POST, + "PUT" => Method::PUT, + "PATCH" => Method::PATCH, + "DELETE" => Method::DELETE, + _ => Method::GET, + } +} + +fn deal_headers(origin: Option>) -> Result { + let headers = origin.unwrap_or_default(); + let mut result = HeaderMap::new(); + for (k, v) in headers { + let name = match HeaderName::from_str(&k) { + Ok(name) => name, + Err(_) => return get_other_error(format!("请求头{}无效", k)), + }; + let value = match HeaderValue::from_str(&v) { + Ok(value) => value, + Err(_) => return get_other_error(format!("请求头{}的值无效", k)), + }; + result.append(name, value); + } + Ok(result) +} + +fn build_cache_key(step: &FlowStep) -> String { + let mut headers: Vec<(String, String)> = step + .headers + .clone() + .unwrap_or_default() + .into_iter() + .collect(); + headers.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1))); + + format!( + "{}|{:?}|{:?}|{:?}|{}", + step.url, + step.method, + headers, + step.query, + step.body.as_deref().unwrap_or("") + ) +} + +fn get_other_error(message: String) -> Result { + Err(FlowError::Other { message }) +} + +fn value_to_string(value: &Value) -> String { + match value { + Value::String(value) => value.clone(), + Value::Number(value) => value.to_string(), + Value::Bool(value) => value.to_string(), + Value::Null => String::new(), + _ => value.to_string(), + } +} diff --git a/ai/src/workflow/mod.rs b/ai/src/workflow/mod.rs new file mode 100644 index 0000000..dbecbec --- /dev/null +++ b/ai/src/workflow/mod.rs @@ -0,0 +1,5 @@ +pub mod core; +pub mod flow; +pub mod resolve; +pub mod template; +pub mod types; diff --git a/ai/src/workflow/resolve.rs b/ai/src/workflow/resolve.rs new file mode 100644 index 0000000..84049f6 --- /dev/null +++ b/ai/src/workflow/resolve.rs @@ -0,0 +1,168 @@ +use regex::Regex; +use serde_json::Value; + +enum Segment { + Key(String), + Index(usize), + All, +} + +/// a.b / a.b[0].c / a.b[].c → Vec +fn parse_path(expr: &str) -> Result, String> { + if expr.is_empty() { + return Ok(vec![]); + } + let mut segments = Vec::new(); + for part in expr.split('.') { + if part.is_empty() { + return Err(format!("路径 \"{}\" 中存在空段", expr)); + } + if let Some(key) = part.strip_suffix("[]") { + if key.is_empty() { + return Err(format!("路径 \"{}\" 中 \"[]\" 前缺少键名", expr)); + } + segments.push(Segment::Key(key.to_string())); + segments.push(Segment::All); + } else if let Some(b) = part.find('[') { + let key = &part[..b]; + let inner = &part[b + 1..]; + let inner = inner + .strip_suffix(']') + .ok_or_else(|| format!("路径 \"{}\" 中 \"{}\" 缺少闭合 ']'", expr, part))?; + if key.is_empty() { + return Err(format!( + "路径 \"{}\" 中索引 \"[{}\" 前缺少键名", + expr, inner + )); + } + let idx: usize = inner + .parse() + .map_err(|_| format!("路径 \"{}\" 中索引 \"{}\" 不是数字", expr, inner))?; + segments.push(Segment::Key(key.to_string())); + segments.push(Segment::Index(idx)); + } else { + segments.push(Segment::Key(part.to_string())); + } + } + Ok(segments) +} + +fn resolve_path(value: &Value, segments: &[Segment]) -> Option { + if segments.is_empty() { + return Some(value.clone()); + } + match &segments[0] { + Segment::Key(k) => value + .as_object()? + .get(k) + .and_then(|v| resolve_path(v, &segments[1..])), + Segment::Index(i) => value + .as_array()? + .get(*i) + .and_then(|v| resolve_path(v, &segments[1..])), + Segment::All => { + let arr = value.as_array()?; + let results: Vec = arr + .iter() + .filter_map(|item| resolve_path(item, &segments[1..])) + .collect(); + Some(Value::Array(results)) + } + } +} + +fn value_to_string(v: &Value) -> String { + match v { + Value::String(s) => s.clone(), + Value::Number(n) => n.to_string(), + Value::Bool(b) => b.to_string(), + Value::Null => String::new(), + _ => v.to_string(), + } +} + +fn detect_and_parse(raw: &str) -> Option { + serde_json::from_str(raw) + .ok() + .or_else(|| serde_yaml::from_str(raw).ok()) +} + +/// 统一提取入口：/pattern/ → 正则引擎，否则 → 路径引擎 +pub fn resolve_extract(raw: &str, expr: &str) -> Result, String> { + if let Some(inner) = expr.strip_prefix('/').and_then(|s| s.strip_suffix('/')) { + let re = Regex::new(inner).map_err(|e| format!("正则语法错误: {}", e))?; // 编译正则 + Ok(re + .captures(raw) + .and_then(|caps| caps.get(1).map(|m| m.as_str().to_string()))) // 取捕获组1 + } else { + let value = + detect_and_parse(raw).ok_or_else(|| "响应体无法解析为 JSON 或 YAML".to_string())?; // JSON → YAML 自动检测 + let segments = parse_path(expr)?; // 解析 a.b[0].c 语法 + Ok(resolve_path(&value, &segments).map(|v| value_to_string(&v))) // None = 键/索引不存在 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_simple() { + let segs = parse_path("a.b.c").unwrap(); + assert_eq!(segs.len(), 3); + } + + #[test] + fn test_parse_index() { + let segs = parse_path("items[0].name").unwrap(); + assert_eq!(segs.len(), 3); // Key("items"), Index(0), Key("name") + } + + #[test] + fn test_parse_all() { + let segs = parse_path("books[].title").unwrap(); + assert_eq!(segs.len(), 3); // Key("books"), All, Key("title") + } + + #[test] + fn test_resolve_simple() { + let json = r#"{"auth":{"token":"abc123"}}"#; + assert_eq!( + resolve_extract(json, "auth.token").unwrap(), + Some("abc123".into()) + ); + } + + #[test] + fn test_resolve_array_index() { + let json = r#"{"items":[{"name":"a"},{"name":"b"}]}"#; + assert_eq!( + resolve_extract(json, "items[0].name").unwrap(), + Some("a".into()) + ); + } + + #[test] + fn test_resolve_array_all() { + let json = r#"{"books":[{"title":"x"},{"title":"y"}]}"#; + assert_eq!( + resolve_extract(json, "books[].title").unwrap(), + Some("[\"x\",\"y\"]".into()) + ); + } + + #[test] + fn test_resolve_regex() { + let raw = r#"token=abc123; path=/"#; + assert_eq!( + resolve_extract(raw, "/token=([^;]+)/").unwrap(), + Some("abc123".into()) + ); + } + + #[test] + fn test_resolve_key_not_found() { + let json = r#"{"a":1}"#; + assert_eq!(resolve_extract(json, "a.b").unwrap(), None); + } +} diff --git a/ai/src/workflow/template.rs b/ai/src/workflow/template.rs new file mode 100644 index 0000000..9f9b7be --- /dev/null +++ b/ai/src/workflow/template.rs @@ -0,0 +1,71 @@ +use std::collections::HashMap; + +use anyhow::Ok; +use regex::{Captures, Regex}; + +use crate::workflow::types::StepResult; + +pub fn render_template( + input: &str, + param: &HashMap, + steps_result: &HashMap, +) -> anyhow::Result { + let re = Regex::new(r"\$\{(env|param|step[0-9]+):([A-Za-z0-9_]+)\}").unwrap(); + let result = re + .replace_all(input, |caps: &Captures| { + let kind = &caps[1]; + let key = &caps[2]; + match kind { + "env" => std::env::var(key).unwrap_or_default(), + "param" => param.get(key).cloned().unwrap_or_default(), + _ if kind.starts_with("step") => { + deal_step_data(kind, key, steps_result).unwrap_or_default() + } + _ => "".to_string(), + } + }) + .to_string(); + Ok(result) +} + +fn deal_step_data( + kind: &str, + key: &str, + steps_result: &HashMap, +) -> anyhow::Result { + Ok(steps_result + .get(kind) + .and_then(|step| step.var.get(key)) + .cloned() + .unwrap_or_default()) +} + +#[test] +fn test_template() { + unsafe { + std::env::set_var("BASE_URL", "https://api.example.com"); + } + + let mut params = HashMap::new(); + params.insert("order_id".to_string(), "12345".to_string()); + + let mut step_vars = HashMap::new(); + step_vars.insert("name".to_string(), "demo".to_string()); + + let mut steps_result = HashMap::new(); + steps_result.insert( + "step0".to_string(), + StepResult { + raw: String::new(), + var: step_vars, + }, + ); + + let input = "${env:BASE_URL}/orders/${param:order_id}?key=${step0:name}"; + let output = render_template(input, ¶ms, &steps_result); + + assert_eq!( + output.unwrap(), + "https://api.example.com/orders/12345?key=demo" + ); +} diff --git a/ai/src/workflow/types.rs b/ai/src/workflow/types.rs new file mode 100644 index 0000000..57f32d8 --- /dev/null +++ b/ai/src/workflow/types.rs @@ -0,0 +1,57 @@ +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone)] +pub struct StepResult { + pub raw: String, + pub var: HashMap, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct FlowConfig { + pub name: String, + pub desc: String, + #[serde(default)] + pub params: Vec, + #[serde(default, skip_serializing)] + pub env: Vec, + #[serde(skip_serializing)] + pub steps: Vec, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct FlowParam { + pub name: String, + #[serde(rename = "type")] + pub param_type: String, + pub required: bool, + pub description: String, + #[serde(skip_serializing)] + pub default: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct FlowStep { + pub url: String, + pub method: Option, + pub headers: Option>, + pub query: Option>, + pub body: Option, + #[serde(default)] + pub var: Vec, + pub timeout: Option, + pub cache: Option, + pub next: Option, + #[serde(default)] + pub result: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct FlowVar { + pub name: String, + pub extract: Option, + pub default: Option, + pub abort: Option, + pub jump: Option, +} diff --git a/common/Cargo.toml b/common/Cargo.toml index 31fa7c3..10eff45 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -4,8 +4,12 @@ version = "0.1.0" edition = "2024" [dependencies] +# ── HTTP 客户端 ── +reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls"] } # HTTP 请求（微信 API + DeepSeek API + 工具调用） # ── 序列化 ── -serde = { version = "1.0", features = ["derive"] } # 序列化/反序列化 -serde_json = "1.0" +serde = { version = "1.0", features = ["derive"] } # 序列化/反序列化 +serde_json = "1.0" # JSON 处理 +serde_yaml = "0.9" # ── 错误处理 ── +anyhow = "1" thiserror = "2" diff --git a/common/src/error/queue.rs b/common/src/error/queue.rs index e69de29..8b13789 100644 --- a/common/src/error/queue.rs +++ b/common/src/error/queue.rs @@ -0,0 +1 @@ + diff --git a/common/src/model/ai.rs b/common/src/model/ai.rs index e5833b0..99b9044 100644 --- a/common/src/model/ai.rs +++ b/common/src/model/ai.rs @@ -43,7 +43,7 @@ impl ChatCompletionRequestMessage { role, content, tool_call_id: None, - tool_calls: tool_calls, + tool_calls, } } diff --git a/common/src/queue.rs b/common/src/queue.rs index 176bbd3..3ebfd94 100644 --- a/common/src/queue.rs +++ b/common/src/queue.rs @@ -5,6 +5,7 @@ use crate::model::ai::ChatCompletionToolCall; #[derive(Debug, Clone, Serialize, Deserialize)] pub enum QueueMessage { Channel(ChannelMessage), + Notice(ChannelMessage, String), Aissitant(ChannelMessage, AssistantMessage), Tool(ChannelMessage, String, String, String), } diff --git a/docs/TOOLKIT_WORKFLOW.md b/docs/TOOLKIT_WORKFLOW.md new file mode 100644 index 0000000..f054fb0 --- /dev/null +++ b/docs/TOOLKIT_WORKFLOW.md @@ -0,0 +1,325 @@ +# TOOLKIT WORKFLOW — YAML 驱动的 HTTP 工具链引擎 + +## 概述 + +TOOLKIT WORKFLOW 让你通过编写 YAML 文件来定义 LLM 工具。一个 YAML 文件描述一个工具：它接受哪些参数、调用哪些 HTTP API、如何从响应中提取数据、以及最终返回什么给 LLM。 + +核心能力： +- 一个 YAML 文件 = 一个完整的 LLM 工具，无需写代码 +- 支持多步 HTTP 请求链，步骤之间可以传递变量 +- 流程控制：`next` 跳转步骤、`jump` 降级到备源、`abort` 终止执行 +- 键级缓存：`cache` 缓存变量，同一调用中缺键自动重请求 +- 统一提取：`extract` 字段，语法自描述引擎（路径 / 正则） + +--- + +## YAML 顶层结构 + +一个工具 YAML 文件包含以下顶层字段： + +| 字段 | 必填 | 类型 | 说明 | +|------|------|------|------| +| `name` | 是 | string | 工具名，对应 LLM function.name | +| `desc` | 是 | string | 工具描述，对应 LLM function.description | +| `params` | 否 | array | LLM 可见的参数定义，映射为 function.parameters JSON Schema | +| `env` | 否 | array[string] | 该工具依赖的环境变量名列表，仅作文档和校验 | +| `steps` | 是 | array | 请求步骤链，按顺序执行 | + +--- + +## `params` — 参数定义 + +定义 LLM 可以传入的参数，映射为 function calling 的 parameters schema。 + +| 字段 | 必填 | 类型 | 说明 | +|------|------|------|------| +| `name` | 是 | string | 参数名，在模板中通过 `${param:key}` 引用 | +| `type` | 是 | string | 取值：`string` / `number` / `integer` / `boolean` | +| `required` | 是 | boolean | 是否必填 | +| `description` | 是 | string | 参数说明，LLM 据此理解参数含义 | +| `default` | 否 | any | 默认值，LLM 未传时自动使用 | + +--- + +## `steps` — 请求步骤 + +每个步骤定义一个 HTTP 请求及其后续处理。 + +### 步骤字段 + +| 字段 | 必填 | 类型 | 默认值 | 说明 | +|------|------|------|--------|------| +| `url` | 是 | string | — | 请求 URL，支持模板语法 | +| `method` | 否 | string | `GET` | HTTP 方法：`GET` / `POST` / `PUT` / `DELETE` / `PATCH` | +| `headers` | 否 | map | — | 请求头键值对，值支持模板语法 | +| `query` | 否 | array | — | URL 查询参数，格式为 `[{key, value}]` 元组列表，值支持模板语法 | +| `body` | 否 | string | — | 请求体，支持模板语法 | +| `timeout` | 否 | integer | 30 | 超时时间（秒） | +| `cache` | 否 | boolean | false | 是否对该步骤启用键级缓存 | +| `next` | 否 | integer | — | 步骤成功后跳转到第几个步骤（0-based），不设则顺序执行下一步 | +| `var` | 否 | array | — | 变量提取列表 | +| `result` | 否 | string | — | 最终返回模板，有此字段则该步为终步 | + +### 步骤示例 + +```yaml +steps: + - url: ${env:BASE_URL}/auth/token + method: POST + headers: + Content-Type: application/json + body: | + {"client_id": "${env:CLIENT_ID}", "client_secret": "${env:CLIENT_SECRET}"} + timeout: 15 + cache: true + var: + - name: token + extract: access_token + abort: true + next: 1 + + - url: ${env:BASE_URL}/orders/${param:order_id} + method: GET + headers: + Authorization: Bearer ${step0:token} + var: + - name: status + extract: data.status + - name: amount + extract: data.amount + result: | + 订单号: ${param:order_id} + 状态: ${step1:status} + 金额: ${step1:amount} +``` + +--- + +## `var` — 变量提取 + +每个步骤执行完 HTTP 请求后，按顺序处理 `var` 列表中的条目，从响应中提取命名变量。提取的变量通过 `${stepN:var_name}` 在后续步骤中引用。 + +### `extract` — 提取表达式 + +一个字段覆盖所有提取场景。值的语法决定使用的引擎： + +| 语法 | 引擎 | 说明 | +|------|------|------| +| 不以 `/` 开头 | 路径引擎 | 对 JSON/YAML 响应执行结构化路径导航 | +| 以 `/` 开头且以 `/` 结尾 | 正则引擎 | 对原始响应文本执行正则匹配，取捕获组 1 | + +**路径引擎**：自动检测响应格式（先 JSON，失败则 YAML），然后按路径导航取值。 + +| 语法 | 含义 | 示例 | +|------|------|------| +| `key` | 访问对象字段 | `data.status` | +| `key[N]` | 访问数组第 N 个元素 | `items[0].name` | +| `key[]` | 遍历数组所有元素 | `books[]` | +| `key[].field` | 遍历数组，对每个元素取字段 | `books[].title` | + +> 键名建议仅使用 `[a-zA-Z0-9_]`，避免含 `.` `[` `]` 的特殊键名。 + +**正则引擎**：`/pattern/` 形式，对原始响应文本执行正则匹配，取第一个捕获组。 + +```yaml +# 路径：从 JSON 响应中提取 +var: + - name: token + extract: auth.access_token + +# 正则：从任意文本中提取 +var: + - name: session + extract: /session[= ]([a-f0-9]+)/ +``` + +### 提取失败处理（每个条目最多选一种） + +| 配置 | 提取成功 | 提取失败 | +|------|---------|---------| +| 不设任何失败处理 | 存入变量 | 存入空字符串 `""`，继续执行 | +| `default: "值"` | 存入变量 | 存入默认值，继续执行 | +| `abort: true` | 存入变量 | 终止整个工具，返回错误给 LLM | +| `jump: N` | 存入变量 | 跳转到第 N 个步骤执行，N 从 0 开始 | + +### var 条目完整字段 + +| 字段 | 必填 | 类型 | 默认值 | 说明 | +|------|------|------|--------|------| +| `name` | 是 | string | — | 变量名，通过 `${stepN:name}` 引用 | +| `extract` | 否 | string | — | 提取表达式，语法自描述引擎（路径 / 正则） | +| `default` | 否 | string | — | 提取失败时的默认值 | +| `abort` | 否 | boolean | false | 提取失败时是否终止工具 | +| `jump` | 否 | integer | — | 提取失败时跳转到第几个步骤（0-based） | + + + +--- + +## 流程控制 + +三种流程控制机制，覆盖顺序执行、条件跳转和异常终止。 + +### `next` — 成功后跳转 + +- 位置：步骤级别 +- 触发条件：步骤所有 var 都没有因 abort/jump 而失败 +- 效果：跳转到指定步骤号（0-based），不设则顺序执行下一步 +- 用途：跳过某些步骤、实现条件分支 + +### `jump` — 提取失败时降级 + +- 位置：var 条目级别 +- 触发条件：该条目的提取失败且未设置 default +- 效果：跳转到指定步骤号（0-based），当前步骤剩余 var 不再执行 +- 用途：主数据源失败时自动切换到备数据源 + +### `abort` — 提取失败时终止 + +- 位置：var 条目级别 +- 触发条件：该条目的提取失败 +- 效果：终止整个工具执行，返回错误信息给 LLM +- 用途：关键变量（如认证 token）获取失败时立即报错 + +--- + +## `cache` — 键级缓存 + +当步骤设置 `cache: true` 时，该步骤提取的所有变量会被缓存。 + +缓存行为： +- **命中**：缓存中存在该步骤所有 var.name → 跳过 HTTP 请求，直接返回缓存值 +- **未命中**：任意一个 var.name 缺失 → 发起 HTTP 请求，提取所有变量，全部存入缓存 +- **生命周期**：单次工具调用期间（内存级），工具返回后释放 +- **缓存键**：由步骤的 `url + method + headers + body` 模板展开后的哈希确定 + +典型用法：第一步获取认证 token 并缓存，后续多步复用 token 时不会重复请求认证接口。 + +--- + +## 模板语法 + +以下字段支持模板语法：`url`、`headers` 的值、`body`、`result`。 + +### 三种引用 + +| 语法 | 来源 | 示例 | +|------|------|------| +| `${param:key}` | LLM 传入的工具参数 | `${param:city}`、`${param:order_id}` | +| `${env:VAR}` | 系统环境变量 | `${env:API_KEY}`、`${env:BASE_URL}` | +| `${stepN:var_name}` | 第 N 个步骤提取的变量，N 从 0 开始 | `${step0:token}`、`${step1:status}` | + +### 约束 + +1. 禁止嵌套引用：`${step0:${env:KEY}}` 不允许 +2. 禁止原始步骤引用：`${step:0}` 不允许，必须先通过 var 提取为命名变量 + + +--- + +## `result` — 最终返回 + +步骤的 `result` 字段定义该工具最终返回给 LLM 的内容。 + +- 如果某步设置了 `result`，该步即为**终步**，执行后不再继续后续步骤 +- 如果所有步骤都未设置 `result`，默认返回最后一步的原始响应体 +- `result` 支持模板语法，可以拼接多个变量和文字 + +```yaml +# 返回单个变量 +result: ${step1:status} + +# 多行模板拼接 +result: | + 订单号: ${param:order_id} + 状态: ${step1:status} + 金额: ${step1:amount} +``` + +--- + +## 执行流程 + +``` +1. 加载 YAML → 解析为 YamlToolDef +2. 解析 LLM 参数 → params HashMap +3. 加载 env 声明的环境变量 → env HashMap +4. 初始化 cache = {} + +5. step_idx = 0 + while step_idx < steps.len(): + + ┌─ 当前步骤执行 ──────────────────────────────┐ + │ │ + │ [cache: true?] │ + │ ├─ 缓存命中（所有 var 键都存在）→ 跳过HTTP │ + │ └─ 缓存未命中 → 继续 │ + │ │ + │ 展开 url/headers/body 中的模板 │ + │ ${param:key} → params["key"] │ + │ ${env:VAR} → env["VAR"] │ + │ ${stepN:var} → ctx.vars[(N, var)] │ + │ │ + │ HttpRequest 执行 → HttpResponse │ + │ 保存到 ctx.steps[step_num] │ + │ │ + │ for each var: │ + │ extract 语法分发: │ + │ ├─ /.../ → 正则引擎，取捕获组1 │ + │ └─ 其他 → 路径引擎（JSON→YAML 自动检测） │ + │ ├─ 成功 → 存入 ctx.vars + step_vars │ + │ └─ 失败 → │ + │ ├─ abort:true → 终止，返回错误 │ + │ ├─ default:x → 使用默认值 │ + │ ├─ jump:N → step_idx = N, break │ + │ └─ (无) → 空字符串 │ + │ │ + │ [cache: true?] → 将 step_vars 存入 cache │ + │ │ + │ [result 存在?] │ + │ └─ 展开 result 模板 → 返回给 LLM → 结束 │ + │ │ + │ [next 存在?] │ + │ ├─ next:N → step_idx = N │ + │ └─ (无) → step_idx += 1 │ + │ │ + └──────────────────────────────────────────────┘ + +6. 循环结束 → 返回最后一步原始响应体 +``` + +--- + +## 错误处理 + +| 场景 | 处理方式 | +|------|---------| +| YAML 文件格式错误 | 启动时打印警告，跳过该文件 | +| 模板引用了未定义的参数 | 替换为空字符串，记录警告日志 | +| 环境变量未设置 | 替换为空字符串，记录警告日志 | +| HTTP 请求失败（连接/超时） | 返回错误详情给 LLM | +| `var.abort: true` 且提取失败 | 终止，返回 "变量 xxx 提取失败" | +| `var.jump: N` 且提取失败 | 跳转到第 N 个步骤，N 从 0 开始，不报错 | +| `var.default` 且提取失败 | 使用默认值，不报错 | +| 路径语法错误（空段、非法索引） | 返回错误给 LLM，附带路径表达式 | +| 路径键不存在 / 索引越界 | 提取失败，触发 default/abort/jump | +| 正则语法错误 | 返回错误给 LLM，附带正则表达式 | +| 正则无匹配 | 提取失败，触发 default/abort/jump | +| 响应体非 JSON 也非 YAML（路径引擎） | 返回错误给 LLM | +| `next` / `jump` 目标步骤不存在 | 终止，返回配置错误 | + +--- + +## 扩展预留 + +以下字段预留供后续版本实现： + +| 字段 | 位置 | 用途 | +|------|------|------| +| `retry` | `steps[]` | 失败重试次数与退避策略 | +| `cache.ttl` | `steps[]` | 缓存有效期（秒） | +| `validate` | `params[]` | 参数级正则校验 | +| `sensitive` | `env[]` | 标记敏感环境变量，日志脱敏 | +| `loop` | `steps[]` | 条件循环（分页自动翻页） | +| `export` | `steps[]` | 将变量导出为环境变量供子进程使用 | +| `pre_request` | 顶层 | 工具级别的前置请求（全局认证） | diff --git a/service/Cargo.toml b/service/Cargo.toml index 0047665..24048ea 100644 --- a/service/Cargo.toml +++ b/service/Cargo.toml @@ -3,6 +3,10 @@ name = "service" version = "0.1.0" edition = "2024" +[[bin]] +name = "ias" +path = "src/main.rs" + [dependencies] # ── 公共模块 ── common = { path = "../common" } @@ -21,6 +25,7 @@ tokio = { version = "1.0", features = [ "process", "signal", "io-util", + "net", ] } # 异步运行时核心 # ── HTTP 客户端 ── reqwest = { version = "0.12", default-features = false, features = [ @@ -46,7 +51,12 @@ sqlx = { version = "0.9.0", features = [ # ── 错误处理 ── anyhow = "1" thiserror = "2" +# ── 日志管理 ── +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } # ── cli ── clap = { version = "4.6.1", features = ["derive"] } # ── 时间处理 ── -chrono = { version = "0.4", features = ["serde"] } # 日期时间 \ No newline at end of file +chrono = { version = "0.4", features = ["serde"] } # 日期时间 +# ── 唯一标识 ── +uuid = { version = "1", features = ["v4", "serde"] } diff --git a/service/migrations/20260706000000_create_web_pages.sql b/service/migrations/20260706000000_create_web_pages.sql new file mode 100644 index 0000000..705a98a --- /dev/null +++ b/service/migrations/20260706000000_create_web_pages.sql @@ -0,0 +1,13 @@ +CREATE TABLE ias_web_pages ( + id BIGSERIAL PRIMARY KEY, + slug VARCHAR(64) NOT NULL UNIQUE, + page_type VARCHAR(64) NOT NULL, + title VARCHAR(255) NOT NULL, + summary TEXT, + content TEXT NOT NULL, + source_label VARCHAR(255), + metadata TEXT NOT NULL DEFAULT '{}', + created_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_ias_web_pages_type_slug ON ias_web_pages(page_type, slug); diff --git a/service/src/channel/mod.rs b/service/src/channel/mod.rs index 49e84a4..5c1fe2e 100644 --- a/service/src/channel/mod.rs +++ b/service/src/channel/mod.rs @@ -1 +1 @@ -pub mod wechat; \ No newline at end of file +pub mod wechat; diff --git a/service/src/channel/wechat/looper.rs b/service/src/channel/wechat/looper.rs index 0d63adc..0130edf 100644 --- a/service/src/channel/wechat/looper.rs +++ b/service/src/channel/wechat/looper.rs @@ -17,6 +17,7 @@ pub async fn start_loop( account_id, updates_buf, } => { + state.runtime.record_wechat_event(account_id.clone()).await; persist_updates_buf(&state.db, account_id, updates_buf).await; } WeChatEvent::Message { @@ -24,6 +25,7 @@ pub async fn start_loop( message, updates_buf, } => { + state.runtime.record_wechat_event(account_id.clone()).await; persist_updates_buf(&state.db, account_id.clone(), updates_buf).await; if message.msg_type != Some(WeixinMessage::TYPE_USER) { @@ -43,7 +45,7 @@ pub async fn start_loop( .clone() .send(QueueMessage::Channel(ChannelMessage { channel: "wechat".to_string(), - account_id: account_id, + account_id, from_user_id: from_user_id.to_string(), content: text.to_string(), context: message.context_token, @@ -52,6 +54,10 @@ pub async fn start_loop( .unwrap(); } WeChatEvent::PollError { account_id, error } => { + state + .runtime + .record_wechat_error(account_id.clone(), error.clone()) + .await; println!("微信账号轮询失败 account_id={account_id}: {error}"); } } diff --git a/service/src/channel/wechat/manager.rs b/service/src/channel/wechat/manager.rs index cddc73c..6f4800c 100644 --- a/service/src/channel/wechat/manager.rs +++ b/service/src/channel/wechat/manager.rs @@ -27,6 +27,10 @@ pub async fn init_wechat(state: Option) -> Option { diff --git a/service/src/channel/wechat/mod.rs b/service/src/channel/wechat/mod.rs index 0b945a2..19c533d 100644 --- a/service/src/channel/wechat/mod.rs +++ b/service/src/channel/wechat/mod.rs @@ -1,2 +1,2 @@ +pub mod looper; pub mod manager; -pub mod looper; \ No newline at end of file diff --git a/service/src/core/cli.rs b/service/src/core/cli.rs index 780a508..d6d6eb6 100644 --- a/service/src/core/cli.rs +++ b/service/src/core/cli.rs @@ -12,7 +12,10 @@ pub struct Cli { #[derive(Debug, Subcommand)] pub enum Commands { // 启动服务 - Service, + Service { + #[command(subcommand)] + command: Option, + }, // 配置渠道 Channel { #[command(subcommand)] @@ -20,6 +23,26 @@ pub enum Commands { }, } +#[derive(Debug, Clone, Subcommand)] +pub enum ServiceAction { + /// 前台运行服务 + Run, + /// 后台启动服务 + Start, + /// 停止后台服务 + Stop, + /// 重启后台服务 + Restart, + /// 查看服务运行状态 + Status, + /// 查看后台服务日志 + Logs { + /// 输出最后多少行日志 + #[arg(short, long, default_value_t = 80)] + lines: usize, + }, +} + #[derive(Debug, Subcommand)] pub enum Channels { // 微信渠道配置 diff --git a/service/src/core/control.rs b/service/src/core/control.rs new file mode 100644 index 0000000..2793179 --- /dev/null +++ b/service/src/core/control.rs @@ -0,0 +1,142 @@ +use std::path::PathBuf; +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::watch; + +use crate::core::runtime::{RuntimeSnapshot, RuntimeState}; + +#[derive(Debug, Clone)] +pub struct RuntimePaths { + pub dir: PathBuf, + pub pid_file: PathBuf, + pub socket_file: PathBuf, + pub state_file: PathBuf, + pub log_file: PathBuf, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ControlResponse { + Status { snapshot: RuntimeSnapshot }, + Ack { message: String }, + Error { message: String }, +} + +impl RuntimePaths { + pub fn new() -> Self { + let dir = std::env::var_os("XDG_RUNTIME_DIR") + .map(PathBuf::from) + .unwrap_or_else(|| { + let user = std::env::var("USER").unwrap_or_else(|_| "user".to_string()); + std::env::temp_dir().join(format!("ias-{user}")) + }) + .join("ias"); + + Self { + pid_file: dir.join("ias.pid"), + socket_file: dir.join("ias.sock"), + state_file: dir.join("ias.state.json"), + log_file: dir.join("ias.log"), + dir, + } + } + + pub fn ensure_dir(&self) -> anyhow::Result<()> { + std::fs::create_dir_all(&self.dir)?; + Ok(()) + } +} + +pub async fn start_control_server( + runtime: Arc, + paths: RuntimePaths, + shutdown_tx: watch::Sender, + mut shutdown_rx: watch::Receiver, +) -> anyhow::Result<()> { + paths.ensure_dir()?; + if paths.socket_file.exists() { + if UnixStream::connect(&paths.socket_file).await.is_ok() { + anyhow::bail!( + "服务控制 socket 已存在且可连接: {}", + paths.socket_file.display() + ); + } + std::fs::remove_file(&paths.socket_file)?; + } + + let listener = UnixListener::bind(&paths.socket_file)?; + + loop { + tokio::select! { + result = listener.accept() => { + let (stream, _) = result?; + let runtime = runtime.clone(); + let shutdown_tx = shutdown_tx.clone(); + tokio::spawn(async move { + if let Err(err) = handle_control_stream(stream, runtime, shutdown_tx).await { + eprintln!("处理服务控制命令失败: {err}"); + } + }); + } + changed = shutdown_rx.changed() => { + if changed.is_err() || *shutdown_rx.borrow() { + break; + } + } + } + } + + Ok(()) +} + +pub async fn request_control( + paths: &RuntimePaths, + command: &str, +) -> anyhow::Result { + let mut stream = UnixStream::connect(&paths.socket_file).await?; + stream.write_all(command.as_bytes()).await?; + stream.write_all(b"\n").await?; + + let mut reader = BufReader::new(stream); + let mut line = String::new(); + reader.read_line(&mut line).await?; + let response = serde_json::from_str(line.trim())?; + Ok(response) +} + +async fn handle_control_stream( + stream: UnixStream, + runtime: Arc, + shutdown_tx: watch::Sender, +) -> anyhow::Result<()> { + let mut reader = BufReader::new(stream); + let mut command = String::new(); + reader.read_line(&mut command).await?; + let command = command.trim(); + + let response = match command { + "status" => ControlResponse::Status { + snapshot: runtime.snapshot().await, + }, + "shutdown" => { + runtime.mark_stopping().await; + let _ = shutdown_tx.send(true); + ControlResponse::Ack { + message: "服务正在停止".to_string(), + } + } + _ => ControlResponse::Error { + message: format!("未知控制命令: {command}"), + }, + }; + + let response = serde_json::to_string(&response)?; + let stream = reader.get_mut(); + stream.write_all(response.as_bytes()).await?; + stream.write_all(b"\n").await?; + stream.flush().await?; + Ok(()) +} diff --git a/service/src/core/mod.rs b/service/src/core/mod.rs index 4369de2..9380747 100644 --- a/service/src/core/mod.rs +++ b/service/src/core/mod.rs @@ -1,3 +1,6 @@ pub mod cli; pub mod context; -pub mod queue; \ No newline at end of file +pub mod control; +pub mod queue; +pub mod runtime; +pub mod supervisor; diff --git a/service/src/core/queue.rs b/service/src/core/queue.rs index e232b86..d2f8173 100644 --- a/service/src/core/queue.rs +++ b/service/src/core/queue.rs @@ -2,40 +2,101 @@ use crate::core::context::{ add_assistant_message, add_tool_message, add_user_message, snapshot_context, }; use ai::core::send_message; -use common::queue::QueueMessage; +use common::queue::{ChannelMessage, QueueMessage}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; use tokio::sync::{mpsc::Receiver, mpsc::Sender}; +use tokio::task::JoinHandle; use wechat::manager::WeChatMultiAccountManager; +struct TypingSession { + client_id: String, + refresh_task: JoinHandle<()>, +} + pub async fn create_queue( sender: Sender, mut receiver: Receiver, manager: WeChatMultiAccountManager, -) -> anyhow::Result<()> { - tokio::spawn(async move { +) -> anyhow::Result> { + let manager = Arc::new(manager); + let handle = tokio::spawn(async move { + let mut typing_sessions: HashMap = HashMap::new(); while let Some(msg) = receiver.recv().await { let queue_sender = sender.clone(); match msg { QueueMessage::Aissitant(channel, assistant) => { let content = assistant.content.clone(); let reasoning = assistant.reasoning.clone().unwrap_or_default(); - println!("\n\n收到llm回复：\n{}\n-------\n{}", reasoning, content); - if !content.trim().is_empty() || !reasoning.trim().is_empty() { + let has_tool_calls = assistant + .tool_calls + .as_ref() + .is_some_and(|calls| !calls.is_empty()); + println!( + "\n\n收到llm回复：\n{}\n-------\n{}\n工具调用数：{}", + reasoning, + content, + assistant.tool_calls.as_ref().map_or(0, Vec::len) + ); + + if has_tool_calls { + refresh_typing(manager.clone(), &typing_sessions, &channel).await; + } else if !content.trim().is_empty() || !reasoning.trim().is_empty() { let text = if !content.trim().is_empty() { content.trim() } else { reasoning.trim() }; - manager + let session_key = typing_key(&channel); + if let Some(session) = typing_sessions.remove(&session_key) { + session.refresh_task.abort(); + manager + .send_text_with_client_id( + &channel.account_id, + &channel.from_user_id, + text, + channel.context.as_deref(), + &session.client_id, + ) + .await + .unwrap(); + } else { + manager + .send_text( + &channel.account_id, + &channel.from_user_id, + text, + channel.context.as_deref(), + ) + .await + .unwrap(); + } + } else if assistant + .tool_calls + .as_ref() + .is_none_or(|calls| calls.is_empty()) + { + stop_typing(&mut typing_sessions, &channel); + } + add_assistant_message(assistant.content, assistant.tool_calls).await; + } + QueueMessage::Notice(channel, text) => { + let text = text.trim(); + if !text.is_empty() { + println!("\n\n发送工具调用提醒：\n{}", text); + if let Err(err) = manager .send_text( &channel.account_id, &channel.from_user_id, - &text, + text, channel.context.as_deref(), ) .await - .unwrap(); + { + eprintln!("发送工具调用提醒失败: {}", err); + } } - add_assistant_message(assistant.content, assistant.tool_calls).await; } QueueMessage::Channel(bundle) => { println!( @@ -43,23 +104,123 @@ pub async fn create_queue( bundle.from_user_id.clone(), bundle.content.clone() ); + start_typing(manager.clone(), &mut typing_sessions, &bundle).await; // todo 目前按单用户处理 add_user_message(bundle.content.clone()).await; - send_message(queue_sender, snapshot_context().await, bundle) - .await - .unwrap(); + let messages = snapshot_context().await; + tokio::spawn(async move { + if let Err(err) = send_message(queue_sender, messages, bundle).await { + eprintln!("发送消息失败: {}", err); + } + }); } QueueMessage::Tool(channel, tool_call_id, name, result) => { println!("\n\n调用{}工具的结果:\n{}", name, result); + refresh_typing(manager.clone(), &typing_sessions, &channel).await; add_tool_message(tool_call_id, result).await; - send_message(queue_sender, snapshot_context().await, channel) - .await - .unwrap(); + let messages = snapshot_context().await; + tokio::spawn(async move { + if let Err(err) = send_message(queue_sender, messages, channel).await { + eprintln!("发送消息失败: {}", err); + } + }); } } } }); - Ok(()) + Ok(handle) +} + +async fn start_typing( + manager: Arc, + sessions: &mut HashMap, + channel: &ChannelMessage, +) { + let key = typing_key(channel); + stop_typing(sessions, channel); + + let client_id = match manager + .send_typing( + &channel.account_id, + &channel.from_user_id, + channel.context.as_deref(), + ) + .await + { + Ok(client_id) => client_id, + Err(err) => { + eprintln!("设置微信输入中状态失败: {}", err); + return; + } + }; + + let account_id = channel.account_id.clone(); + let from_user_id = channel.from_user_id.clone(); + let context = channel.context.clone(); + let refresh_client_id = client_id.clone(); + let refresh_manager = manager.clone(); + let refresh_task = tokio::spawn(async move { + loop { + tokio::time::sleep(Duration::from_secs(4)).await; + if let Err(err) = refresh_manager + .send_typing_with_client_id( + &account_id, + &from_user_id, + context.as_deref(), + &refresh_client_id, + ) + .await + { + eprintln!("刷新微信输入中状态失败: {}", err); + return; + } + } + }); + + sessions.insert( + key, + TypingSession { + client_id, + refresh_task, + }, + ); +} + +async fn refresh_typing( + manager: Arc, + sessions: &HashMap, + channel: &ChannelMessage, +) { + let Some(session) = sessions.get(&typing_key(channel)) else { + return; + }; + + if let Err(err) = manager + .send_typing_with_client_id( + &channel.account_id, + &channel.from_user_id, + channel.context.as_deref(), + &session.client_id, + ) + .await + { + eprintln!("刷新微信输入中状态失败: {}", err); + } +} + +fn stop_typing(sessions: &mut HashMap, channel: &ChannelMessage) { + if let Some(session) = sessions.remove(&typing_key(channel)) { + session.refresh_task.abort(); + } +} + +fn typing_key(channel: &ChannelMessage) -> String { + format!( + "{}:{}:{}", + channel.account_id, + channel.from_user_id, + channel.context.as_deref().unwrap_or_default() + ) } // pub async fn send_to_queue( diff --git a/service/src/core/runtime.rs b/service/src/core/runtime.rs new file mode 100644 index 0000000..ae6dca3 --- /dev/null +++ b/service/src/core/runtime.rs @@ -0,0 +1,195 @@ +use std::collections::HashMap; +use std::path::Path; +use std::process; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use tokio::sync::Mutex; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ServicePhase { + Starting, + Running, + Stopping, + Stopped, + Failed, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AccountRuntimePhase { + Running, + Retrying, + Stopped, + Failed, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AccountRuntimeStatus { + pub account_id: String, + pub phase: AccountRuntimePhase, + pub last_error: Option, + pub last_event_at: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RuntimeSnapshot { + pub phase: ServicePhase, + pub pid: u32, + pub started_at: DateTime, + pub updated_at: DateTime, + pub http_addr: String, + pub queue_alive: bool, + pub wechat_accounts: Vec, + pub last_error: Option, +} + +struct RuntimeInner { + phase: ServicePhase, + started_at: DateTime, + updated_at: DateTime, + http_addr: String, + queue_alive: bool, + wechat_accounts: HashMap, + last_error: Option, +} + +pub struct RuntimeState { + inner: Mutex, +} + +impl RuntimeState { + pub fn new(http_addr: String) -> Self { + let now = Utc::now(); + Self { + inner: Mutex::new(RuntimeInner { + phase: ServicePhase::Starting, + started_at: now, + updated_at: now, + http_addr, + queue_alive: false, + wechat_accounts: HashMap::new(), + last_error: None, + }), + } + } + + pub async fn mark_running(&self) { + let mut inner = self.inner.lock().await; + inner.phase = ServicePhase::Running; + inner.updated_at = Utc::now(); + inner.last_error = None; + } + + pub async fn mark_stopping(&self) { + let mut inner = self.inner.lock().await; + inner.phase = ServicePhase::Stopping; + inner.queue_alive = false; + inner.updated_at = Utc::now(); + } + + pub async fn mark_stopped(&self) { + let mut inner = self.inner.lock().await; + inner.phase = ServicePhase::Stopped; + inner.queue_alive = false; + inner.updated_at = Utc::now(); + } + + pub async fn mark_failed(&self, error: impl Into) { + let mut inner = self.inner.lock().await; + inner.phase = ServicePhase::Failed; + inner.queue_alive = false; + inner.updated_at = Utc::now(); + inner.last_error = Some(error.into()); + } + + pub async fn set_queue_alive(&self, alive: bool) { + let mut inner = self.inner.lock().await; + inner.queue_alive = alive; + inner.updated_at = Utc::now(); + } + + pub async fn replace_wechat_accounts(&self, account_ids: Vec) { + let mut inner = self.inner.lock().await; + let now = Utc::now(); + inner.wechat_accounts = account_ids + .into_iter() + .map(|account_id| { + let status = AccountRuntimeStatus { + account_id: account_id.clone(), + phase: AccountRuntimePhase::Running, + last_error: None, + last_event_at: Some(now), + }; + (account_id, status) + }) + .collect(); + inner.updated_at = now; + } + + pub async fn record_wechat_event(&self, account_id: impl Into) { + self.update_wechat_account(account_id.into(), AccountRuntimePhase::Running, None) + .await; + } + + pub async fn record_wechat_error( + &self, + account_id: impl Into, + error: impl Into, + ) { + self.update_wechat_account( + account_id.into(), + AccountRuntimePhase::Retrying, + Some(error.into()), + ) + .await; + } + + async fn update_wechat_account( + &self, + account_id: String, + phase: AccountRuntimePhase, + last_error: Option, + ) { + let mut inner = self.inner.lock().await; + let now = Utc::now(); + let status = inner + .wechat_accounts + .entry(account_id.clone()) + .or_insert_with(|| AccountRuntimeStatus { + account_id, + phase: AccountRuntimePhase::Running, + last_error: None, + last_event_at: None, + }); + status.phase = phase; + status.last_error = last_error; + status.last_event_at = Some(now); + inner.updated_at = now; + } + + pub async fn snapshot(&self) -> RuntimeSnapshot { + let inner = self.inner.lock().await; + let mut wechat_accounts = inner.wechat_accounts.values().cloned().collect::>(); + wechat_accounts.sort_by(|left, right| left.account_id.cmp(&right.account_id)); + + RuntimeSnapshot { + phase: inner.phase.clone(), + pid: process::id(), + started_at: inner.started_at, + updated_at: inner.updated_at, + http_addr: inner.http_addr.clone(), + queue_alive: inner.queue_alive, + wechat_accounts, + last_error: inner.last_error.clone(), + } + } + + pub async fn write_snapshot(&self, path: &Path) -> anyhow::Result<()> { + let snapshot = self.snapshot().await; + let data = serde_json::to_vec_pretty(&snapshot)?; + std::fs::write(path, data)?; + Ok(()) + } +} diff --git a/service/src/core/supervisor.rs b/service/src/core/supervisor.rs new file mode 100644 index 0000000..9b72b95 --- /dev/null +++ b/service/src/core/supervisor.rs @@ -0,0 +1,275 @@ +use std::fs::{File, OpenOptions}; +use std::io::{Read, Write}; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use crate::core::control::{ControlResponse, RuntimePaths, request_control}; +use crate::core::runtime::{RuntimeSnapshot, ServicePhase}; + +pub async fn start_service() -> anyhow::Result<()> { + let paths = RuntimePaths::new(); + paths.ensure_dir()?; + + if let Ok(ControlResponse::Status { snapshot }) = request_control(&paths, "status").await { + println!("服务已经在运行，PID: {}", snapshot.pid); + print_snapshot(&snapshot); + return Ok(()); + } + + cleanup_stale_files(&paths)?; + + let mut log = OpenOptions::new() + .create(true) + .append(true) + .open(&paths.log_file)?; + writeln!(log, "\n===== 启动后台服务 =====")?; + + let exe = std::env::current_exe()?; + let stdout = log.try_clone()?; + let stderr = log; + let child = Command::new(exe) + .arg("service") + .arg("run") + .env("IAS_BACKGROUND", "1") + .stdin(Stdio::null()) + .stdout(Stdio::from(stdout)) + .stderr(Stdio::from(stderr)) + .spawn()?; + + println!("后台服务启动中，PID: {}", child.id()); + wait_until_ready(&paths, Duration::from_secs(8)).await?; + println!("日志文件: {}", paths.log_file.display()); + Ok(()) +} + +pub async fn stop_service() -> anyhow::Result<()> { + let paths = RuntimePaths::new(); + paths.ensure_dir()?; + + match request_control(&paths, "shutdown").await { + Ok(ControlResponse::Ack { message }) => println!("{message}"), + Ok(ControlResponse::Error { message }) => println!("停止失败: {message}"), + Ok(ControlResponse::Status { .. }) => println!("停止失败: 控制端返回了意外状态响应"), + Err(_) => { + if let Some(pid) = read_pid(&paths)? { + if process_is_alive(pid) { + println!("控制 socket 不可用，发送 SIGTERM 到 PID: {pid}"); + send_sigterm(pid)?; + } else { + println!("服务未运行，清理过期运行文件"); + cleanup_stale_files(&paths)?; + return Ok(()); + } + } else { + println!("服务未运行"); + return Ok(()); + } + } + } + + wait_until_stopped(&paths, Duration::from_secs(10)).await?; + cleanup_stale_files(&paths)?; + println!("服务已停止"); + Ok(()) +} + +pub async fn restart_service() -> anyhow::Result<()> { + stop_service().await?; + start_service().await +} + +pub async fn status_service() -> anyhow::Result<()> { + let paths = RuntimePaths::new(); + paths.ensure_dir()?; + + match request_control(&paths, "status").await { + Ok(ControlResponse::Status { snapshot }) => print_snapshot(&snapshot), + Ok(ControlResponse::Error { message }) => println!("状态查询失败: {message}"), + Ok(ControlResponse::Ack { message }) => println!("{message}"), + Err(_) => print_offline_status(&paths)?, + } + + Ok(()) +} + +pub fn print_logs(lines: usize) -> anyhow::Result<()> { + let paths = RuntimePaths::new(); + if !paths.log_file.exists() { + println!("日志文件不存在: {}", paths.log_file.display()); + return Ok(()); + } + + let mut file = File::open(&paths.log_file)?; + let mut content = String::new(); + file.read_to_string(&mut content)?; + let mut tail = content.lines().rev().take(lines).collect::>(); + tail.reverse(); + for line in tail { + println!("{line}"); + } + Ok(()) +} + +async fn wait_until_ready(paths: &RuntimePaths, timeout: Duration) -> anyhow::Result<()> { + let started = Instant::now(); + loop { + if let Ok(ControlResponse::Status { snapshot }) = request_control(paths, "status").await { + println!("服务已启动"); + print_snapshot(&snapshot); + return Ok(()); + } + + if started.elapsed() >= timeout { + anyhow::bail!( + "服务未在 {} 秒内就绪，请查看日志: {}", + timeout.as_secs(), + paths.log_file.display() + ); + } + + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +async fn wait_until_stopped(paths: &RuntimePaths, timeout: Duration) -> anyhow::Result<()> { + let started = Instant::now(); + loop { + if request_control(paths, "status").await.is_err() + && read_pid(paths)?.is_none_or(|pid| !process_is_alive(pid)) + { + return Ok(()); + } + + if started.elapsed() >= timeout { + anyhow::bail!( + "服务未在 {} 秒内停止，请检查 PID 文件: {}", + timeout.as_secs(), + paths.pid_file.display() + ); + } + + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +pub fn write_pid_file(paths: &RuntimePaths) -> anyhow::Result<()> { + paths.ensure_dir()?; + std::fs::write(&paths.pid_file, std::process::id().to_string())?; + Ok(()) +} + +pub fn cleanup_runtime_files(paths: &RuntimePaths) { + if let Some(pid) = read_pid(paths).ok().flatten() + && pid != std::process::id() + { + return; + } + + let _ = std::fs::remove_file(&paths.pid_file); + let _ = std::fs::remove_file(&paths.socket_file); +} + +fn cleanup_stale_files(paths: &RuntimePaths) -> anyhow::Result<()> { + if let Some(pid) = read_pid(paths)? + && process_is_alive(pid) + { + anyhow::bail!( + "已有服务进程存在但控制 socket 不可用，PID: {}。请先执行 stop 或检查日志: {}", + pid, + paths.log_file.display() + ); + } + + let _ = std::fs::remove_file(&paths.pid_file); + let _ = std::fs::remove_file(&paths.socket_file); + Ok(()) +} + +fn print_offline_status(paths: &RuntimePaths) -> anyhow::Result<()> { + if let Some(pid) = read_pid(paths)? + && process_is_alive(pid) + { + println!("服务进程存在但控制 socket 不可用，PID: {pid}"); + println!("socket: {}", paths.socket_file.display()); + println!("日志文件: {}", paths.log_file.display()); + return Ok(()); + } + + println!("服务状态: stopped"); + if paths.state_file.exists() { + println!("最后状态快照: {}", paths.state_file.display()); + } + println!("日志文件: {}", paths.log_file.display()); + Ok(()) +} + +fn print_snapshot(snapshot: &RuntimeSnapshot) { + println!("服务状态: {}", phase_name(&snapshot.phase)); + println!("PID: {}", snapshot.pid); + println!("HTTP: {}", snapshot.http_addr); + println!("启动时间: {}", snapshot.started_at); + println!( + "队列: {}", + if snapshot.queue_alive { + "running" + } else { + "stopped" + } + ); + + if let Some(error) = &snapshot.last_error { + println!("最近错误: {error}"); + } + + if snapshot.wechat_accounts.is_empty() { + println!("微信账号: 0"); + } else { + println!("微信账号:"); + for account in &snapshot.wechat_accounts { + let last_event = account + .last_event_at + .map(|time| time.to_string()) + .unwrap_or_else(|| "-".to_string()); + let last_error = account.last_error.as_deref().unwrap_or("-"); + println!( + " - {}: {:?}, last_event={}, last_error={}", + account.account_id, account.phase, last_event, last_error + ); + } + } +} + +fn phase_name(phase: &ServicePhase) -> &'static str { + match phase { + ServicePhase::Starting => "starting", + ServicePhase::Running => "running", + ServicePhase::Stopping => "stopping", + ServicePhase::Stopped => "stopped", + ServicePhase::Failed => "failed", + } +} + +fn read_pid(paths: &RuntimePaths) -> anyhow::Result> { + if !paths.pid_file.exists() { + return Ok(None); + } + + let content = std::fs::read_to_string(&paths.pid_file)?; + let pid = content.trim().parse::()?; + Ok(Some(pid)) +} + +fn process_is_alive(pid: u32) -> bool { + std::path::Path::new("/proc").join(pid.to_string()).exists() +} + +fn send_sigterm(pid: u32) -> anyhow::Result<()> { + let status = Command::new("kill") + .arg("-TERM") + .arg(pid.to_string()) + .status()?; + if !status.success() { + anyhow::bail!("发送 SIGTERM 失败，PID: {pid}"); + } + Ok(()) +} diff --git a/service/src/http/app.rs b/service/src/http/app.rs index 8145099..2d5ae4b 100644 --- a/service/src/http/app.rs +++ b/service/src/http/app.rs @@ -1,20 +1,44 @@ use std::sync::Arc; -use anyhow::Ok; -use axum::routing::{get}; +use anyhow::anyhow; +use axum::routing::get; use axum::{Json, Router}; +use tokio::sync::watch; use crate::AppState; +use crate::web::routes::routes as web_routes; // use crate::core::queue::send_to_queue; -pub async fn create_http(state: AppState, host: String) -> anyhow::Result<()> { +pub async fn create_http( + state: AppState, + host: String, + mut shutdown_rx: watch::Receiver, +) -> anyhow::Result<()> { let app = Router::new() // .route("/message", post(send_to_queue)) .route("/health", get(health)) + .merge(web_routes()) .with_state(Arc::new(state)); - let listener = tokio::net::TcpListener::bind(host.clone()).await?; + let listener = tokio::net::TcpListener::bind(&host).await.map_err(|err| { + if err.kind() == std::io::ErrorKind::AddrInUse { + anyhow!( + "HTTP 地址已被占用: {}。请停止已有服务，或修改 HOST 后重试", + host + ) + } else { + anyhow!("HTTP 地址监听失败: {}: {}", host, err) + } + })?; println!("服务启动在：http://{}", listener.local_addr()?); - axum::serve(listener, app).await?; + axum::serve(listener, app) + .with_graceful_shutdown(async move { + while shutdown_rx.changed().await.is_ok() { + if *shutdown_rx.borrow() { + break; + } + } + }) + .await?; Ok(()) } diff --git a/service/src/http/mod.rs b/service/src/http/mod.rs index 02c0277..309be62 100644 --- a/service/src/http/mod.rs +++ b/service/src/http/mod.rs @@ -1 +1 @@ -pub mod app; \ No newline at end of file +pub mod app; diff --git a/service/src/main.rs b/service/src/main.rs index 09e78ba..b736504 100644 --- a/service/src/main.rs +++ b/service/src/main.rs @@ -3,71 +3,175 @@ mod core; mod http; mod model; mod repository; +mod web; use clap::Parser; use common::queue::QueueMessage; +use core::control::{ControlResponse, RuntimePaths, request_control, start_control_server}; use core::queue::create_queue; +use core::runtime::RuntimeState; +use core::supervisor::{ + cleanup_runtime_files, print_logs, restart_service, start_service, status_service, + stop_service, write_pid_file, +}; use http::app::create_http; use sqlx::{PgPool, postgres::PgPoolOptions}; use std::env; +use std::sync::Arc; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, watch}; use crate::channel::wechat::manager::{ delete_account, init_wechat, list_accounts, login_user, rename_account, }; -use crate::core::cli::{Channels, Cli, Commands, WechatAction}; +use crate::core::cli::{Channels, Cli, Commands, ServiceAction, WechatAction}; #[derive(Clone)] pub struct AppState { pub sender: mpsc::Sender, pub db: PgPool, + pub runtime: Arc, } #[tokio::main] async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - let db_url = env::var("DATABASE_URL").expect("必须设置DATABASE_URL"); - let pool = PgPoolOptions::new() - .max_connections(10) - .connect(&db_url) - .await?; + init_logging(); let cli = Cli::parse(); match cli.command { - Commands::Service => { - service(pool).await.unwrap(); - } - Commands::Channel { command } => match command { - Channels::Wechat { command } => match command { - WechatAction::Login => login_user(&pool).await, - WechatAction::List => list_accounts(&pool).await, - WechatAction::Delete { account } => delete_account(&pool, account).await, - WechatAction::Rename { account, name } => { - rename_account(&pool, account, name).await + Commands::Service { command } => { + let action = command.unwrap_or(ServiceAction::Run); + match action { + ServiceAction::Run => { + let pool = connect_db().await?; + run_service(pool).await?; } - }, - }, + ServiceAction::Start => start_service().await?, + ServiceAction::Stop => stop_service().await?, + ServiceAction::Restart => restart_service().await?, + ServiceAction::Status => status_service().await?, + ServiceAction::Logs { lines } => print_logs(lines)?, + } + } + Commands::Channel { command } => { + let pool = connect_db().await?; + match command { + Channels::Wechat { command } => match command { + WechatAction::Login => login_user(&pool).await, + WechatAction::List => list_accounts(&pool).await, + WechatAction::Delete { account } => delete_account(&pool, account).await, + WechatAction::Rename { account, name } => { + rename_account(&pool, account, name).await + } + }, + } + } } Ok(()) } -async fn service(pool: PgPool) -> anyhow::Result<()> { +async fn connect_db() -> anyhow::Result { + let db_url = env::var("DATABASE_URL").map_err(|_| anyhow::anyhow!("必须设置DATABASE_URL"))?; + let pool = PgPoolOptions::new() + .max_connections(10) + .connect(&db_url) + .await?; + Ok(pool) +} + +fn init_logging() { + let filter = tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")); + let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init(); +} + +async fn run_service(pool: PgPool) -> anyhow::Result<()> { let name: String = env::var("NAME").unwrap_or_else(|_| "ias".to_string()); let host = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9003".to_string()); + let paths = RuntimePaths::new(); + paths.ensure_dir()?; + if let Ok(ControlResponse::Status { snapshot }) = request_control(&paths, "status").await { + anyhow::bail!("服务已在运行，PID: {}", snapshot.pid); + } println!("{}已启动...", name); + let runtime = Arc::new(RuntimeState::new(host.clone())); let (sender, receiver) = mpsc::channel::(100); let state = AppState { sender: sender.clone(), db: pool, + runtime: runtime.clone(), }; - let manger = init_wechat(Some(state.clone())).await.unwrap(); - create_queue(sender.clone(), receiver, manger) + let manger = init_wechat(Some(state.clone())) .await - .unwrap(); + .ok_or_else(|| anyhow::anyhow!("微信初始化失败：未加载到微信账号"))?; + let queue_task = create_queue(sender.clone(), receiver, manger).await?; + runtime.set_queue_alive(true).await; - // 启动http服务器，务必放在最下面执行 - create_http(state.clone(), host.clone()).await.unwrap(); - Ok(()) + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let mut control_task = tokio::spawn(start_control_server( + runtime.clone(), + paths.clone(), + shutdown_tx.clone(), + shutdown_tx.subscribe(), + )); + tokio::select! { + result = &mut control_task => { + result??; + anyhow::bail!("服务控制通道已退出"); + } + _ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {} + } + write_pid_file(&paths)?; + let signal_task = tokio::spawn(wait_for_shutdown_signal(shutdown_tx.clone())); + + runtime.mark_running().await; + runtime.write_snapshot(&paths.state_file).await?; + + let result = create_http(state.clone(), host.clone(), shutdown_rx).await; + runtime.mark_stopping().await; + runtime.write_snapshot(&paths.state_file).await?; + + queue_task.abort(); + control_task.abort(); + signal_task.abort(); + + if let Err(err) = &result { + runtime.mark_failed(err.to_string()).await; + runtime.write_snapshot(&paths.state_file).await?; + } else { + runtime.mark_stopped().await; + runtime.write_snapshot(&paths.state_file).await?; + } + + cleanup_runtime_files(&paths); + result +} + +async fn wait_for_shutdown_signal(shutdown_tx: watch::Sender) { + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + + let mut terminate = match signal(SignalKind::terminate()) { + Ok(signal) => signal, + Err(err) => { + eprintln!("监听 SIGTERM 失败: {err}"); + return; + } + }; + + tokio::select! { + _ = tokio::signal::ctrl_c() => {} + _ = terminate.recv() => {} + } + } + + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + } + + let _ = shutdown_tx.send(true); } diff --git a/service/src/web/mod.rs b/service/src/web/mod.rs new file mode 100644 index 0000000..f230dcc --- /dev/null +++ b/service/src/web/mod.rs @@ -0,0 +1,3 @@ +pub mod model; +pub mod repository; +pub mod routes; diff --git a/service/src/web/model.rs b/service/src/web/model.rs new file mode 100644 index 0000000..a4bb708 --- /dev/null +++ b/service/src/web/model.rs @@ -0,0 +1,59 @@ +use chrono::NaiveDateTime; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sqlx::prelude::FromRow; + +#[derive(Debug, Clone, Serialize, FromRow)] +pub struct WebPage { + pub id: i64, + pub slug: String, + pub page_type: String, + pub title: String, + pub summary: Option, + pub content: String, + pub source_label: Option, + pub metadata: String, + pub created_at: Option, +} + +#[derive(Debug, Clone)] +pub struct NewWebPage { + pub slug: String, + pub page_type: String, + pub title: String, + pub summary: Option, + pub content: String, + pub source_label: Option, + pub metadata: String, +} + +#[derive(Debug, Deserialize)] +pub struct CreateWebPageRequest { + pub page_type: Option, + pub title: String, + pub summary: Option, + pub content: String, + pub source_label: Option, + pub metadata: Option, +} + +#[derive(Debug, Deserialize)] +pub struct CreateMailPageRequest { + pub from: String, + pub subject: String, + pub reminder: Option, + pub content: String, + pub received_at: Option, +} + +#[derive(Debug, Serialize)] +pub struct CreateWebPageResponse { + pub success: bool, + pub id: i64, + pub slug: String, + pub page_type: String, + pub title: String, + pub summary: Option, + pub path: String, + pub url: String, +} diff --git a/service/src/web/repository.rs b/service/src/web/repository.rs new file mode 100644 index 0000000..65d50ec --- /dev/null +++ b/service/src/web/repository.rs @@ -0,0 +1,56 @@ +use sqlx::PgPool; + +use crate::web::model::{NewWebPage, WebPage}; + +pub struct WebPageRepository; + +impl WebPageRepository { + pub async fn create(pool: &PgPool, page: NewWebPage) -> anyhow::Result { + let row = sqlx::query_as::<_, WebPage>( + r#" + INSERT INTO ias_web_pages ( + slug, + page_type, + title, + summary, + content, + source_label, + metadata + ) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING id, slug, page_type, title, summary, content, source_label, metadata, created_at + "#, + ) + .bind(page.slug) + .bind(page.page_type) + .bind(page.title) + .bind(page.summary) + .bind(page.content) + .bind(page.source_label) + .bind(page.metadata) + .fetch_one(pool) + .await?; + + Ok(row) + } + + pub async fn find_by_type_and_slug( + pool: &PgPool, + page_type: &str, + slug: &str, + ) -> anyhow::Result> { + let row = sqlx::query_as::<_, WebPage>( + r#" + SELECT id, slug, page_type, title, summary, content, source_label, metadata, created_at + FROM ias_web_pages + WHERE page_type = $1 AND slug = $2 + "#, + ) + .bind(page_type) + .bind(slug) + .fetch_optional(pool) + .await?; + + Ok(row) + } +} diff --git a/service/src/web/routes.rs b/service/src/web/routes.rs new file mode 100644 index 0000000..6f59e14 --- /dev/null +++ b/service/src/web/routes.rs @@ -0,0 +1,479 @@ +use std::sync::Arc; + +use axum::extract::{Path, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::Html; +use axum::routing::{get, post}; +use axum::{Json, Router}; +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::AppState; +use crate::web::model::{ + CreateMailPageRequest, CreateWebPageRequest, CreateWebPageResponse, NewWebPage, WebPage, +}; +use crate::web::repository::WebPageRepository; + +type ApiError = (StatusCode, Json); +type HtmlError = (StatusCode, Html); + +pub fn routes() -> Router> { + Router::new() + .route("/web/pages", post(create_page)) + .route("/web/mail", post(create_mail_page)) + .route("/mail/{slug}", get(show_mail_page)) + .route("/web/{page_type}/{slug}", get(show_typed_page)) +} + +async fn create_page( + State(state): State>, + headers: HeaderMap, + Json(request): Json, +) -> Result, ApiError> { + let page_type = normalize_key(request.page_type.as_deref().unwrap_or("page"), "page")?; + let title = required_text(&request.title, "title")?; + let content = required_text(&request.content, "content")?; + let metadata = request.metadata.unwrap_or_else(|| json!({})).to_string(); + + let page = NewWebPage { + slug: generate_slug(), + page_type, + title, + summary: optional_text(request.summary), + content, + source_label: optional_text(request.source_label), + metadata, + }; + + let page = WebPageRepository::create(&state.db, page) + .await + .map_err(api_internal_error)?; + + Ok(Json(response_for_page(&page, &headers))) +} + +async fn create_mail_page( + State(state): State>, + headers: HeaderMap, + Json(request): Json, +) -> Result, ApiError> { + let from = required_text(&request.from, "from")?; + let subject = required_text(&request.subject, "subject")?; + let content = required_text(&request.content, "content")?; + let reminder = optional_text(request.reminder); + let received_at = optional_text(request.received_at); + let summary = reminder + .clone() + .unwrap_or_else(|| format!("您收到了一份来自 {from} 的邮件提醒")); + + let metadata = json!({ + "from": from, + "subject": subject, + "reminder": reminder, + "received_at": received_at, + }) + .to_string(); + + let page = NewWebPage { + slug: generate_slug(), + page_type: "mail".to_string(), + title: subject, + summary: Some(summary), + content, + source_label: Some(from), + metadata, + }; + + let page = WebPageRepository::create(&state.db, page) + .await + .map_err(api_internal_error)?; + + Ok(Json(response_for_page(&page, &headers))) +} + +async fn show_mail_page( + State(state): State>, + Path(slug): Path, +) -> Result, HtmlError> { + let page = WebPageRepository::find_by_type_and_slug(&state.db, "mail", &slug) + .await + .map_err(html_internal_error)? + .ok_or_else(html_not_found)?; + + Ok(Html(render_page(&page))) +} + +async fn show_typed_page( + State(state): State>, + Path((page_type, slug)): Path<(String, String)>, +) -> Result, HtmlError> { + let page_type = normalize_key(&page_type, "page").map_err(|_| html_not_found())?; + let page = WebPageRepository::find_by_type_and_slug(&state.db, &page_type, &slug) + .await + .map_err(html_internal_error)? + .ok_or_else(html_not_found)?; + + Ok(Html(render_page(&page))) +} + +fn response_for_page(page: &WebPage, headers: &HeaderMap) -> CreateWebPageResponse { + let path = public_path(page); + let url = format!("{}{}", request_base_url(headers), path); + + CreateWebPageResponse { + success: true, + id: page.id, + slug: page.slug.clone(), + page_type: page.page_type.clone(), + title: page.title.clone(), + summary: page.summary.clone(), + path, + url, + } +} + +fn public_path(page: &WebPage) -> String { + if page.page_type == "mail" { + format!("/mail/{}", page.slug) + } else { + format!("/web/{}/{}", page.page_type, page.slug) + } +} + +fn request_base_url(headers: &HeaderMap) -> String { + if let Ok(base_url) = std::env::var("WEB_BASE_URL") + && !base_url.trim().is_empty() + { + return base_url.trim().trim_end_matches('/').to_string(); + } + + let host = headers + .get("host") + .and_then(|value| value.to_str().ok()) + .filter(|value| !value.trim().is_empty()) + .unwrap_or("127.0.0.1:9003"); + let proto = headers + .get("x-forwarded-proto") + .and_then(|value| value.to_str().ok()) + .filter(|value| !value.trim().is_empty()) + .unwrap_or("http"); + + format!("{}://{}", proto.trim(), host.trim()) +} + +fn generate_slug() -> String { + Uuid::new_v4().simple().to_string() +} + +fn required_text(value: &str, field: &str) -> Result { + let value = value.trim(); + if value.is_empty() { + return Err(api_bad_request(format!("{field} 不能为空"))); + } + Ok(value.to_string()) +} + +fn optional_text(value: Option) -> Option { + value + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +fn normalize_key(value: &str, field: &str) -> Result { + let normalized = value.trim().to_ascii_lowercase(); + if normalized.is_empty() || normalized.len() > 64 { + return Err(api_bad_request(format!("{field} 长度必须在 1 到 64 之间"))); + } + if !normalized + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_') + { + return Err(api_bad_request(format!( + "{field} 只能包含字母、数字、下划线或连字符" + ))); + } + Ok(normalized) +} + +fn render_page(page: &WebPage) -> String { + let summary = page + .summary + .as_deref() + .map(|summary| format!(r#"

{}

"#, escape_html(summary))) + .unwrap_or_default(); + let source = page + .source_label + .as_deref() + .map(|source| { + format!( + r#"来源：{}"#, + escape_html(source) + ) + }) + .unwrap_or_default(); + let created_at = page + .created_at + .map(|time| time.format("%Y-%m-%d %H:%M:%S").to_string()) + .unwrap_or_else(|| "-".to_string()); + let metadata = render_metadata(&page.metadata); + let content = render_text_content(&page.content); + let page_type = escape_html(&page.page_type); + + format!( + r#" + + + + + {title} + + + +
+ {page_type} +

{title}

+ {summary} +
{source}创建时间：{created_at}
+ {metadata} +
{content}
+
+ +"#, + title = escape_html(&page.title), + summary = summary, + source = source, + created_at = escape_html(&created_at), + metadata = metadata, + content = content, + page_type = page_type + ) +} + +fn render_metadata(metadata: &str) -> String { + let Ok(Value::Object(values)) = serde_json::from_str::(metadata) else { + return String::new(); + }; + + let items = values + .into_iter() + .filter_map(|(key, value)| { + let value = metadata_value_to_text(value)?; + Some(format!( + "
{}
{}
", + escape_html(metadata_label(&key)), + escape_html(&value) + )) + }) + .collect::>() + .join(""); + + if items.is_empty() { + String::new() + } else { + format!(r#"
{items}
"#) + } +} + +fn metadata_label(key: &str) -> &str { + match key { + "from" => "发件人", + "subject" => "主题", + "reminder" => "提醒", + "received_at" => "收件时间", + other => other, + } +} + +fn metadata_value_to_text(value: Value) -> Option { + match value { + Value::Null => None, + Value::String(value) => { + let value = value.trim().to_string(); + (!value.is_empty()).then_some(value) + } + Value::Array(values) if values.is_empty() => None, + Value::Object(values) if values.is_empty() => None, + other => Some(other.to_string()), + } +} + +fn render_text_content(content: &str) -> String { + content + .lines() + .map(|line| { + if line.trim().is_empty() { + "
".to_string() + } else { + escape_html(line) + } + }) + .collect::>() + .join("
\n") +} + +fn escape_html(value: &str) -> String { + let mut escaped = String::with_capacity(value.len()); + for ch in value.chars() { + match ch { + '&' => escaped.push_str("&"), + '<' => escaped.push_str("<"), + '>' => escaped.push_str(">"), + '"' => escaped.push_str("""), + '\'' => escaped.push_str("'"), + _ => escaped.push(ch), + } + } + escaped +} + +fn api_bad_request(message: impl Into) -> ApiError { + ( + StatusCode::BAD_REQUEST, + Json(json!({ + "success": false, + "message": message.into(), + })), + ) +} + +fn api_internal_error(error: anyhow::Error) -> ApiError { + eprintln!("创建 web 页面失败: {error}"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ + "success": false, + "message": "创建 web 页面失败", + })), + ) +} + +fn html_not_found() -> HtmlError { + ( + StatusCode::NOT_FOUND, + Html(render_error_page("页面不存在", "没有找到对应的内容页面")), + ) +} + +fn html_internal_error(error: anyhow::Error) -> HtmlError { + eprintln!("读取 web 页面失败: {error}"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Html(render_error_page("页面暂时不可用", "读取页面内容失败")), + ) +} + +fn render_error_page(title: &str, message: &str) -> String { + format!( + r#" + + + + + {title} + + +
+

{title}

+

{message}

+
+ +"#, + title = escape_html(title), + message = escape_html(message) + ) +} diff --git a/tools/anysearch_search.yaml b/tools/anysearch_search.yaml new file mode 100644 index 0000000..775d603 --- /dev/null +++ b/tools/anysearch_search.yaml @@ -0,0 +1,47 @@ +name: anysearch_search +desc: 使用 AnySearch 统一搜索 API 搜索互联网、文档、学术、代码等信息，返回结果列表和搜索元数据。默认匿名访问。 +params: + - name: query + type: string + required: true + description: 搜索关键词或问题 + - name: max_results + type: integer + required: false + default: "10" + description: 返回结果数量，范围 1-20，默认 10 +steps: + - url: https://api.anysearch.com/v1/search + method: POST + headers: + Content-Type: application/json + body: | + { + "query": "${param:query}", + "max_results": ${param:max_results}, + "format": "json" + } + timeout: 30 + var: + - name: code + extract: code + default: "-1" + - name: message + extract: message + default: "" + - name: request_id + extract: request_id + default: "" + - name: results + extract: data.results + default: "[]" + - name: total_results + extract: data.metadata.total_results + default: "0" + - name: search_time_ms + extract: data.metadata.search_time_ms + default: "0" + result: | + AnySearch code=${step0:code}, message=${step0:message}, request_id=${step0:request_id} + total_results=${step0:total_results}, search_time_ms=${step0:search_time_ms} + results=${step0:results} diff --git a/tools/query_weatcher.yaml b/tools/query_weatcher.yaml new file mode 100644 index 0000000..bdb4a86 --- /dev/null +++ b/tools/query_weatcher.yaml @@ -0,0 +1,21 @@ +name: get_weather +desc: 查询指定城市实时天气 +params: + - name: city + type: string + required: true + description: 城市名称 +env: + - WEATHER_API_KEY +steps: + - url: https://api.weather.com/current?q=${param:city}&key=${env:WEATHER_API_KEY} + method: GET + var: + - name: city_name + extract: location.name + - name: temp + extract: current.temp_c + - name: condition + extract: current.condition.text + result: | + ${step0:city_name} 当前温度 ${step0:temp}°C，${step0:condition} diff --git a/wechat/src/client.rs b/wechat/src/client.rs index 5481f33..986ffd6 100644 --- a/wechat/src/client.rs +++ b/wechat/src/client.rs @@ -63,7 +63,7 @@ impl WeChatClient { .unwrap_or_else(|_| "https://ilinkai.weixin.qq.com".to_string()) } /// 版本号编码：0x00MMNNPP - const CLIENT_VERSION: u32 = 0x000100_00; // 1.0.0 + const CLIENT_VERSION: u32 = 0x0001_0000; // 1.0.0 pub fn new(base_url: Option) -> Self { // 优先级: 显式参数 > WEIXIN_BASE_URL 环境变量 > 默认值 @@ -363,14 +363,72 @@ impl WeChatClient { context_token: Option<&str>, ) -> Result { let client_id = uuid::Uuid::new_v4().to_string(); + self.send_text_with_client_id(to, text, context_token, &client_id) + .await + } + /// 发送“输入中”状态，返回用于最终完成消息的 client_id。 + pub async fn send_typing( + &self, + to: &str, + context_token: Option<&str>, + ) -> Result { + let client_id = uuid::Uuid::new_v4().to_string(); + self.send_typing_with_client_id(to, context_token, &client_id) + .await?; + Ok(client_id.to_string()) + } + + /// 用已有 client_id 刷新“输入中”状态。 + pub async fn send_typing_with_client_id( + &self, + to: &str, + context_token: Option<&str>, + client_id: &str, + ) -> Result { + self.send_bot_message( + to, + client_id, + WeixinMessage::STATE_TYPING, + None, + context_token, + ) + .await + } + + /// 使用已有 client_id 完成文本消息。 + pub async fn send_text_with_client_id( + &self, + to: &str, + text: &str, + context_token: Option<&str>, + client_id: &str, + ) -> Result { + self.send_bot_message( + to, + client_id, + WeixinMessage::STATE_FINISH, + Some(vec![MessageItem::text(text)]), + context_token, + ) + .await + } + + async fn send_bot_message( + &self, + to: &str, + client_id: &str, + message_state: i32, + item_list: Option>, + context_token: Option<&str>, + ) -> Result { let msg = WeixinMessage { from_user_id: Some(String::new()), to_user_id: Some(to.to_string()), - client_id: Some(client_id.clone()), + client_id: Some(client_id.to_string()), msg_type: Some(WeixinMessage::TYPE_BOT), - message_state: Some(2), // FINISH - item_list: Some(vec![MessageItem::text(text)]), + message_state: Some(message_state), + item_list, context_token: context_token.map(|s| s.to_string()), ..Default::default() }; @@ -387,7 +445,7 @@ impl WeChatClient { .await .map_err(|e| format!("发送消息网络错误: {}", e))?; - Ok(client_id) + Ok(client_id.to_string()) } /// 设置 auth 状态（从持久化恢复） diff --git a/wechat/src/manager.rs b/wechat/src/manager.rs index c4875d0..fe7c55e 100644 --- a/wechat/src/manager.rs +++ b/wechat/src/manager.rs @@ -47,7 +47,7 @@ pub enum WeChatEvent { /// 收到一条微信消息。 Message { account_id: String, - message: WeixinMessage, + message: Box, updates_buf: String, }, /// 某个账号的轮询发生错误。管理器会继续重试，除非任务被停止。 @@ -145,7 +145,7 @@ impl WeChatMultiAccountManager { for message in resp.msgs { let event = WeChatEvent::Message { account_id: account_id.clone(), - message, + message: Box::new(message), updates_buf: updates_buf.clone(), }; if event_tx.send(event).await.is_err() { @@ -230,6 +230,60 @@ impl WeChatMultiAccountManager { runtime.client.send_text(to, text, context_token).await } + + /// 使用指定账号发送输入中状态，返回后续完成消息复用的 client_id。 + pub async fn send_typing( + &self, + account_id: &str, + to: &str, + context_token: Option<&str>, + ) -> Result { + let runtime = self + .accounts + .get(account_id) + .ok_or_else(|| format!("微信账号未启动: {}", account_id))?; + + runtime.client.send_typing(to, context_token).await + } + + /// 使用指定账号刷新输入中状态。 + pub async fn send_typing_with_client_id( + &self, + account_id: &str, + to: &str, + context_token: Option<&str>, + client_id: &str, + ) -> Result { + let runtime = self + .accounts + .get(account_id) + .ok_or_else(|| format!("微信账号未启动: {}", account_id))?; + + runtime + .client + .send_typing_with_client_id(to, context_token, client_id) + .await + } + + /// 使用指定账号完成已有输入中消息。 + pub async fn send_text_with_client_id( + &self, + account_id: &str, + to: &str, + text: &str, + context_token: Option<&str>, + client_id: &str, + ) -> Result { + let runtime = self + .accounts + .get(account_id) + .ok_or_else(|| format!("微信账号未启动: {}", account_id))?; + + runtime + .client + .send_text_with_client_id(to, text, context_token, client_id) + .await + } } impl Drop for WeChatMultiAccountManager { diff --git a/wechat/src/types.rs b/wechat/src/types.rs index 5a72493..a050f88 100644 --- a/wechat/src/types.rs +++ b/wechat/src/types.rs @@ -30,6 +30,12 @@ impl BaseInfo { } } +impl Default for BaseInfo { + fn default() -> Self { + Self::new() + } +} + // ─── 消息类型 ─── #[derive(Debug, Clone, Serialize, Deserialize)] @@ -232,6 +238,8 @@ impl WeixinMessage { pub const TYPE_NONE: i32 = 0; pub const TYPE_USER: i32 = 1; pub const TYPE_BOT: i32 = 2; + pub const STATE_TYPING: i32 = 1; + pub const STATE_FINISH: i32 = 2; /// 获取文本内容（如果有纯文本 item） pub fn text_content(&self) -> Option<&str> {