diff --git a/.codegraph/.gitignore b/.codegraph/.gitignore new file mode 100644 index 0000000..9de0f16 --- /dev/null +++ b/.codegraph/.gitignore @@ -0,0 +1,16 @@ +# CodeGraph data files +# These are local to each machine and should not be committed + +# Database +*.db +*.db-wal +*.db-shm + +# Cache +cache/ + +# Logs +*.log + +# Hook markers +.dirty diff --git a/.gitignore b/.gitignore index e697f2b..7e2ac07 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,5 @@ */target .env .vscode -.agents \ No newline at end of file +.agents +.codegraph diff --git a/Cargo.lock b/Cargo.lock index df5845d..9c8903d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -39,6 +39,56 @@ dependencies = [ "libc", ] +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + [[package]] name = "anyhow" version = "1.0.102" @@ -211,16 +261,63 @@ dependencies = [ "iana-time-zone", "js-sys", "num-traits", + "serde", "wasm-bindgen", "windows-link", ] +[[package]] +name = "clap" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + [[package]] name = "cmov" version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + [[package]] name = "common" version = "0.1.0" @@ -903,6 +1000,12 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itoa" version = "1.0.18" @@ -1047,6 +1150,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "parking" version = "2.2.1" @@ -1481,6 +1590,8 @@ dependencies = [ "ai", "anyhow", "axum", + "chrono", + "clap", "common", "dotenvy", "reqwest", @@ -1779,6 +1890,12 @@ dependencies = [ "unicode-properties", ] +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "subtle" version = "2.6.1" @@ -2157,6 +2274,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + [[package]] name = "uuid" version = "1.23.4" diff --git a/ai/src/core.rs b/ai/src/core.rs index 7d6ec62..d1430a6 100644 --- a/ai/src/core.rs +++ b/ai/src/core.rs @@ -3,13 +3,12 @@ use common::ai::{ ChatCompletionRequestBody, ChatCompletionRequestExtra, ChatCompletionRequestMessage, ChatCompletionRequestThinking, ChatCompletionResponseBody, }; -use common::queue::QueueMessage; - - +use common::queue::{AssistantMessage, ChannelMessage, QueueMessage}; pub async fn send_message( sender: tokio::sync::mpsc::Sender, messages: Vec, + channel: ChannelMessage, ) -> Result<(), reqwest::Error> { let result = request_llm(messages).await; match result { @@ -18,25 +17,30 @@ pub async fn send_message( for choice in result.unwrap().choices { let messsage_content = choice.message.content; let messsage_reasoning = choice.message.reasoning_content.unwrap_or_default(); - if !messsage_content.trim().is_empty() || !messsage_content.trim().is_empty() { - sender - .send(QueueMessage::Aissitant( - messsage_reasoning, - messsage_content, - )) - .await - .unwrap(); - } + let tool_calls = choice.message.tool_calls.clone(); + + sender + .send(QueueMessage::Aissitant( + channel.clone(), + AssistantMessage { + reasoning: Some(messsage_reasoning), + content: messsage_content, + tool_calls, + }, + )) + .await + .unwrap(); + match choice.message.tool_calls { Some(tool_calls) => { - // eprintln!("调用了{}个工具", tool_calls.len()); - let queue_message = call_tools(tool_calls).await.unwrap(); + 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(); } } None => { - // eprintln!("没有调用工具"); + eprintln!("没有调用工具"); } } } diff --git a/ai/src/toolkit.rs b/ai/src/toolkit.rs index dbc36e3..38647fe 100644 --- a/ai/src/toolkit.rs +++ b/ai/src/toolkit.rs @@ -1,10 +1,14 @@ use chrono::Local; use serde_json::error; -use common::{ai::ChatCompletionResponseToolCall, queue::QueueMessage}; +use common::{ + ai::ChatCompletionToolCall, + queue::{ChannelMessage, QueueMessage}, +}; pub async fn call_tools( - tools: Vec, + channel: ChannelMessage, + tools: Vec, ) -> Result, error::Error> { if let Some(tool) = tools.first() { if let Some(funcation) = tool.function.as_ref() { @@ -12,6 +16,7 @@ pub async fn call_tools( 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), @@ -59,4 +64,4 @@ pub fn get_tool() -> Vec { // } // }), ]; -} \ No newline at end of file +} diff --git a/common/Cargo.toml b/common/Cargo.toml index 3180525..126cec1 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -6,4 +6,4 @@ edition = "2024" [dependencies] # ── 序列化 ── serde = { version = "1.0", features = ["derive"] } # 序列化/反序列化 -serde_json = "1.0" +serde_json = "1.0" \ No newline at end of file diff --git a/common/src/ai.rs b/common/src/ai.rs index a96866b..e5833b0 100644 --- a/common/src/ai.rs +++ b/common/src/ai.rs @@ -23,6 +23,7 @@ pub struct ChatCompletionRequestMessage { pub role: String, pub content: String, pub tool_call_id: Option, + pub tool_calls: Option>, } impl ChatCompletionRequestMessage { pub fn new(role: String, content: String) -> Self { @@ -30,13 +31,28 @@ impl ChatCompletionRequestMessage { role, content, tool_call_id: None, + tool_calls: None, } } + pub fn assistant( + role: String, + content: String, + tool_calls: Option>, + ) -> Self { + Self { + role, + content, + tool_call_id: None, + tool_calls: tool_calls, + } + } + pub fn tool(role: String, content: String, tool_call_id: String) -> Self { Self { role, content, tool_call_id: Some(tool_call_id), + tool_calls: None, } } } @@ -51,7 +67,7 @@ pub struct ChatCompletionResponseBody { model: String, service_tier: String, pub choices: Vec, - pub usage: ChatCompletionResponseUsage, + pub usage: ChatCompletionUsage, } #[allow(dead_code)] @@ -68,40 +84,37 @@ pub struct ChatCompletionResponseMessage { pub role: String, pub content: String, pub reasoning_content: Option, - pub tool_calls: Option>, + pub tool_calls: Option>, } -#[allow(dead_code)] -#[derive(Debug, Deserialize)] -pub struct ChatCompletionResponseToolCall { - pub function: Option, +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ChatCompletionToolCall { + pub function: Option, pub id: String, #[serde(rename = "type")] pub call_type: String, } -#[allow(dead_code)] -#[derive(Debug, Deserialize)] -pub struct ChatCompletionResponseFunction { +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ChatCompletionFunction { pub arguments: String, pub name: String, } -#[allow(dead_code)] #[derive(Debug, Deserialize)] -pub struct ChatCompletionResponseUsage { +pub struct ChatCompletionUsage { pub prompt_tokens: u32, pub completion_tokens: u32, pub total_tokens: u32, - pub prompt_tokens_details: ChatCompletionResponseUsagePrompt, - pub completion_tokens_details: ChatCompletionResponseUsageCompletion, + pub prompt_tokens_details: ChatCompletionUsagePrompt, + pub completion_tokens_details: ChatCompletionUsageCompletion, } #[allow(dead_code)] #[derive(Debug, Deserialize)] -pub struct ChatCompletionResponseUsagePrompt { +pub struct ChatCompletionUsagePrompt { pub cached_tokens: u32, } #[allow(dead_code)] #[derive(Debug, Deserialize)] -pub struct ChatCompletionResponseUsageCompletion { +pub struct ChatCompletionUsageCompletion { pub reasoning_tokens: u32, } diff --git a/common/src/db/user.rs b/common/src/db/user.rs deleted file mode 100644 index 9ada404..0000000 --- a/common/src/db/user.rs +++ /dev/null @@ -1,22 +0,0 @@ -use serde::Serialize; - -#[derive(Debug, Clone, Serialize)] -pub struct WechatUser { - pub id: isize, - pub user_name: String, - pub token: String, - pub account_id: String, - pub base_url: String, - pub user_id: String, - pub user_status: isize, - pub created_at: isize, -} - -#[derive(Debug, Clone, Serialize)] -pub struct WechatUserCreate { - pub user_name: String, - pub token: String, - pub account_id: String, - pub base_url: String, - pub user_id: String, -} diff --git a/common/src/lib.rs b/common/src/lib.rs index 1596c37..fac6dcd 100644 --- a/common/src/lib.rs +++ b/common/src/lib.rs @@ -1,3 +1,2 @@ pub mod ai; -pub mod db; pub mod queue; diff --git a/common/src/queue.rs b/common/src/queue.rs index 09820ee..36ecaa6 100644 --- a/common/src/queue.rs +++ b/common/src/queue.rs @@ -1,14 +1,33 @@ use serde::{Deserialize, Serialize}; +use crate::ai::ChatCompletionToolCall; + #[derive(Debug, Clone, Serialize, Deserialize)] pub enum QueueMessage { - Channel(String, String), - Aissitant(String, String), - Tool(String, String, String), + Channel(ChannelMessage), + Aissitant(ChannelMessage, AssistantMessage), + Tool(ChannelMessage, String, String, String), } -#[derive(Clone, Deserialize, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChannelMessage { pub channel: String, + pub account_id: String, + pub from_user_id: String, pub content: String, + pub context: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AssistantMessage { + pub reasoning: Option, + pub content: String, + pub tool_calls: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ToolMessage { + pub tool_call_id: String, + pub name: String, + pub result: String, } diff --git a/service/Cargo.toml b/service/Cargo.toml index 3bdd5a8..0047665 100644 --- a/service/Cargo.toml +++ b/service/Cargo.toml @@ -46,3 +46,7 @@ sqlx = { version = "0.9.0", features = [ # ── 错误处理 ── anyhow = "1" thiserror = "2" +# ── cli ── +clap = { version = "4.6.1", features = ["derive"] } +# ── 时间处理 ── +chrono = { version = "0.4", features = ["serde"] } # 日期时间 \ No newline at end of file diff --git a/service/migrations/20260601000001_init.sql b/service/migrations/20260601000001_init.sql deleted file mode 100644 index 9e3c22b..0000000 --- a/service/migrations/20260601000001_init.sql +++ /dev/null @@ -1,29 +0,0 @@ --- iAs 初始数据库迁移 --- 认证信息和聊天记录存储 - --- 应用状态表（key-value 存储，用于认证信息等） -CREATE TABLE IF NOT EXISTS app_state ( - key TEXT PRIMARY KEY, - value JSONB NOT NULL, - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - --- 聊天记录表 -CREATE TABLE IF NOT EXISTS chat_records ( - id BIGSERIAL PRIMARY KEY, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - direction TEXT NOT NULL, -- 'inbound' | 'outbound' - user_id TEXT NOT NULL, -- 微信用户 ID - account_id TEXT NOT NULL, -- 机器人账号 ID - text TEXT NOT NULL, -- 消息内容 - source TEXT NOT NULL DEFAULT '', -- 消息来源 - context_token TEXT NOT NULL DEFAULT '', -- 上下文令牌 - message_id TEXT NOT NULL DEFAULT '', - meta JSONB NOT NULL DEFAULT '{}'::JSONB -); - -CREATE INDEX IF NOT EXISTS idx_chat_records_user_created - ON chat_records (user_id, created_at DESC); - -CREATE INDEX IF NOT EXISTS idx_chat_records_account_created - ON chat_records (account_id, created_at DESC); diff --git a/service/migrations/20260601000002_pending_approvals.sql b/service/migrations/20260601000002_pending_approvals.sql deleted file mode 100644 index 6904e3e..0000000 --- a/service/migrations/20260601000002_pending_approvals.sql +++ /dev/null @@ -1,17 +0,0 @@ --- 待审批的工具调用 -CREATE TABLE IF NOT EXISTS pending_approvals ( - id UUID PRIMARY KEY, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - expires_at TIMESTAMPTZ NOT NULL, - user_id TEXT NOT NULL, - skill_name TEXT NOT NULL, - params JSONB NOT NULL, - code_hash TEXT NOT NULL, - attempts_left INTEGER NOT NULL DEFAULT 3, - status TEXT NOT NULL DEFAULT 'pending', - result JSONB, - consumed_at TIMESTAMPTZ -); - -CREATE INDEX IF NOT EXISTS idx_pending_approvals_user_status - ON pending_approvals (user_id, status); diff --git a/service/migrations/20260601000003_scheduled_tasks.sql b/service/migrations/20260601000003_scheduled_tasks.sql deleted file mode 100644 index 7801670..0000000 --- a/service/migrations/20260601000003_scheduled_tasks.sql +++ /dev/null @@ -1,20 +0,0 @@ --- 定时任务表（Phase 6） -CREATE TABLE IF NOT EXISTS scheduled_tasks ( - id UUID PRIMARY KEY, - name TEXT NOT NULL UNIQUE, - description TEXT NOT NULL DEFAULT '', - user_id TEXT NOT NULL, - command TEXT NOT NULL, - shell TEXT NOT NULL DEFAULT '/bin/bash', - cwd TEXT NOT NULL DEFAULT '', - interval_seconds INTEGER NOT NULL, - enabled BOOLEAN NOT NULL DEFAULT true, - next_run_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - last_run_at TIMESTAMPTZ, - last_status TEXT, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - -CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_due - ON scheduled_tasks (enabled, next_run_at); diff --git a/service/migrations/20260601000004_llm_usage.sql b/service/migrations/20260601000004_llm_usage.sql deleted file mode 100644 index 3bb60e8..0000000 --- a/service/migrations/20260601000004_llm_usage.sql +++ /dev/null @@ -1,15 +0,0 @@ --- LLM Token 用量统计 -CREATE TABLE IF NOT EXISTS llm_usage ( - id BIGSERIAL PRIMARY KEY, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - user_id TEXT NOT NULL DEFAULT '', - model TEXT NOT NULL DEFAULT '', - provider TEXT NOT NULL DEFAULT 'deepseek', - prompt_tokens INTEGER NOT NULL DEFAULT 0, - completion_tokens INTEGER NOT NULL DEFAULT 0, - total_tokens INTEGER NOT NULL DEFAULT 0, - cache_hit_tokens INTEGER NOT NULL DEFAULT 0, - cache_miss_tokens INTEGER NOT NULL DEFAULT 0 -); - -CREATE INDEX IF NOT EXISTS idx_llm_usage_created ON llm_usage (created_at DESC); diff --git a/service/migrations/20260601000005_user_memories.sql b/service/migrations/20260601000005_user_memories.sql deleted file mode 100644 index 0195ae6..0000000 --- a/service/migrations/20260601000005_user_memories.sql +++ /dev/null @@ -1,9 +0,0 @@ --- 长期记忆表 -CREATE TABLE IF NOT EXISTS user_memories ( - id SERIAL PRIMARY KEY, - user_id TEXT NOT NULL DEFAULT 'default', - content TEXT NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - -CREATE INDEX IF NOT EXISTS idx_user_memories_user ON user_memories (user_id); diff --git a/service/migrations/20260601000006_session_summaries.sql b/service/migrations/20260601000006_session_summaries.sql deleted file mode 100644 index 13adf1a..0000000 --- a/service/migrations/20260601000006_session_summaries.sql +++ /dev/null @@ -1,12 +0,0 @@ --- 会话摘要持久化 -CREATE TABLE IF NOT EXISTS session_summaries ( - id SERIAL PRIMARY KEY, - user_id TEXT NOT NULL DEFAULT 'default', - summary_text TEXT NOT NULL, - reason TEXT NOT NULL DEFAULT 'overflow', - message_count INTEGER NOT NULL DEFAULT 0, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - -CREATE INDEX IF NOT EXISTS idx_session_summaries_created - ON session_summaries (created_at DESC); diff --git a/service/migrations/20260601000007_fix_llm_usage.sql b/service/migrations/20260601000007_fix_llm_usage.sql deleted file mode 100644 index 76d6857..0000000 --- a/service/migrations/20260601000007_fix_llm_usage.sql +++ /dev/null @@ -1,11 +0,0 @@ --- 修复 llm_usage 表：从 iPet 旧格式迁移到 iAs 新格式 --- 旧表有 kind/entry 字段，新表需要 prompt_tokens/completion_tokens 等 - -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS prompt_tokens INTEGER NOT NULL DEFAULT 0; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS completion_tokens INTEGER NOT NULL DEFAULT 0; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS total_tokens INTEGER NOT NULL DEFAULT 0; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS cache_hit_tokens INTEGER NOT NULL DEFAULT 0; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS cache_miss_tokens INTEGER NOT NULL DEFAULT 0; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS user_id TEXT NOT NULL DEFAULT ''; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS model TEXT NOT NULL DEFAULT ''; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS provider TEXT NOT NULL DEFAULT 'deepseek'; diff --git a/service/migrations/20260601000008_fix_scheduled_tasks_user_id.sql b/service/migrations/20260601000008_fix_scheduled_tasks_user_id.sql deleted file mode 100644 index 526a07f..0000000 --- a/service/migrations/20260601000008_fix_scheduled_tasks_user_id.sql +++ /dev/null @@ -1,3 +0,0 @@ --- scheduled_tasks.user_id 默认值修复 -ALTER TABLE scheduled_tasks ALTER COLUMN user_id SET DEFAULT ''; -ALTER TABLE scheduled_tasks ALTER COLUMN user_id DROP NOT NULL; diff --git a/service/migrations/20260601000009_chat_dedup.sql b/service/migrations/20260601000009_chat_dedup.sql deleted file mode 100644 index e9ece19..0000000 --- a/service/migrations/20260601000009_chat_dedup.sql +++ /dev/null @@ -1,3 +0,0 @@ --- 聊天记录去重索引（普通索引，不用 UNIQUE 避免存量重复数据冲突） -CREATE INDEX IF NOT EXISTS idx_chat_records_message_id - ON chat_records (message_id) WHERE message_id != ''; diff --git a/service/migrations/20260601000010_fix_llm_usage_legacy_columns.sql b/service/migrations/20260601000010_fix_llm_usage_legacy_columns.sql deleted file mode 100644 index de9f930..0000000 --- a/service/migrations/20260601000010_fix_llm_usage_legacy_columns.sql +++ /dev/null @@ -1,4 +0,0 @@ --- 移除 llm_usage 表中旧 iPet 遗留列的 NOT NULL 约束 --- kind/entry 是旧格式字段，新代码不使用它们 -ALTER TABLE llm_usage ALTER COLUMN kind DROP NOT NULL; -ALTER TABLE llm_usage ALTER COLUMN entry DROP NOT NULL; \ No newline at end of file diff --git a/service/migrations/20260601000011_fix_chat_records_context_token.sql b/service/migrations/20260601000011_fix_chat_records_context_token.sql deleted file mode 100644 index d55af5b..0000000 --- a/service/migrations/20260601000011_fix_chat_records_context_token.sql +++ /dev/null @@ -1,26 +0,0 @@ --- 修复 chat_records 表中 iPet 遗留的 context_token_present (boolean) → context_token (text) --- 迁移 0001 创建的是 context_token TEXT，但旧库中列名是 context_token_present BOOLEAN -DO $$ -BEGIN - IF EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_name = 'chat_records' AND column_name = 'context_token_present') THEN - - IF EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_name = 'chat_records' AND column_name = 'context_token') THEN - -- 两列都存在（混合/半迁移库）：回填数据后删除旧列 - UPDATE chat_records - SET context_token = CASE WHEN context_token_present::boolean THEN '1' ELSE '' END - WHERE context_token = ''; - ALTER TABLE chat_records DROP COLUMN context_token_present; - ELSE - -- 只有旧列：类型转换 boolean → text (true → '1', false → '')，再重命名 - ALTER TABLE chat_records - ALTER COLUMN context_token_present TYPE TEXT - USING CASE WHEN context_token_present::boolean THEN '1' ELSE '' END; - ALTER TABLE chat_records RENAME COLUMN context_token_present TO context_token; - ALTER TABLE chat_records ALTER COLUMN context_token SET DEFAULT ''; - END IF; - - END IF; -END; -$$; diff --git a/service/migrations/20260601000012_fix_chat_dedup.sql b/service/migrations/20260601000012_fix_chat_dedup.sql deleted file mode 100644 index 0cee8d1..0000000 --- a/service/migrations/20260601000012_fix_chat_dedup.sql +++ /dev/null @@ -1,12 +0,0 @@ --- 聊天记录去重：清理存量重复 → 建 partial UNIQUE 索引支持 ON CONFLICT --- 1) 删除重复记录（保留每个 message_id 最早的那条） -DELETE FROM chat_records a -USING chat_records b -WHERE a.message_id = b.message_id - AND a.message_id NOT IN ('', '0') - AND a.id > b.id; - --- 2) 替换旧普通索引为 UNIQUE partial index（排除空和占位值） -DROP INDEX IF EXISTS idx_chat_records_message_id; -CREATE UNIQUE INDEX idx_chat_records_message_id - ON chat_records (message_id) WHERE message_id NOT IN ('', '0'); diff --git a/service/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql b/service/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql deleted file mode 100644 index 276885b..0000000 --- a/service/migrations/20260601000013_fix_llm_usage_kind_entry_newdb.sql +++ /dev/null @@ -1,3 +0,0 @@ --- 确保 llm_usage 表在全新数据库上也有 kind/entry 列（迁移 0010 仅解除了旧库的 NOT NULL） -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS kind TEXT; -ALTER TABLE llm_usage ADD COLUMN IF NOT EXISTS entry JSONB; diff --git a/service/migrations/20260601000014_scheduled_tasks_type.sql b/service/migrations/20260601000014_scheduled_tasks_type.sql deleted file mode 100644 index 6ee32af..0000000 --- a/service/migrations/20260601000014_scheduled_tasks_type.sql +++ /dev/null @@ -1,6 +0,0 @@ --- 定时任务增加 task_type 字段：system（系统任务）或 model（模型任务） --- 模型只能管理 model 类型的任务，不能修改 system 任务 -ALTER TABLE scheduled_tasks ADD COLUMN IF NOT EXISTS task_type TEXT NOT NULL DEFAULT 'model'; - -CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_type - ON scheduled_tasks (task_type, enabled, next_run_at); diff --git a/service/migrations/20260601000015_dummy_noop.sql b/service/migrations/20260601000015_dummy_noop.sql deleted file mode 100644 index 9eba19c..0000000 --- a/service/migrations/20260601000015_dummy_noop.sql +++ /dev/null @@ -1,4 +0,0 @@ --- 空迁移：用于对齐 _sqlx_migrations 表中已记录但文件丢失的迁移 --- 该迁移已被之前的某次运行应用过，但对应文件被删除/重命名 --- 此文件仅用于让 sqlx 迁移校验通过，不执行任何 DDL -SELECT 1; diff --git a/service/migrations/20260601000016_session_summaries_user_idx.sql b/service/migrations/20260601000016_session_summaries_user_idx.sql deleted file mode 100644 index a549a43..0000000 --- a/service/migrations/20260601000016_session_summaries_user_idx.sql +++ /dev/null @@ -1,4 +0,0 @@ --- 为"按用户查询最近的会话边界（clear/timeout 摘要）"添加索引。 --- list_recent_chat_records_for_context 会据此过滤掉已过期会话的消息。 -CREATE INDEX IF NOT EXISTS idx_session_summaries_user_created - ON session_summaries (user_id, created_at DESC); diff --git a/service/migrations/20260625031750_create_users.sql b/service/migrations/20260625031750_create_account.sql similarity index 73% rename from service/migrations/20260625031750_create_users.sql rename to service/migrations/20260625031750_create_account.sql index e46f8b5..c8b1702 100644 --- a/service/migrations/20260625031750_create_users.sql +++ b/service/migrations/20260625031750_create_account.sql @@ -1,10 +1,12 @@ -- Add migration script here -CREATE TABLE ias_wechat_users ( +CREATE TABLE ias_wechat_accounts ( id BIGSERIAL PRIMARY KEY, user_name VARCHAR(100) NOT NULL, token VARCHAR(255), account_id VARCHAR(255), base_url VARCHAR(255), + user_id VARCHAR(255), + updates_buf VARCHAR(255), user_status INTEGER, created_at TIMESTAMP NOT NULL DEFAULT NOW() ); \ No newline at end of file diff --git a/service/migrations/20260625035946_wecaht_user_add_id.sql b/service/migrations/20260625035946_wecaht_user_add_id.sql deleted file mode 100644 index a3bee0e..0000000 --- a/service/migrations/20260625035946_wecaht_user_add_id.sql +++ /dev/null @@ -1,2 +0,0 @@ --- Add migration script here -ALTER TABLE ias_wechat_users ADD COLUMN user_id VARCHAR(255); \ No newline at end of file diff --git a/service/src/channel/wechat.rs b/service/src/channel/wechat.rs deleted file mode 100644 index df0de5a..0000000 --- a/service/src/channel/wechat.rs +++ /dev/null @@ -1,63 +0,0 @@ -use std::time::Duration; -use wechat::{client::WeChatClient, types::WeixinMessage}; - -use crate::AppState; - -pub async fn init_wechat(state: Option) -> Result<(), Box> { - match state { - Some(state)=>{ - load_config(state); - } - None=>{} - } - - let client = WeChatClient::new(None); - - - let login = client - .login( - |qrcode_url| { - println!("二维码地址: {qrcode_url}"); - }, - 120, - ) - .await?; - - // println!("登录成功: account_id={}", login.account_id); - - // client.notify_start().await?; - - // loop { - // let updates = client.receive_messages().await?; - - // for msg in updates.msgs { - // if msg.msg_type != Some(WeixinMessage::TYPE_USER) { - // continue; - // } - - // let Some(from_user_id) = msg.from_user_id.as_deref() else { - // continue; - // }; - - // let Some(text) = msg.text_content() else { - // continue; - // }; - - // client - // .send_text( - // from_user_id, - // &format!("收到: {text}"), - // msg.context_token.as_deref(), - // ) - // .await?; - // } - - // tokio::time::sleep(Duration::from_secs(1)).await; - // } - Ok(()) -} - -fn load_config(state: AppState){ - let db = state.db; - -} \ No newline at end of file diff --git a/service/src/channel/wechat/looper.rs b/service/src/channel/wechat/looper.rs new file mode 100644 index 0000000..0d63adc --- /dev/null +++ b/service/src/channel/wechat/looper.rs @@ -0,0 +1,67 @@ +use std::sync::Arc; + +use crate::{AppState, repository::user_repository::UserRepositor}; +use common::queue::{ChannelMessage, QueueMessage}; +use sqlx::PgPool; +use tokio::sync::mpsc::Receiver; +use wechat::{manager::WeChatEvent, types::WeixinMessage}; + +pub async fn start_loop( + state: Arc, + mut event_rx: Receiver, +) -> Result<(), String> { + tokio::spawn(async move { + while let Some(event) = event_rx.recv().await { + match event { + WeChatEvent::UpdatesBuf { + account_id, + updates_buf, + } => { + persist_updates_buf(&state.db, account_id, updates_buf).await; + } + WeChatEvent::Message { + account_id, + message, + updates_buf, + } => { + persist_updates_buf(&state.db, account_id.clone(), updates_buf).await; + + if message.msg_type != Some(WeixinMessage::TYPE_USER) { + continue; + } + + let Some(from_user_id) = message.from_user_id.as_deref() else { + continue; + }; + let Some(text) = message.text_content() else { + continue; + }; + println!("Message account_id:{}=>{}", account_id.clone(), text); + + state + .sender + .clone() + .send(QueueMessage::Channel(ChannelMessage { + channel: "wechat".to_string(), + account_id: account_id, + from_user_id: from_user_id.to_string(), + content: text.to_string(), + context: message.context_token, + })) + .await + .unwrap(); + } + WeChatEvent::PollError { account_id, error } => { + println!("微信账号轮询失败 account_id={account_id}: {error}"); + } + } + } + }); + Ok(()) +} + +async fn persist_updates_buf(pool: &PgPool, account_id: String, buf: String) { + UserRepositor::update_account_buf(pool, account_id, buf) + .await + .unwrap(); +} diff --git a/service/src/channel/wechat/manager.rs b/service/src/channel/wechat/manager.rs new file mode 100644 index 0000000..cddc73c --- /dev/null +++ b/service/src/channel/wechat/manager.rs @@ -0,0 +1,123 @@ +use crate::model::user::{WechatAccount, WechatAccountCreate}; +use sqlx::PgPool; +use std::io; +use std::sync::Arc; +use wechat::client::WeChatClient; +use wechat::manager::{WeChatAccountConfig, WeChatMultiAccountManager}; + +use crate::AppState; +use crate::channel::wechat::looper::start_loop; +use crate::repository::user_repository::UserRepositor; + +pub async fn init_wechat(state: Option) -> Option { + if let Some(state) = state { + println!("init_wechat"); + if let Some(accounts) = load_accounts(&state.db).await { + println!("accounts:{}", accounts.len()); + let (mut manager, event_rx) = WeChatMultiAccountManager::new(1024); + for account in accounts { + println!("account:{}", account.account_id.clone().unwrap_or_default()); + manager + .start_account(WeChatAccountConfig { + token: account.token.unwrap_or_default(), + base_url: account.base_url.unwrap_or_default(), + account_id: account.account_id.unwrap_or_default(), + updates_buf: account.updates_buf.unwrap_or_default(), + }) + .await + .unwrap(); + } + + start_loop(Arc::new(state.clone()), event_rx).await.unwrap(); + return Some(manager); + } + } + None +} + +pub async fn login_user(pool: &PgPool) { + let client = WeChatClient::new(None); + + let login = client + .login( + |qrcode_url| { + println!("二维码地址: {qrcode_url}"); + }, + 120, + ) + .await; + match login { + Ok(account) => { + let account_id = account.account_id; + if let Ok(exist) = UserRepositor::find_wechat_account(pool, account_id.clone()).await { + eprintln!( + "账号已存在:{}:{}", + exist.account_id.unwrap_or_default(), + exist.user_name.unwrap_or_default() + ); + } else { + let mut user_name = String::new(); + + io::stdin().read_line(&mut user_name).unwrap(); + + let user_name = if user_name.trim().is_empty() { + let user_start: String = account_id.clone().chars().take(6).collect(); + format!("user{}", user_start).to_string() + } else { + user_name.trim().to_string() + }; + let user = WechatAccountCreate { + account_id: account_id, + token: account.token, + base_url: account.base_url, + user_id: account.user_id, + user_name: user_name.to_string(), + }; + + // todo 后续需要处理错误 + UserRepositor::add_wechat_account(&pool, user) + .await + .unwrap(); + } + } + Err(err) => { + eprintln!("登录失败，请重试:{}", err) + } + } +} + +pub async fn list_accounts(pool: &PgPool) { + println!("已登录用户如下："); + if let Some(accounts) = load_accounts(pool).await { + for account in accounts { + let user_name = account.user_name.unwrap_or_default(); + let account_id = account.account_id.unwrap_or_default(); + println!(" - {}:{}", user_name, account_id); + } + } +} + +pub async fn rename_account(pool: &PgPool, account: String, name: String) { + if let Ok(lines) = UserRepositor::update_account_name(pool, account, name).await + && lines > 0 + { + println!("修改成功") + } +} +pub async fn delete_account(pool: &PgPool, account: String) { + if let Ok(lines) = UserRepositor::delete_account(pool, account).await + && lines > 0 + { + println!("删除成功") + } +} + +async fn load_accounts(pool: &PgPool) -> Option> { + let accounts = UserRepositor::list_wechat_accounts(pool) + .await + .unwrap_or_default(); + if !accounts.is_empty() { + return Some(accounts); + } + None +} diff --git a/service/src/channel/wechat/mod.rs b/service/src/channel/wechat/mod.rs new file mode 100644 index 0000000..0b945a2 --- /dev/null +++ b/service/src/channel/wechat/mod.rs @@ -0,0 +1,2 @@ +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 new file mode 100644 index 0000000..780a508 --- /dev/null +++ b/service/src/core/cli.rs @@ -0,0 +1,52 @@ +use clap::{Parser, Subcommand}; + +#[derive(Parser)] +#[command(name = "ias")] +#[command(version = "1.0")] +#[command(about = "Rust驱动的智能AI助手。")] +pub struct Cli { + #[command(subcommand)] + pub command: Commands, +} + +#[derive(Debug, Subcommand)] +pub enum Commands { + // 启动服务 + Service, + // 配置渠道 + Channel { + #[command(subcommand)] + command: Channels, + }, +} + +#[derive(Debug, Subcommand)] +pub enum Channels { + // 微信渠道配置 + Wechat { + #[command(subcommand)] + command: WechatAction, + }, +} +#[derive(Debug, Subcommand)] +pub enum WechatAction { + /// 登录新的微信账号 + Login, + /// 查看已经登录的微信账号 + List, + /// 删除已登录的微信账号 + Delete { + /// 要删除的微信账号 + #[arg(short, long)] + account: String, + }, + /// 修改账号名称 - 仅做标记，与微信名无关 + Rename { + /// 要修改的微信账号 + #[arg(short, long)] + account: String, + /// 名称 仅做标记 + #[arg(short, long)] + name: String, + }, +} diff --git a/service/src/core/context.rs b/service/src/core/context.rs index 322bc0e..0fef2b9 100644 --- a/service/src/core/context.rs +++ b/service/src/core/context.rs @@ -2,7 +2,7 @@ use std::sync::OnceLock; use tokio::sync::Mutex; -use common::ai::ChatCompletionRequestMessage; +use common::ai::{ChatCompletionRequestMessage, ChatCompletionToolCall}; static CHAT_CONTEXT: OnceLock>> = OnceLock::new(); @@ -20,8 +20,15 @@ pub async fn snapshot_context() -> Vec { context.clone() } -pub async fn add_assistant_message(content: String) { - let message = ChatCompletionRequestMessage::new("assistant".to_string(), content); +pub async fn add_assistant_message( + content: String, + tool_calls: Option>, +) { + let message = ChatCompletionRequestMessage::assistant( + "assistant".to_string(), + content, + tool_calls, + ); push_context_message(message).await; } @@ -31,6 +38,6 @@ pub async fn add_user_message(content: String) { } pub async fn add_tool_message(call_id: String, result: String) { - let message = ChatCompletionRequestMessage::tool("user".to_string(), result, call_id); + let message = ChatCompletionRequestMessage::tool("tool".to_string(), result, call_id); push_context_message(message).await; } diff --git a/service/src/core/mod.rs b/service/src/core/mod.rs index 7675c2e..4369de2 100644 --- a/service/src/core/mod.rs +++ b/service/src/core/mod.rs @@ -1,2 +1,3 @@ -pub mod queue; -pub mod context; \ No newline at end of file +pub mod cli; +pub mod context; +pub mod queue; \ No newline at end of file diff --git a/service/src/core/queue.rs b/service/src/core/queue.rs index 7769525..3733161 100644 --- a/service/src/core/queue.rs +++ b/service/src/core/queue.rs @@ -1,62 +1,84 @@ -use crate::{ - AppState, - core::context::{add_assistant_message, add_tool_message, add_user_message, snapshot_context}, +use crate::core::context::{ + add_assistant_message, add_tool_message, add_user_message, snapshot_context, }; use ai::core::send_message; -use axum::Json; -use axum::extract::State; -use common::queue::{ChannelMessage, QueueMessage}; +use common::queue::QueueMessage; use serde_json::error; -use std::sync::Arc; -use tokio::sync::{mpsc, mpsc::Sender}; +use tokio::sync::{mpsc::Receiver, mpsc::Sender}; +use wechat::manager::WeChatMultiAccountManager; -pub async fn create_queue() -> Result, error::Error> { - let (sender, mut receiver) = mpsc::channel::(100); - let result_sender = sender.clone(); +pub async fn create_queue( + sender: Sender, + mut receiver: Receiver, + manager: WeChatMultiAccountManager, +) -> Result<(), error::Error> { tokio::spawn(async move { while let Some(msg) = receiver.recv().await { let queue_sender = sender.clone(); match msg { - QueueMessage::Aissitant(reasoning, content) => { + 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); - add_assistant_message(content).await; + if !content.trim().is_empty() || !reasoning.trim().is_empty() { + let text = if !content.trim().is_empty() { + content.trim() + } else { + reasoning.trim() + }; + manager + .send_text( + &channel.account_id, + &channel.from_user_id, + &text, + channel.context.as_deref(), + ) + .await + .unwrap(); + } + add_assistant_message(assistant.content, assistant.tool_calls).await; } - QueueMessage::Channel(name, text) => { - println!("\n\n收到来自{}消息：\n{}", name, text); - add_user_message(text).await; - send_message(queue_sender, snapshot_context().await) + QueueMessage::Channel(bundle) => { + println!( + "\n\n收到来自{}消息：\n{}", + bundle.from_user_id.clone(), + bundle.content.clone() + ); + // todo 目前按单用户处理 + add_user_message(bundle.content.clone()).await; + send_message(queue_sender, snapshot_context().await, bundle) .await .unwrap(); } - QueueMessage::Tool(tool_call_id, name, result) => { + QueueMessage::Tool(channel, tool_call_id, name, result) => { println!("\n\n调用{}工具的结果:\n{}", name, result); add_tool_message(tool_call_id, result).await; - send_message(queue_sender, snapshot_context().await) + send_message(queue_sender, snapshot_context().await, channel) .await .unwrap(); } } } }); - Ok(result_sender) + Ok(()) } -pub async fn send_to_queue( - State(state): State>, - Json(body): Json, -) -> Json { - match state - .sender - .send(QueueMessage::Channel(body.channel, body.content)) - .await - { - Ok(_) => Json(serde_json::json!({ - "success":true, - "message":"queued" - })), - Err(_) => Json(serde_json::json!({ - "success":true, - "message":"queue closed" - })), - } -} +// pub async fn send_to_queue( +// State(state): State>, +// Json(body): Json, +// ) -> Json { +// match state +// .sender +// .send(QueueMessage::Channel(body.channel, body.content)) +// .await +// { +// Ok(_) => Json(serde_json::json!({ +// "success":true, +// "message":"queued" +// })), +// Err(_) => Json(serde_json::json!({ +// "success":true, +// "message":"queue closed" +// })), +// } +// } diff --git a/service/src/error.rs b/service/src/error.rs index 615c689..0ddcc8f 100644 --- a/service/src/error.rs +++ b/service/src/error.rs @@ -5,7 +5,7 @@ use axum::{ }; use serde_json::json; - +#[allow(unused)] #[derive(Debug, thiserror::Error)] pub enum AppError { #[error("not found")] @@ -30,4 +30,5 @@ impl IntoResponse for AppError { } } +#[allow(unused)] pub type AppResult = Result; diff --git a/service/src/http/app.rs b/service/src/http/app.rs index c5c7551..8145099 100644 --- a/service/src/http/app.rs +++ b/service/src/http/app.rs @@ -1,20 +1,21 @@ use std::sync::Arc; -use axum::routing::{get, post}; +use anyhow::Ok; +use axum::routing::{get}; use axum::{Json, Router}; use crate::AppState; -use crate::core::queue::send_to_queue; +// use crate::core::queue::send_to_queue; -pub async fn create_http(state: AppState, host: String) { +pub async fn create_http(state: AppState, host: String) -> anyhow::Result<()> { let app = Router::new() - .route("/message", post(send_to_queue)) + // .route("/message", post(send_to_queue)) .route("/health", get(health)) .with_state(Arc::new(state)); - - let listener = tokio::net::TcpListener::bind(host.clone()).await.unwrap(); - axum::serve(listener, app).await.unwrap(); - println!("服务启动在：{}", host); + let listener = tokio::net::TcpListener::bind(host.clone()).await?; + println!("服务启动在：http://{}", listener.local_addr()?); + axum::serve(listener, app).await?; + Ok(()) } async fn health() -> Json { diff --git a/service/src/main.rs b/service/src/main.rs index dbe0471..af46532 100644 --- a/service/src/main.rs +++ b/service/src/main.rs @@ -2,8 +2,10 @@ mod channel; mod core; mod error; mod http; +mod model; mod repository; +use clap::Parser; use common::queue::QueueMessage; use core::queue::create_queue; use http::app::create_http; @@ -12,7 +14,10 @@ use std::env; use tokio::sync::mpsc; -use crate::channel::wechat::init_wechat; +use crate::channel::wechat::manager::{ + delete_account, init_wechat, list_accounts, login_user, rename_account, +}; +use crate::core::cli::{Channels, Cli, Commands, WechatAction}; #[derive(Clone)] pub struct AppState { @@ -21,35 +26,49 @@ pub struct AppState { } #[tokio::main] -async fn main() -> Result<(), Box> { - if std::env::var("RUST_ANALYZER").is_err() { - dotenv::dotenv().ok(); - } - // dotenvy::dotenv().ok(); - let name = env::var("NAME").unwrap_or_else(|_| "ias".to_string()); - let host = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9003".to_string()); +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?; - - let sender = create_queue().await.unwrap(); - let state = AppState { - sender: sender, - db: pool, - }; - - println!("{} start in : {}", name, host); - create_http(state.clone(), host).await; - init_wechat(Some(state)).await.unwrap(); + 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 + } + }, + }, + } Ok(()) } -#[test] -fn test() { - let api_key = std::env::var("ROUTER_API_KEY").unwrap(); - println!("{}", api_key); +async fn 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()); + + println!("{}已启动...", name); + let (sender, receiver) = mpsc::channel::(100); + let state = AppState { + sender: sender.clone(), + db: pool, + }; + let manger = init_wechat(Some(state.clone())).await.unwrap(); + create_queue(sender.clone(), receiver, manger) + .await + .unwrap(); + + // 启动http服务器，务必放在最下面执行 + create_http(state.clone(), host.clone()).await.unwrap(); + Ok(()) } diff --git a/common/src/db/mod.rs b/service/src/model/mod.rs similarity index 100% rename from common/src/db/mod.rs rename to service/src/model/mod.rs diff --git a/service/src/model/user.rs b/service/src/model/user.rs new file mode 100644 index 0000000..7270100 --- /dev/null +++ b/service/src/model/user.rs @@ -0,0 +1,25 @@ +use chrono::NaiveDateTime; +use serde::Serialize; +use sqlx::prelude::FromRow; + +#[derive(Debug, Clone, Serialize, FromRow)] +pub struct WechatAccount { + pub id: i64, + pub user_name: Option, + pub token: Option, + pub account_id: Option, + pub base_url: Option, + pub user_id: Option, + pub updates_buf: Option, + pub user_status: Option, + pub created_at: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct WechatAccountCreate { + pub user_name: String, + pub token: String, + pub account_id: String, + pub base_url: String, + pub user_id: String, +} diff --git a/service/src/repository/user_repository.rs b/service/src/repository/user_repository.rs index 6215d39..58ac400 100644 --- a/service/src/repository/user_repository.rs +++ b/service/src/repository/user_repository.rs @@ -1,39 +1,108 @@ -use sqlx::PgPool; +use anyhow::Ok; +use sqlx::{PgPool, Row}; -use common::db::user::{WechatUser, WechatUserCreate}; +use crate::model::user::{WechatAccount, WechatAccountCreate}; pub struct UserRepositor; impl UserRepositor { - pub async fn create(pool: &PgPool, user: WechatUserCreate) -> anyhow::Result { - let record = sqlx::query!( + pub async fn add_wechat_account( + pool: &PgPool, + user: WechatAccountCreate, + ) -> anyhow::Result { + let row = sqlx::query( r#" - INSERT INTO ias_wechat_users (user_name,token,account_id,base_url,user_id,user_status) - VALUE ($1,$2,$3,$4,$5,1) + INSERT INTO ias_wechat_accounts (user_name,token,account_id,base_url,user_id,user_status) + VALUES ($1,$2,$3,$4,$5,1) RETURNING id - "#, - user.user_name, - user.token, - user.account_id, - user.base_url, - user.user_id, + "# ) + .bind ( user.user_name) + .bind ( user.token) + .bind ( user.account_id) + .bind ( user.base_url) + .bind ( user.user_id) .fetch_one(pool) .await?; - Ok(record.id) + let id = row.try_get::("id"); + Ok(id.unwrap_or_default()) } - pub async fn find_by_user_id(pool: &PgPool, user_id: isize) -> anyhow::Result { - let user = sqlx::query_as!( - WechatUser, + + pub async fn find_wechat_account( + pool: &PgPool, + account_id: String, + ) -> anyhow::Result { + let row = sqlx::query_as::<_, WechatAccount>( r#" - SELECT * - FROM ias_wechat_users - WHERE user_id = $1 + SELECT * + FROM ias_wechat_accounts + WHERE account_id = $1 + AND user_status = 1 "#, - user_id, ) - .fetch_optional(pool) + .bind(account_id) + .fetch_one(pool) .await?; - Ok(user) + Ok(row) + } + pub async fn list_wechat_accounts(pool: &PgPool) -> anyhow::Result> { + let row = sqlx::query_as::<_, WechatAccount>( + r#" + SELECT * + FROM ias_wechat_accounts + WHERE user_status = 1 + "#, + ) + .fetch_all(pool) + .await?; + Ok(row) + } + pub async fn update_account_name( + pool: &PgPool, + account_id: String, + new_name: String, + ) -> anyhow::Result { + let result = sqlx::query( + r#" + UPDATE ias_wechat_accounts + SET user_name = $1 + WHERE account_id = $2 + "#, + ) + .bind(new_name) + .bind(account_id) + .execute(pool) + .await?; + Ok(result.rows_affected()) + } + pub async fn update_account_buf( + pool: &PgPool, + account_id: String, + buf: String, + ) -> anyhow::Result { + let result = sqlx::query( + r#" + UPDATE ias_wechat_accounts + SET updates_buf = $1 + WHERE account_id = $2 + "#, + ) + .bind(buf) + .bind(account_id) + .execute(pool) + .await?; + Ok(result.rows_affected()) + } + pub async fn delete_account(pool: &PgPool, account_id: String) -> anyhow::Result { + let result = sqlx::query( + r#" + DELETE from ias_wechat_accounts + WHERE account_id = $1 + "#, + ) + .bind(account_id) + .execute(pool) + .await?; + Ok(result.rows_affected()) } } diff --git a/wechat/README.md b/wechat/README.md index f6e4c0e..76a86af 100644 --- a/wechat/README.md +++ b/wechat/README.md @@ -1,8 +1,11 @@ # wechat 模块使用文档 -`wechat` 是一个封装微信 iLink Bot API 的 Rust crate，负责微信 Bot 的扫码登录、监听器生命周期管理、长轮询收消息和文本消息发送。 +`wechat` 是微信 iLink Bot API 的 Rust 封装，提供两层能力： -当前模块只提供库能力，不会自动启动后台任务，也不会自动持久化登录态。业务服务需要主动创建 `WeChatClient`、完成登录或恢复登录态，然后调用收发消息接口。 +- `WeChatClient`：单账号客户端，负责扫码登录、恢复登录态、注册监听器、长轮询收消息、发送文本消息。 +- `WeChatMultiAccountManager`：多账号运行时管理器，按账号启动多个独立长轮询任务，并通过事件队列把消息交给业务层。 + +设计原则：**一个 `WeChatClient` 只代表一个微信 Bot 账号；多账号监听就是每个账号一个独立长轮询任务**。这样可以保证 `token`、`base_url`、`get_updates_buf` 和 `context_token` 不串账号。 ## 模块结构 @@ -10,34 +13,30 @@ wechat/ src/ lib.rs # 模块导出 - client.rs # WeChatClient，封装所有 HTTP API 调用 - types.rs # iLink Bot API 请求、响应和消息结构 + client.rs # 单账号 WeChatClient + manager.rs # 多账号 WeChatMultiAccountManager + types.rs # iLink Bot API 协议类型 ``` 公开模块： ```rust pub mod client; +pub mod manager; pub mod types; ``` -常用类型： +常用导入： ```rust use wechat::client::WeChatClient; -use wechat::types::{GetUpdatesResp, LoginResult, MessageItem, WeixinMessage}; +use wechat::manager::{WeChatAccountConfig, WeChatEvent, WeChatMultiAccountManager}; +use wechat::types::{MessageItem, WeixinMessage}; ``` ## Cargo 引用 -本仓库根目录已经把 `wechat` 加入 workspace： - -```toml -[workspace] -members = ["service", "ai", "common", "wechat"] -``` - -如果其他 workspace crate 需要使用它，在对应 `Cargo.toml` 里加入： +workspace 内其他 crate 使用： ```toml [dependencies] @@ -48,29 +47,52 @@ wechat = { path = "../wechat" } | 变量 | 必填 | 默认值 | 说明 | | --- | --- | --- | --- | -| `WEIXIN_BASE_URL` | 否 | `https://ilinkai.weixin.qq.com` | iLink Bot API 地址。测试、代理或私有部署时可覆盖。 | +| `WEIXIN_BASE_URL` | 否 | `https://ilinkai.weixin.qq.com` | iLink Bot API 地址。显式传入 `base_url` 时优先使用显式值。 | -示例： +## 单账号使用 -```bash -export WEIXIN_BASE_URL="https://ilinkai.weixin.qq.com" +单账号生命周期： + +```text +WeChatClient::new +login 或 set_auth +set_updates_buf +notify_start +receive_messages loop +send_text +notify_stop ``` -## 基本生命周期 +### 扫码登录 -推荐调用顺序： +```rust +use wechat::client::WeChatClient; -1. 创建客户端：`WeChatClient::new(None)` -2. 扫码登录：`login(...)` -3. 持久化登录结果：保存 `token`、`account_id`、`base_url` -4. 注册监听器：`notify_start()` -5. 循环拉取消息：`receive_messages()` -6. 处理用户消息并回复：`send_text(...)` -7. 服务退出前注销监听器：`notify_stop()` +#[tokio::main] +async fn main() -> Result<(), String> { + let client = WeChatClient::new(None); -## 快速开始 + let login = client + .login( + |qrcode_url| println!("二维码地址: {qrcode_url}"), + 120, + ) + .await?; -下面示例会扫码登录，注册监听器，然后持续接收用户文本消息并原样回复。 + println!("account_id={}", login.account_id); + println!("base_url={}", login.base_url); + + // 业务层需要持久化: + // login.token + // login.account_id + // login.base_url + // login.user_id + + Ok(()) +} +``` + +### 恢复登录态并收消息 ```rust use std::time::Duration; @@ -80,23 +102,27 @@ use wechat::types::WeixinMessage; #[tokio::main] async fn main() -> Result<(), String> { - let client = WeChatClient::new(None); + let mut client = WeChatClient::new(None); - let login = client - .login( - |qrcode_url| { - println!("二维码地址: {qrcode_url}"); - }, - 120, + client + .set_auth( + "持久化保存的 token", + "持久化保存的 account_id", + "持久化保存的 base_url", ) - .await?; - - println!("登录成功: account_id={}", login.account_id); + .await; + client + .set_updates_buf("持久化保存的 get_updates_buf") + .await; client.notify_start().await?; loop { let updates = client.receive_messages().await?; + let latest_buf = client.get_updates_buf().await; + + // 建议每次成功 receive 后持久化 latest_buf。 + persist_updates_buf(&latest_buf).await; for msg in updates.msgs { if msg.msg_type != Some(WeixinMessage::TYPE_USER) { @@ -106,323 +132,210 @@ async fn main() -> Result<(), String> { let Some(from_user_id) = msg.from_user_id.as_deref() else { continue; }; - let Some(text) = msg.text_content() else { continue; }; client - .send_text(from_user_id, &format!("收到: {text}"), msg.context_token.as_deref()) + .send_text( + from_user_id, + &format!("收到: {text}"), + msg.context_token.as_deref(), + ) .await?; } tokio::time::sleep(Duration::from_secs(1)).await; } } + +async fn persist_updates_buf(_buf: &str) {} ``` -## 登录 +## 多账号监听 -完整登录接口： +多账号监听使用 `WeChatMultiAccountManager`。它会为每个账号创建一个独立 `WeChatClient` 和一个独立 tokio 任务： + +```text +account A -> WeChatClient A -> getupdates(token A, buf A) +account B -> WeChatClient B -> getupdates(token B, buf B) +account C -> WeChatClient C -> getupdates(token C, buf C) +``` + +这不是短轮询刷接口，而是长轮询。单次 `getupdates` 请求通常会被服务端挂起到有消息或超时，因此多个账号对应多个挂起请求是预期模型。 + +### 账号配置 ```rust -pub async fn login( - &self, - on_qrcode: impl Fn(&str), - timeout_secs: u64, -) -> Result +use wechat::manager::WeChatAccountConfig; + +let account = WeChatAccountConfig::new( + "account_id", + "token", + "base_url", + "get_updates_buf", +); ``` -参数说明： - -| 参数 | 说明 | -| --- | --- | -| `on_qrcode` | 获取二维码内容后的回调。当前实现会传入 `qrcode_img_content`，通常是可打开或可展示为二维码的内容。 | -| `timeout_secs` | 扫码确认总超时时间，单位秒。 | - -返回值： - -```rust -pub struct LoginResult { - pub token: String, - pub account_id: String, - pub base_url: String, - pub user_id: String, -} -``` - -登录内部流程： - -1. `get_qrcode()` 调用 `/ilink/bot/get_bot_qrcode` 获取二维码。 -2. 终端渲染二维码，同时打印二维码链接。 -3. `poll_qr_status()` 轮询 `/ilink/bot/get_qrcode_status`。 -4. 用户扫码并确认后，把 `bot_token` 和 `ilink_bot_id` 写入客户端内存。 -5. 返回 `LoginResult`，供业务层持久化。 - -扫码状态值： - -| 状态 | 含义 | 当前处理 | -| --- | --- | --- | -| `wait` | 等待扫码 | 继续轮询 | -| `scaned` | 已扫码，等待微信确认 | 提示用户确认 | -| `scaned_but_redirect` | 需要切换 IDC | 使用 `redirect_host` 更新轮询地址 | -| `confirmed` | 登录成功 | 保存 token/account_id 并返回 | -| `expired` | 二维码过期 | 返回错误 | - -## 恢复登录态 - -`wechat` 模块不负责把登录态写入数据库或文件。业务层应在登录成功后保存： - -- `LoginResult.token` -- `LoginResult.account_id` -- `LoginResult.base_url` -- `get_updates_buf()` 返回值，建议在每次 `receive_messages()` 成功后保存 - -下次启动时恢复： - -```rust -let mut client = WeChatClient::new(None); - -client - .set_auth( - "持久化保存的 token", - "持久化保存的 account_id", - "持久化保存的 base_url", - ) - .await; - -client - .set_updates_buf("持久化保存的 get_updates_buf") - .await; - -client.notify_start().await?; -``` - -注意：`set_auth` 需要 `&mut self`，因此恢复登录态时变量要声明为 `mut`。 - -## 注册和注销监听器 - -注册监听器： - -```rust -client.notify_start().await?; -``` - -注销监听器： - -```rust -client.notify_stop().await?; -``` - -这两个接口分别调用： - -- `POST /ilink/bot/msg/notifystart` -- `POST /ilink/bot/msg/notifystop` - -建议： - -- 登录或恢复登录态后先调用 `notify_start()`，再开始拉消息。 -- 进程收到退出信号时调用 `notify_stop()`。 -- 如果进程异常退出，下一次启动后仍应重新调用 `notify_start()`。 - -## 接收消息 - -接口： - -```rust -pub async fn receive_messages(&self) -> Result -``` - -响应结构： - -```rust -pub struct GetUpdatesResp { - pub ret: i32, - pub errcode: Option, - pub errmsg: Option, - pub msgs: Vec, - pub get_updates_buf: String, - pub longpolling_timeout_ms: Option, -} -``` - -说明： - -- 该接口调用 `/ilink/bot/getupdates`。 -- 客户端内部维护 `updates_buf`，并在响应里的 `get_updates_buf` 非空时自动更新。 -- 如果请求超时，当前实现返回一个空消息列表，不视为致命错误。 -- 如果 `ret != 0` 或 `errcode != 0`，当前实现会写 warn 日志，但仍把响应返回给调用方。 - -文本消息提取： - -```rust -for msg in updates.msgs { - if msg.msg_type == Some(WeixinMessage::TYPE_USER) { - if let Some(text) = msg.text_content() { - println!("收到文本: {text}"); - } - } -} -``` - -消息关键字段： +字段说明： | 字段 | 说明 | | --- | --- | -| `msg_type` | `1` 表示用户消息，`2` 表示 Bot 消息。 | -| `from_user_id` | 发送者 ID。回复私聊消息时通常作为 `send_text` 的 `to`。 | -| `to_user_id` | 接收者 ID。 | -| `group_id` | 群消息 ID，群聊场景可能非空。 | -| `item_list` | 消息内容列表，可包含文本、图片、语音、文件、视频。 | -| `context_token` | 微信上下文令牌。回复时透传可保持多轮上下文。 | +| `account_id` | 微信 iLink Bot 账号 ID，来自登录返回的 `LoginResult.account_id`。 | +| `token` | 登录返回的 `LoginResult.token`。 | +| `base_url` | 登录返回的 `LoginResult.base_url`。 | +| `updates_buf` | 上次轮询保存的 `get_updates_buf`，新账号可传空字符串。 | -## 发送文本消息 - -接口： +### 启动多个账号 ```rust -pub async fn send_text( - &self, - to: &str, - text: &str, - context_token: Option<&str>, -) -> Result -``` +use wechat::manager::{WeChatAccountConfig, WeChatEvent, WeChatMultiAccountManager}; +use wechat::types::WeixinMessage; -参数说明： +#[tokio::main] +async fn main() -> Result<(), String> { + let (mut manager, mut event_rx) = WeChatMultiAccountManager::new(1024); -| 参数 | 说明 | -| --- | --- | -| `to` | 接收者用户 ID。通常来自收到消息的 `from_user_id`。 | -| `text` | 要发送的文本内容。 | -| `context_token` | 可选上下文令牌，建议透传收到消息里的 `context_token`。 | - -返回值是本地生成的 `client_id`，可用于日志追踪。 - -示例： - -```rust -let client_id = client - .send_text( - from_user_id, - "你好，我是 Bot", - msg.context_token.as_deref(), - ) - .await?; - -println!("消息已提交: client_id={client_id}"); -``` - -当前发送实现固定构造： - -- `message_type = 2`，表示 Bot 消息 -- `message_state = 2`，表示完成态 -- `item_list = [MessageItem::text(text)]` - -## 非文本消息 - -协议类型已经定义了以下消息 item： - -- `TextItem` -- `ImageItem` -- `VoiceItem` -- `FileItem` -- `VideoItem` - -当前便捷发送接口只实现了 `send_text()`。如果需要发送图片、文件或视频，需要新增对应的发送方法，构造 `WeixinMessage.item_list` 中的 `MessageItem`。 - -接收侧可以通过 `MessageItem.item_type` 判断类型： - -```rust -for item in msg.item_list.unwrap_or_default() { - match item.item_type { - Some(MessageItem::TYPE_TEXT) => { - let text = item.text_item.and_then(|v| v.text); - println!("文本: {:?}", text); - } - Some(MessageItem::TYPE_IMAGE) => { - println!("图片: {:?}", item.image_item); - } - Some(MessageItem::TYPE_VOICE) => { - println!("语音: {:?}", item.voice_item); - } - Some(MessageItem::TYPE_FILE) => { - println!("文件: {:?}", item.file_item); - } - Some(MessageItem::TYPE_VIDEO) => { - println!("视频: {:?}", item.video_item); - } - _ => {} + let accounts = load_accounts().await; + for account in accounts { + manager.start_account(account).await?; } -} -``` -## 长轮询循环建议 + while let Some(event) = event_rx.recv().await { + match event { + WeChatEvent::UpdatesBuf { + account_id, + updates_buf, + } => { + persist_updates_buf(&account_id, &updates_buf).await; + } + WeChatEvent::Message { + account_id, + message, + updates_buf, + } => { + persist_updates_buf(&account_id, &updates_buf).await; -推荐业务层把接收循环放到独立 tokio task： - -```rust -let client_for_loop = client.clone(); - -tokio::spawn(async move { - loop { - match client_for_loop.receive_messages().await { - Ok(updates) => { - for msg in updates.msgs { - // 处理消息 + if message.msg_type != Some(WeixinMessage::TYPE_USER) { + continue; } + + let Some(from_user_id) = message.from_user_id.as_deref() else { + continue; + }; + let Some(text) = message.text_content() else { + continue; + }; + + manager + .send_text( + &account_id, + from_user_id, + &format!("账号 {account_id} 收到: {text}"), + message.context_token.as_deref(), + ) + .await?; } - Err(err) => { - tracing::warn!("接收微信消息失败: {err}"); - tokio::time::sleep(std::time::Duration::from_secs(3)).await; + WeChatEvent::PollError { account_id, error } => { + tracing::warn!("微信账号轮询失败 account_id={account_id}: {error}"); } } } -}); -``` - -注意： - -- `WeChatClient` 实现了 `Clone`，内部 token、account_id 和 updates_buf 使用 `Arc>` 共享。 -- `receive_messages()` 已经处理普通超时，业务层不需要把空消息视为错误。 -- 处理消息时建议过滤 `WeixinMessage::TYPE_BOT`，避免重复处理 Bot 自己发出的消息。 -- 每次成功拉取后可以调用 `get_updates_buf()` 并持久化，降低重启后重复收消息的概率。 - -## 与 AI 回复集成示例 - -示例逻辑：收到用户文本后调用业务自己的 AI 函数，再把结果发回微信。 - -```rust -async fn handle_wechat_message(client: &WeChatClient, msg: WeixinMessage) -> Result<(), String> { - if msg.msg_type != Some(WeixinMessage::TYPE_USER) { - return Ok(()); - } - - let Some(from_user_id) = msg.from_user_id.as_deref() else { - return Ok(()); - }; - - let Some(user_text) = msg.text_content() else { - return Ok(()); - }; - - let answer = call_ai(user_text).await?; - - client - .send_text(from_user_id, &answer, msg.context_token.as_deref()) - .await?; Ok(()) } -async fn call_ai(input: &str) -> Result { - Ok(format!("AI 回复: {input}")) +async fn load_accounts() -> Vec { + vec![] +} + +async fn persist_updates_buf(_account_id: &str, _buf: &str) {} +``` + +### 停止账号 + +停止单个账号： + +```rust +manager.stop_account("account_id").await?; +``` + +停止全部账号： + +```rust +let results = manager.stop_all().await; +for (account_id, result) in results { + if let Err(err) = result { + tracing::warn!("停止微信账号失败 account_id={account_id}: {err}"); + } } ``` +`stop_account` 会先中止本地轮询任务，再尽力调用 `notify_stop()`。如果进程直接退出，`Drop` 会 abort 本地任务，但无法异步调用 `notify_stop()`，所以生产环境建议在收到退出信号时显式调用 `stop_all()`。 + +## 为什么不能一个轮询监听多个账号 + +iLink Bot 的 `getupdates` 是按 `Authorization: Bearer ` 鉴权的。一个 token 只对应一个 Bot 账号，服务端不会在同一个请求里返回其他账号的消息。 + +必须隔离的账号状态： + +| 状态 | 为什么必须隔离 | +| --- | --- | +| `token` | 决定当前请求属于哪个账号。 | +| `base_url` | 登录可能返回账号实际使用的 IDC 地址。 | +| `get_updates_buf` | 轮询游标，混用会导致消息重复、丢失或串账号。 | +| `context_token` | 回复上下文，混用会导致回复到错误会话或上下文断裂。 | + +因此多账号正确模型是“多个长轮询任务”，不是“一个共享轮询”。 + +## 持久化建议 + +本 crate 不绑定数据库或文件系统，业务层应自行持久化以下数据。 + +账号表建议字段： + +```text +account_id string primary key +token string +base_url string +user_id string optional +updates_buf string +updated_at timestamp +``` + +收到 `WeChatEvent::Message` 后，使用事件里的 `updates_buf` 更新对应账号的游标。 + +如果要支持主动发送消息，建议额外维护： + +```text +account_id +from_user_id +context_token +updated_at +``` + +收到用户消息时按 `account_id + from_user_id` 保存 `context_token`。主动发送时优先显式传 `account_id`，再取对应用户最近的 `context_token`。 + +## 发送消息路由 + +多账号场景下发送消息必须明确使用哪个账号： + +```rust +manager + .send_text( + "account_id", + "to_user_id", + "回复内容", + Some("context_token"), + ) + .await?; +``` + +不建议只按 `to_user_id` 自动推断账号。原因是同一个用户可能和多个 Bot 账号都有会话，自动推断会产生歧义。 + ## API 端点 -`WeChatClient` 当前使用以下 iLink Bot API： +当前封装使用以下 iLink Bot API： | 方法 | 路径 | 用途 | | --- | --- | --- | @@ -433,25 +346,34 @@ async fn call_ai(input: &str) -> Result { | `POST` | `/ilink/bot/getupdates` | 长轮询拉取消息 | | `POST` | `/ilink/bot/sendmessage` | 发送消息 | -请求头由模块内部生成： +登录后请求会带： -- `Content-Type: application/json` -- `Content-Length` -- `iLink-App-ClientVersion` -- `X-WECHAT-UIN` -- 登录后请求额外带： - - `Authorization: Bearer ` - - `AuthorizationType: ilink_bot_token` +```text +Authorization: Bearer +AuthorizationType: ilink_bot_token +``` + +## 错误处理 + +| 场景 | 当前行为 | 建议 | +| --- | --- | --- | +| 二维码过期 | `login()` 返回 `二维码已过期` | 重新登录。 | +| 登录超时 | `login()` 返回 `登录超时` | 增大超时或重新登录。 | +| 长轮询超时 | `receive_messages()` 返回空消息 | 正常情况，继续轮询。 | +| 游标更新 | manager 发送 `WeChatEvent::UpdatesBuf` | 立即持久化对应账号的 `updates_buf`。 | +| 轮询错误 | manager 发送 `WeChatEvent::PollError` 并继续重试 | 记录日志，必要时触发重新登录。 | +| token 失效 | API 返回错误或 HTTP 鉴权失败 | 清理该账号 token，重新扫码登录。 | +| 多账号重复启动 | `start_account()` 返回错误 | 保证同一 `account_id` 只启动一次。 | ## 日志 -模块使用 `tracing` 输出日志，主要 target 为： +模块使用 `tracing`，主要 target： ```text ias::auth ``` -业务程序可初始化日志： +推荐初始化： ```rust tracing_subscriber::fmt() @@ -459,34 +381,19 @@ tracing_subscriber::fmt() .init(); ``` -## 错误处理 - -所有公开异步方法当前返回 `Result<_, String>`。 - -常见错误： - -| 场景 | 典型错误 | 处理建议 | -| --- | --- | --- | -| 二维码过期 | `二维码已过期` | 重新调用 `login()`。 | -| 登录超时 | `登录超时` | 增大 `timeout_secs` 或重新登录。 | -| token 无效 | HTTP 401/403 或接口返回错误 | 清理持久化登录态，重新扫码登录。 | -| 长轮询超时 | 返回空 `msgs` | 正常情况，继续下一轮。 | -| 网络错误 | `请求失败` | 延迟重试，必要时检查 `WEIXIN_BASE_URL`。 | - ## 当前限制 -- `ILINK_APP_ID` 当前是空字符串。如接入方要求 app id，需要在 `client.rs` 中配置或改造成环境变量。 -- 登录态只保存在内存中，重启恢复需要业务层调用 `set_auth()` 和 `set_updates_buf()`。 -- 只提供文本发送便捷方法 `send_text()`。 -- 群聊回复策略需要业务层结合 `group_id` 和实际 iLink 协议进一步确认。 -- `send_text()` 当前不解析服务端响应体，只在 HTTP 层确认请求成功后返回本地 `client_id`。 +- `send_text()` 只封装文本发送。 +- manager 不内置账号持久化，业务层负责数据库或文件存储。 +- manager 通过 `JoinHandle::abort()` 停止长轮询任务；如需立即优雅取消 HTTP 请求，可后续在 client 层增加 cancellation 支持。 +- `ILINK_APP_ID` 当前为空字符串；如果接入方要求 app id，需要在 `client.rs` 中配置或改为环境变量。 ## 排障清单 -1. 确认能访问 `WEIXIN_BASE_URL`。 -2. 确认扫码后在微信侧完成确认。 -3. 登录后先调用 `notify_start()`，再调用 `receive_messages()`。 -4. 如果收不到消息，检查 `get_updates_buf` 是否恢复了过旧状态；必要时清空后重试。 -5. 如果发送失败，检查 token 是否过期，以及 `to` 是否使用了正确的 `from_user_id`。 -6. 打开 `ias::auth=debug` 日志观察 API 请求和状态变化。 - +1. 确认 `base_url` 可访问。 +2. 确认账号已经扫码登录并保存了 token。 +3. 每个账号只启动一个监听任务。 +4. 每个账号独立保存 `updates_buf`。 +5. 登录或恢复后先调用 `notify_start()`，再开始 `getupdates`。 +6. 发送消息时显式指定 `account_id`，并尽量透传 `context_token`。 +7. 打开 `ias::auth=debug` 查看请求和轮询状态。 diff --git a/wechat/src/lib.rs b/wechat/src/lib.rs index 2e0b64e..d0255dc 100644 --- a/wechat/src/lib.rs +++ b/wechat/src/lib.rs @@ -3,6 +3,7 @@ //! 封装与微信 iLink Bot API 的所有通信： //! //! - `client` — HTTP 客户端（登录、长轮询收消息、发消息、生命周期管理） +//! - `manager` — 多账号监听管理器（每账号一个独立长轮询任务） //! - `types` — API 协议类型定义（消息结构、请求/响应体） //! //! ### 核心流程 @@ -13,4 +14,5 @@ //! 5. `notify_stop()` — 注销监听器 pub mod client; -pub mod types; \ No newline at end of file +pub mod manager; +pub mod types; diff --git a/wechat/src/manager.rs b/wechat/src/manager.rs new file mode 100644 index 0000000..c4875d0 --- /dev/null +++ b/wechat/src/manager.rs @@ -0,0 +1,241 @@ +//! ## 微信多账号监听管理器 +//! +//! `WeChatClient` 表示一个账号的一条 iLink Bot 连接；本模块在其上提供 +//! 多账号运行时管理：每个账号独立 token、base_url、get_updates_buf 和轮询任务。 + +use crate::client::WeChatClient; +use crate::types::WeixinMessage; +use std::collections::HashMap; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tracing::{info, warn}; + +/// 单个微信账号的恢复配置。 +#[derive(Debug, Clone)] +pub struct WeChatAccountConfig { + pub account_id: String, + pub token: String, + pub base_url: String, + pub updates_buf: String, +} + +impl WeChatAccountConfig { + pub fn new( + account_id: impl Into, + token: impl Into, + base_url: impl Into, + updates_buf: impl Into, + ) -> Self { + Self { + account_id: account_id.into(), + token: token.into(), + base_url: base_url.into(), + updates_buf: updates_buf.into(), + } + } +} + +/// 多账号监听事件。 +#[derive(Debug, Clone)] +pub enum WeChatEvent { + /// 账号游标已更新。即使本轮没有消息，业务层也应持久化该值。 + UpdatesBuf { + account_id: String, + updates_buf: String, + }, + /// 收到一条微信消息。 + Message { + account_id: String, + message: WeixinMessage, + updates_buf: String, + }, + /// 某个账号的轮询发生错误。管理器会继续重试，除非任务被停止。 + PollError { account_id: String, error: String }, +} + +struct AccountRuntime { + client: WeChatClient, + task: JoinHandle<()>, +} + +/// 多账号监听管理器。 +/// +/// 设计约束： +/// - 一个 `WeChatClient` 只代表一个账号。 +/// - 一个账号对应一个长轮询任务。 +/// - `get_updates_buf` 由账号独立维护，不能跨账号共享。 +pub struct WeChatMultiAccountManager { + accounts: HashMap, + event_tx: mpsc::Sender, + poll_idle_sleep: Duration, + error_retry_sleep: Duration, +} + +impl WeChatMultiAccountManager { + /// 创建管理器并返回事件接收端。 + pub fn new(event_buffer: usize) -> (Self, mpsc::Receiver) { + let (event_tx, event_rx) = mpsc::channel(event_buffer.max(1)); + ( + Self { + accounts: HashMap::new(), + event_tx, + poll_idle_sleep: Duration::from_secs(1), + error_retry_sleep: Duration::from_secs(3), + }, + event_rx, + ) + } + + /// 设置空轮询后的休眠时间。 + pub fn with_poll_idle_sleep(mut self, duration: Duration) -> Self { + self.poll_idle_sleep = duration; + self + } + + /// 设置轮询错误后的重试间隔。 + pub fn with_error_retry_sleep(mut self, duration: Duration) -> Self { + self.error_retry_sleep = duration; + self + } + + /// 启动一个账号的监听任务。 + /// + /// 如果同名账号已启动，返回错误，避免两个任务竞争同一个账号游标。 + pub async fn start_account(&mut self, config: WeChatAccountConfig) -> Result<(), String> { + if self.accounts.contains_key(&config.account_id) { + return Err(format!("微信账号已启动: {}", config.account_id)); + } + + let mut client = WeChatClient::new(Some(config.base_url.clone())); + client + .set_auth(&config.token, &config.account_id, &config.base_url) + .await; + client.set_updates_buf(&config.updates_buf).await; + client.notify_start().await?; + + let account_id = config.account_id.clone(); + let loop_client = client.clone(); + let event_tx = self.event_tx.clone(); + let poll_idle_sleep = self.poll_idle_sleep; + let error_retry_sleep = self.error_retry_sleep; + + let task = tokio::spawn(async move { + info!(target: "ias::auth", "微信账号监听已启动: {}", account_id); + + loop { + match loop_client.receive_messages().await { + Ok(resp) => { + let updates_buf = loop_client.get_updates_buf().await; + if !updates_buf.is_empty() { + let event = WeChatEvent::UpdatesBuf { + account_id: account_id.clone(), + updates_buf: updates_buf.clone(), + }; + if event_tx.send(event).await.is_err() { + warn!( + target: "ias::auth", + "微信事件接收端已关闭，停止账号监听: {}", + account_id + ); + return; + } + } + + for message in resp.msgs { + let event = WeChatEvent::Message { + account_id: account_id.clone(), + message, + updates_buf: updates_buf.clone(), + }; + if event_tx.send(event).await.is_err() { + warn!( + target: "ias::auth", + "微信事件接收端已关闭，停止账号监听: {}", + account_id + ); + return; + } + } + + tokio::time::sleep(poll_idle_sleep).await; + } + Err(error) => { + let event = WeChatEvent::PollError { + account_id: account_id.clone(), + error, + }; + if event_tx.send(event).await.is_err() { + warn!( + target: "ias::auth", + "微信事件接收端已关闭，停止账号监听: {}", + account_id + ); + return; + } + tokio::time::sleep(error_retry_sleep).await; + } + } + } + }); + + self.accounts + .insert(config.account_id, AccountRuntime { client, task }); + Ok(()) + } + + /// 停止一个账号的监听任务，并尽力调用 notify_stop。 + pub async fn stop_account(&mut self, account_id: &str) -> Result<(), String> { + let Some(runtime) = self.accounts.remove(account_id) else { + return Ok(()); + }; + + runtime.task.abort(); + runtime.client.notify_stop().await + } + + /// 停止所有账号。 + pub async fn stop_all(&mut self) -> Vec<(String, Result<(), String>)> { + let account_ids = self.account_ids(); + let mut results = Vec::with_capacity(account_ids.len()); + for account_id in account_ids { + let result = self.stop_account(&account_id).await; + results.push((account_id, result)); + } + results + } + + /// 返回当前已启动的账号 ID。 + pub fn account_ids(&self) -> Vec { + self.accounts.keys().cloned().collect() + } + + /// 判断账号是否正在监听。 + pub fn is_running(&self, account_id: &str) -> bool { + self.accounts.contains_key(account_id) + } + + /// 使用指定账号发送文本消息。 + pub async fn send_text( + &self, + account_id: &str, + to: &str, + text: &str, + context_token: Option<&str>, + ) -> Result { + let runtime = self + .accounts + .get(account_id) + .ok_or_else(|| format!("微信账号未启动: {}", account_id))?; + + runtime.client.send_text(to, text, context_token).await + } +} + +impl Drop for WeChatMultiAccountManager { + fn drop(&mut self) { + for runtime in self.accounts.values() { + runtime.task.abort(); + } + } +}