From 277bf201530a695886d603bbe0915a7232951fb1 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 14 May 2026 02:10:52 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=87=AD=E8=AF=81?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E4=B8=8E=E6=B8=A0=E9=81=93=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E5=B1=95=E7=A4=BA=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 新增渠道凭证相关API与前端展示逻辑 2. 实现凭证状态自动拉取与手动刷新功能 3. 优化聊天查询支持内部用户ID参数 4. 修复部分代码格式与异常处理逻辑 5. 新增项目代码维基文档 --- CODE_WIKI.md | 892 ++++++++++++++++ .../channel_message_record_repository.py | 12 +- .../yuxi/storage/postgres/models_channels.py | 48 +- backend/server/main.py | 13 +- backend/server/routers/channels_router.py | 957 ++++++++++++++---- backend/server/routers/ws_chat_router.py | 68 +- web/src/apis/channel_api.js | 9 +- .../channels/ChannelMappingBrowser.vue | 12 +- .../components/channels/ChannelStatusCard.vue | 143 ++- web/src/stores/channel.js | 48 + web/src/views/ChannelManageView.vue | 151 ++- 11 files changed, 2082 insertions(+), 271 deletions(-) create mode 100644 CODE_WIKI.md diff --git a/CODE_WIKI.md b/CODE_WIKI.md new file mode 100644 index 00000000..05787746 --- /dev/null +++ b/CODE_WIKI.md @@ -0,0 +1,892 @@ +# Yuxi 项目 Code Wiki + +> 本文档是 Yuxi（语析）项目的结构化代码百科，涵盖项目整体架构、主要模块职责、关键类与函数说明、依赖关系以及项目运行方式等关键信息。 +> +> 项目版本：v0.6.2 | 最后更新：2026-05-14 + +--- + +## 目录 + +1. [项目概述](#1-项目概述) +2. [整体架构](#2-整体架构) +3. [后端核心模块](#3-后端核心模块) +4. [前端核心模块](#4-前端核心模块) +5. [关键类与函数详解](#5-关键类与函数详解) +6. [数据流与运行链路](#6-数据流与运行链路) +7. [依赖关系](#7-依赖关系) +8. [项目运行方式](#8-项目运行方式) +9. [测试体系](#9-测试体系) +10. [附录：目录结构总览](#10-附录目录结构总览) + +--- + +## 1. 项目概述 + +**Yuxi（语析）** 是一个基于大模型的智能知识库与知识图谱智能体开发平台，融合了 RAG 技术与知识图谱技术，基于 **LangGraph v1 + Vue.js + FastAPI + LightRAG** 架构构建。 + +### 核心特性 + +- **智能体开发**：基于 LangGraph，支持子智能体、Skills、MCPs、Tools 与中间件机制 +- **知识库（RAG）**：多格式文档解析，支持 Embedding / Rerank 配置及知识库评估 +- **知识图谱**：基于 LightRAG 的图谱构建与可视化，支持属性图谱并参与智能体推理 +- **平台与工程化**：Vue + FastAPI 架构，支持暗黑模式、Docker 与生产级部署 +- **多渠道网关**：支持 Slack、Discord、Telegram、飞书、微信、QQ 等 20+ 渠道接入 + +### 技术栈 + +| 层级 | 技术 | +|------|------| +| 前端 | Vue 3 + Vite + Pinia + Ant Design Vue + Sigma.js/G6 | +| 后端 API | FastAPI + Uvicorn | +| 智能体框架 | LangGraph + LangChain | +| 任务队列 | ARQ (Redis) | +| 数据库 | PostgreSQL (业务数据) + Neo4j (图谱) + Milvus (向量) | +| 对象存储 | MinIO | +| 缓存/消息 | Redis | +| 文档解析 | MinerU + PaddleX + RapidOCR + DeepSeek OCR | +| 沙盒执行 | Docker Sandbox | + +--- + +## 2. 整体架构 + +### 2.1 架构分层 + +``` +┌─────────────────────────────────────────────────────────────┐ +│ 前端层 (Frontend) │ +│ Vue 3 + Vite + Pinia + Ant Design Vue + Sigma.js/G6 │ +│ web/src/ │ +├─────────────────────────────────────────────────────────────┤ +│ API 网关层 (Gateway) │ +│ FastAPI + 路由注册 + 认证中间件 + CORS + 限流 │ +│ backend/server/ │ +├─────────────────────────────────────────────────────────────┤ +│ 业务服务层 (Services) │ +│ 聊天服务、运行队列、知识库、Skills、MCP、SubAgents │ +│ backend/package/yuxi/services/ │ +├─────────────────────────────────────────────────────────────┤ +│ 智能体层 (Agents) │ +│ BaseAgent + LangGraph + 中间件 + 工具集 + 沙盒后端 │ +│ backend/package/yuxi/agents/ │ +├─────────────────────────────────────────────────────────────┤ +│ 数据访问层 (Repositories) │ +│ SQLAlchemy + 异步会话 + 业务模型 │ +│ backend/package/yuxi/repositories/ │ +├─────────────────────────────────────────────────────────────┤ +│ 基础设施层 (Infrastructure) │ +│ PostgreSQL + Neo4j + Milvus + Redis + MinIO │ +│ backend/package/yuxi/storage/ │ +└─────────────────────────────────────────────────────────────┘ +``` + +### 2.2 Docker Compose 服务拓扑 + +| 服务 | 容器名 | 端口 | 职责 | +|------|--------|------|------| +| api-dev | api-dev | 5050 | FastAPI 主服务（热重载） | +| worker-dev | worker-dev | - | ARQ 异步任务 Worker | +| web-dev | web-dev | 5173 | Vue 前端（热重载） | +| sandbox-provisioner | sandbox-provisioner | 8002 | 沙盒环境供应器 | +| postgres | postgres | 5432 | 业务与知识库元数据 | +| redis | redis | 6379 | 运行事件、队列状态、缓存 | +| minio | minio | 9000/9001 | 对象存储 | +| milvus | milvus | 19530 | 向量检索 | +| graph | graph | 7474/7687 | Neo4j 知识图谱 | +| mineru-vllm-server | mineru-vllm-server | 30000 | 文档解析 VLLM 服务（profile: all） | +| mineru-api | mineru-api | 30001 | 文档解析 API（profile: all） | +| paddlex | paddlex-ocr | 8080 | PaddleX OCR（profile: all） | + +--- + +## 3. 后端核心模块 + +后端分为两个顶层边界：`backend/server` 是 Web 应用入口与 HTTP 适配层，`backend/package/yuxi` 是可复用业务包。 + +### 3.1 入口与路由层 (`backend/server/`) + +#### 3.1.1 主入口 (`server/main.py`) + +- 创建 FastAPI 应用，注册 lifespan 生命周期 +- 挂载所有业务路由到 `/api` 前缀 +- 注册中间件：CORS、访问日志、登录限流、认证 +- 集成 WebSocket 聊天端点和 Slack Webhook + +#### 3.1.2 路由注册 (`server/routers/__init__.py`) + +| 路由模块 | API 前缀 | 职责 | LITE 模式 | +|----------|----------|------|-----------| +| `system_router.py` | `/api/system/*` | 健康检查、全局配置 | ✓ | +| `auth_router.py` | `/api/auth/*` | 登录、用户信息、OIDC | ✓ | +| `chat_router.py` | `/api/chat/*` | 对话、消息流、运行态 | ✓ | +| `dashboard_router.py` | `/api/dashboard/*` | 仪表盘聚合数据 | ✓ | +| `auth_dept_router.py` | `/api/departments/*` | 部门与权限 | ✓ | +| `system_task_router.py` | `/api/tasks/*` | 后台任务管理 | ✓ | +| `mcp_router.py` | `/api/system/mcp-servers/*` | MCP 服务管理 | ✓ | +| `model_provider_router.py` | `/api/system/model-providers/*` | 模型供应商配置 | ✓ | +| `skill_router.py` | `/api/system/skills/*` | Skills 管理 | ✓ | +| `subagent_router.py` | `/api/system/subagents/*` | 子智能体管理 | ✓ | +| `tool_router.py` | `/api/system/tools/*` | 工具列表与配置 | ✓ | +| `auth_apikey_router.py` | `/api/apikey/*` | API Key 管理 | ✓ | +| `filesystem_router.py` | `/api/viewer/filesystem/*` | 工作台文件系统视图 | ✓ | +| `workspace_router.py` | `/api/workspace/*` | 用户个人工作区 | ✓ | +| `channels_router.py` | `/api/channels/*` | 多渠道网关管理 | ✓ | +| `webhook_router.py` | `/api/webhook/*` | 统一 Webhook 分发 | ✓ | +| `knowledge_router.py` | `/api/knowledge/*` | 知识库管理与检索 | ✗ | +| `knowledge_eval_router.py` | `/api/evaluation/*` | 知识库评估 | ✗ | +| `knowledge_mindmap_router.py` | `/api/mindmap/*` | 思维导图生成与查询 | ✗ | +| `graph_router.py` | `/api/graph/*` | 图谱查询与管理 | ✗ | + +#### 3.1.3 Worker 入口 (`server/worker_main.py`) + +- ARQ Worker 入口，导入 `WorkerSettings` +- 处理智能体运行等异步任务 + +### 3.2 智能体体系 (`backend/package/yuxi/agents/`) + +#### 3.2.1 核心基类 + +| 文件 | 类/函数 | 职责 | +|------|---------|------| +| `base.py` | `BaseAgent` | 智能体基类，定义 graph 编译、消息流式处理、历史记录、checkpointer 管理 | +| `context.py` | `BaseContext` | 智能体运行配置上下文，包含模型、工具、知识库、MCP、Skills、子智能体等配置 | +| `state.py` | `BaseState` | LangGraph 状态定义，包含 messages 和 artifacts | +| `state.py` | `AgentStatePayload` | 前端可消费的序列化状态结构 | +| `models.py` | `load_chat_model()` | 根据 v1/v2 spec 加载聊天模型，支持多供应商 | + +#### 3.2.2 内置智能体 (`agents/buildin/`) + +| 智能体 | 文件 | 职责 | +|--------|------|------| +| ChatbotAgent | `chatbot/graph.py` | 基础对话机器人，支持文件上传、知识库、MCP、Skills、子智能体 | +| DeepAgent | `deep_agent/graph.py` | 深度研究智能体，用于复杂任务分解与执行 | + +#### 3.2.3 中间件 (`agents/middlewares/`) + +中间件负责把各种能力挂载到智能体运行时： + +| 中间件 | 职责 | +|--------|------| +| `KnowledgeBaseMiddleware` | 知识库检索工具注入 | +| `SkillsMiddleware` | Skills 提示词注入、依赖展开、动态激活 | +| `RuntimeConfigMiddleware` | 运行时配置应用（模型/工具/MCP/提示词） | +| `SummaryOffloadMiddleware` | 上下文摘要优化（token 阈值触发） | +| `SubAgentMiddleware` | 子智能体调度 | +| `FilesystemMiddleware` | 沙盒文件系统后端 | +| `TodoListMiddleware` | 待办事项管理 | +| `PatchToolCallsMiddleware` | 工具调用补丁 | +| `ModelRetryMiddleware` | 模型重试机制 | + +#### 3.2.4 工具集 (`agents/toolkits/`) + +| 模块 | 职责 | +|------|------| +| `buildin/tools.py` | 内置工具（如 tavily_search、ask_user_question） | +| `kbs/tools.py` | 知识库相关工具 | +| `mysql/tools.py` | MySQL 数据库工具 | +| `registry.py` | 工具注册中心 | + +#### 3.2.5 后端执行 (`agents/backends/`) + +| 模块 | 职责 | +|------|------| +| `sandbox/backend.py` | 沙盒执行后端 | +| `sandbox/paths.py` | 沙盒路径管理 | +| `composite.py` | 复合后端（知识库 + Skills + 沙盒） | +| `skills_backend.py` | Skills 执行后端 | + +### 3.3 服务层 (`backend/package/yuxi/services/`) + +服务层是用例层，负责串联 repositories、agents、knowledge、storage 和外部系统。 + +| 服务文件 | 职责 | 关键函数 | +|----------|------|----------| +| `chat_service.py` | 聊天核心服务 | `agent_chat()`, `stream_agent_chat()`, `stream_agent_resume()`, `get_agent_state_view()` | +| `agent_run_service.py` | Agent 运行管理（创建/轮询/取消） | `create_agent_run_view()`, `stream_agent_run_events()`, `cancel_agent_run_view()` | +| `run_worker.py` | ARQ Worker 任务处理 | `process_agent_run()`, `RunContext`, `ChunkedEventWriter` | +| `run_queue_service.py` | 运行队列与事件流 | `get_arq_pool()`, `append_run_stream_event()`, `publish_cancel_signal()` | +| `skill_service.py` | Skills 业务逻辑 | `import_skill_zip()`, `list_skills()`, `install_builtin_skill()`, `update_builtin_skill()` | +| `mcp_service.py` | MCP 服务管理 | `get_mcp_tools()`, `get_enabled_mcp_server_config()`, `ensure_builtin_mcp_servers_in_db()` | +| `subagent_service.py` | 子智能体管理 | `get_subagents_from_names()`, `init_builtin_subagents()` | +| `langfuse_service.py` | 可观测性追踪 | `build_run_context()`, `get_trace_info()`, `flush_langfuse()` | +| `model_provider_service.py` | 模型供应商配置 | `ensure_builtin_model_providers_in_db()`, `get_all_model_providers()` | +| `model_cache.py` | 模型信息缓存 | `model_cache.rebuild()`, `is_v2_spec_format()` | +| `conversation_service.py` | 对话管理 | - | +| `filesystem_service.py` | 文件系统服务 | - | +| `knowledge_fs_service.py` | 知识库文件服务 | - | +| `task_service.py` | 后台任务调度 | `tasker.start()`, `tasker.shutdown()` | +| `upload_utils.py` | 文件上传工具 | - | +| `workspace_service.py` | 工作区服务 | - | +| `evaluation_service.py` | 知识库评估 | - | +| `feedback_service.py` | 反馈服务 | - | +| `oidc_service.py` | OIDC 认证 | - | +| `tool_service.py` | 工具元数据 | `get_tool_metadata()` | + +### 3.4 知识库领域 (`backend/package/yuxi/knowledge/`) + +| 文件 | 类/函数 | 职责 | +|------|---------|------| +| `base.py` | `KnowledgeBase` (ABC) | 知识库抽象基类，定义统一接口（创建/删除/查询/文件管理） | +| `manager.py` | `KnowledgeBaseManager` | 知识库管理器，统一管理多种类型知识库实例 | +| `factory.py` | `KnowledgeBaseFactory` | 知识库工厂，根据类型创建实例 | +| `implementations/dify.py` | `DifyKnowledgeBase` | Dify 知识库实现 | +| `implementations/milvus.py` | `MilvusKnowledgeBase` | Milvus 向量知识库实现 | +| `chunking/` | - | 文档分块策略（RAGflow-like 语义分块） | +| `graphs/adapters/` | - | 图谱适配与上传服务 | +| `utils/kb_utils.py` | - | 知识库通用工具 | + +### 3.5 数据访问层 (`backend/package/yuxi/repositories/`) + +| 仓库文件 | 职责 | +|----------|------| +| `skill_repository.py` | Skills CRUD | +| `task_repository.py` | 后台任务 CRUD | +| `user_repository.py` | 用户 CRUD | +| `agent_config_repository.py` | 智能体配置 CRUD | +| `agent_run_repository.py` | 智能体运行记录 CRUD | +| `conversation_repository.py` | 对话与消息 CRUD | +| `knowledge_base_repository.py` | 知识库元数据 CRUD | +| `knowledge_file_repository.py` | 知识库文件 CRUD | +| `evaluation_repository.py` | 评估基准与结果 CRUD | + +### 3.6 存储层 (`backend/package/yuxi/storage/`) + +| 模块 | 职责 | +|------|------| +| `postgres/manager.py` | `PostgresManager` - 数据库连接池、会话管理、schema 迁移 | +| `postgres/models_business.py` | 业务数据模型（User、Conversation、AgentRun 等） | +| `postgres/models_knowledge.py` | 知识库数据模型（KnowledgeBase、KnowledgeFile 等） | +| `postgres/models_channels.py` | 渠道网关数据模型 | +| `minio/client.py` | MinIO 客户端封装 | + +### 3.7 多渠道网关 (`backend/package/yuxi/channels/`) + +支持 20+ 即时通讯渠道接入： + +| 渠道 | 适配器位置 | +|------|-----------| +| Slack | `channels/adapters/slack/` | +| Discord | `channels/adapters/discord/` | +| Telegram | `channels/adapters/telegram/` | +| 飞书 (Feishu) | `channels/adapters/feishu/` | +| 微信 (WeChat) | `channels/adapters/wechat/` | +| QQ 机器人 | `channels/adapters/qqbot/` | +| 钉钉 | `channels/adapters/dingding/` | +| Microsoft Teams | `channels/adapters/msteams/` | +| Matrix | `channels/adapters/matrix/` | +| IRC | `channels/adapters/irc/` | +| Line | `channels/adapters/line/` | +| Twitch | `channels/adapters/twitch/` | +| Nostr | `channels/adapters/nostr/` | +| Signal | `channels/adapters/signal/` | +| WhatsApp | `channels/adapters/whatsapp/` | +| Urbit | `channels/adapters/urbit/` | +| 元宝 (Yuanbao) | `channels/adapters/yuanbao/` | +| Zalo | `channels/adapters/zalo_oa/`, `zalo_user/` | + +核心文件： +- `channels/manager.py` - 渠道管理器 +- `channels/base.py` - 渠道基类 +- `channels/router.py` - 消息路由 +- `channels/message_actions.py` - 消息动作处理 + +### 3.8 配置与模型 (`backend/package/yuxi/config/`, `models/`) + +| 文件 | 职责 | +|------|------| +| `config/app.py` | 应用配置管理 | +| `config/builtin_providers.py` | 内置模型供应商配置 | +| `config/static/models.py` | 静态模型信息 | +| `models/chat.py` | 聊天模型适配 | +| `models/embed.py` | Embedding 模型适配 | +| `models/rerank.py` | Rerank 模型适配 | + +### 3.9 文档解析 (`backend/package/yuxi/plugins/parser/`) + +| 文件 | 职责 | +|------|------| +| `unified.py` | `Parser` 统一入口，封装所有解析实现 | +| `factory.py` | 解析器工厂 | +| `base.py` | 解析器抽象基类 | +| `mineru.py` | MinerU 解析实现 | +| `mineru_official.py` | MinerU 官方 API 实现 | +| `pp_structure_v3.py` | PaddleX 结构解析 | +| `rapid_ocr.py` | RapidOCR 实现 | +| `deepseek_ocr.py` | DeepSeek OCR 实现 | + +--- + +## 4. 前端核心模块 + +前端是 Vue 3 + Vite 应用，业务入口集中在 `web/src/`。 + +### 4.1 入口与路由 (`web/src/`) + +| 文件 | 职责 | +|------|------| +| `main.js` | 应用挂载入口 | +| `App.vue` | 根组件 | +| `router/index.js` | 路由配置，含权限守卫（requiresAuth/requiresAdmin/requiresSuperAdmin） | + +**路由表：** + +| 路径 | 页面 | 权限 | +|------|------|------| +| `/` | HomeView | 公开 | +| `/login` | LoginView | 公开 | +| `/agent` | AgentView | 登录用户 | +| `/agent/:thread_id` | AgentView（带线程ID） | 登录用户 | +| `/workspace` | WorkspaceView | 登录用户 | +| `/graph` | GraphView | 管理员 | +| `/database` | DataBaseView | 管理员 | +| `/database/:database_id` | DataBaseInfoView | 管理员 | +| `/dashboard` | DashboardView | 管理员 | +| `/model-config` | ModelConfigView | 管理员 | +| `/channels` | ChannelManageView | 管理员 | +| `/extensions` | ExtensionsView | 超级管理员 | + +### 4.2 API 封装 (`web/src/apis/`) + +所有后端接口统一封装，复用 `base.js` 的请求、鉴权和错误处理。 + +| API 文件 | 职责 | +|----------|------| +| `base.js` | HTTP 客户端、请求/响应拦截、权限头处理 | +| `agent_api.js` | 智能体管理、聊天接口 | +| `knowledge_api.js` | 知识库管理、文档管理、查询 | +| `graph_api.js` | 知识图谱操作 | +| `mcp_api.js` | MCP 服务器管理 | +| `skill_api.js` | Skills 管理 | +| `subagent_api.js` | 子智能体管理 | +| `system_api.js` | 系统状态、配置 | +| `auth_api.js` | 认证、用户信息 | +| `dashboard_api.js` | 仪表盘数据 | +| `department_api.js` | 部门管理 | +| `channel_api.js` | 渠道管理 | +| `tool_api.js` | 工具信息 | +| `tasker.js` | 任务管理 | +| `mindmap_api.js` | 思维导图 | +| `workspace_api.js` | 工作区 | +| `viewer_filesystem.js` | 文件系统视图 | +| `apikey_api.js` | API Key 管理 | + +### 4.3 状态管理 (`web/src/stores/`) + +使用 Pinia + `pinia-plugin-persistedstate` 持久化。 + +| Store 文件 | 职责 | +|-----------|------| +| `user.js` | 用户信息、登录状态、权限 | +| `agent.js` | 智能体配置、初始化状态 | +| `chatThreads.js` | 聊天线程列表、当前线程 | +| `database.js` | 知识库列表、当前知识库 | +| `graphStore.js` | 图谱数据、可视化状态 | +| `theme.js` | 主题（暗黑/亮色） | +| `config.js` | 系统配置 | +| `chatUI.js` | 聊天界面状态 | +| `tasker.js` | 任务状态 | +| `channel.js` | 渠道配置 | +| `info.js` | 系统信息 | + +### 4.4 可组合逻辑 (`web/src/composables/`) + +| 文件 | 职责 | +|------|------| +| `useAgentStreamHandler.js` | Agent 流式响应处理 | +| `useAgentRunStream.js` | Agent runs SSE 流处理 | +| `useApproval.js` | 审批逻辑处理 | +| `useMention.js` | @提及功能 | +| `useStreamSmoother.js` | 流式消息平滑调度 | + +### 4.5 视图与组件 (`web/src/views/`, `components/`) + +| 视图 | 职责 | +|------|------| +| `AgentView.vue` | 智能体对话主页面 | +| `WorkspaceView.vue` | 个人工作区（文件管理） | +| `DataBaseView.vue` | 知识库列表与管理 | +| `DataBaseInfoView.vue` | 知识库详情与文件管理 | +| `GraphView.vue` | 知识图谱可视化 | +| `DashboardView.vue` | 数据统计仪表盘 | +| `ModelConfigView.vue` | 模型供应商配置 | +| `ChannelManageView.vue` | 渠道网关管理 | +| `ExtensionsView.vue` | 扩展管理（Skills/MCP/SubAgents） | +| `LoginView.vue` | 登录页面 | +| `HomeView.vue` | 首页 | + +--- + +## 5. 关键类与函数详解 + +### 5.1 智能体基类 + +#### `BaseAgent` (`agents/base.py`) + +```python +class BaseAgent: + name = "base_agent" + description = "base_agent" + capabilities: list[str] = [] + context_schema: type[BaseContext] = BaseContext + + # 核心方法 + async def get_graph(self, **kwargs) -> CompiledStateGraph # 子类必须实现 + async def stream_messages(self, messages, input_context=None, **kwargs) # 流式消息 + async def stream_messages_with_state(self, messages, input_context=None, **kwargs) # 流式消息+状态 + async def invoke_messages(self, messages, input_context=None, **kwargs) # 同步调用 + async def get_history(self, user_id, thread_id) -> list[dict] # 获取历史 + async def get_info(self, include_configurable_items=True) # 获取元数据 + def reload_graph(self) # 重置 graph 缓存 +``` + +#### `BaseContext` (`agents/context.py`) + +```python +@dataclass(kw_only=True) +class BaseContext: + thread_id: str # 对话线程ID + user_id: str # 用户ID + system_prompt: str # 系统提示词 + model: str # 主模型（v2 spec: provider_id:model_id） + tools: list[str] # 启用工具列表 + knowledges: list[str] # 启用知识库列表 + mcps: list[str] # 启用 MCP 服务器列表 + skills: list[str] # 启用 Skills 列表 + subagents_model: str # 子智能体默认模型 + subagents: list[str] # 启用子智能体列表 + summary_threshold: int # 上下文摘要触发阈值（KB） +``` + +### 5.2 聊天服务 + +#### `chat_service.py` 核心函数 + +| 函数 | 职责 | +|------|------| +| `agent_chat()` | 非流式对话，返回完整响应 | +| `stream_agent_chat()` | 流式对话，yield 事件块 | +| `stream_agent_resume()` | 恢复中断的对话（处理 human-in-the-loop） | +| `get_agent_state_view()` | 获取 Agent 当前状态 | +| `save_messages_from_langgraph_state()` | 从 LangGraph state 持久化消息 | +| `extract_agent_state()` | 从 state 提取 todos/files/artifacts | + +### 5.3 Agent Run 服务 + +#### `agent_run_service.py` 核心函数 + +| 函数 | 职责 | +|------|------| +| `create_agent_run_view()` | 创建后台运行任务，入队 ARQ | +| `stream_agent_run_events()` | SSE 流式推送运行事件 | +| `cancel_agent_run_view()` | 取消运行任务 | +| `get_active_run_by_thread()` | 获取线程的活跃运行 | + +### 5.4 Worker 任务处理 + +#### `run_worker.py` 核心类/函数 + +| 类/函数 | 职责 | +|---------|------| +| `process_agent_run()` | ARQ 任务主函数，消费流并写入 Redis | +| `RunContext` | 运行上下文，管理取消信号监听 | +| `ChunkedEventWriter` | 事件块缓冲写入器 | +| `WorkerSettings` | ARQ Worker 配置（max_tries=2, job_timeout=900s） | + +### 5.5 知识库管理 + +#### `KnowledgeBaseManager` (`knowledge/manager.py`) + +```python +class KnowledgeBaseManager: + async def create_database(name, description, kb_type="lightrag", ...) # 创建知识库 + async def delete_database(db_id) # 删除知识库 + async def add_file_record(db_id, item, params) # 添加文件记录 + async def parse_file(db_id, file_id) # 解析文件为 Markdown + async def index_file(db_id, file_id) # 索引文件到向量库 + async def aquery(query_text, db_id, **kwargs) # 异步查询 + def get_retrievers() -> dict[str, dict] # 获取所有检索器 +``` + +### 5.6 MCP 服务 + +#### `mcp_service.py` 核心函数 + +| 函数 | 职责 | +|------|------| +| `get_mcp_tools()` | 获取指定服务器的工具（带缓存） | +| `get_enabled_mcp_tools()` | Agent 统一入口（自动过滤 disabled_tools） | +| `get_tools_from_all_servers()` | 获取所有启用服务器的工具 | +| `ensure_builtin_mcp_servers_in_db()` | 同步内置 MCP 配置到数据库 | +| `create_mcp_server()` / `update_mcp_server()` / `delete_mcp_server()` | CRUD | + +### 5.7 Skills 服务 + +#### `skill_service.py` 核心函数 + +| 函数 | 职责 | +|------|------| +| `import_skill_zip()` | 从 ZIP 导入 Skill | +| `list_skills()` | 列出所有 Skills | +| `install_builtin_skill()` | 安装内置 Skill | +| `update_builtin_skill()` | 更新内置 Skill（带覆盖确认） | +| `sync_thread_visible_skills()` | 同步线程可见 Skills 到沙盒 | + +### 5.8 数据库管理 + +#### `PostgresManager` (`storage/postgres/manager.py`) + +```python +class PostgresManager(metaclass=SingletonMeta): + def initialize() # 初始化连接池 + async def create_business_tables() # 创建业务表 + async def ensure_business_schema() # 确保业务 schema（增量迁移） + async def ensure_knowledge_schema() # 确保知识库 schema + async def get_async_session_context() # 异步会话上下文管理器 +``` + +--- + +## 6. 数据流与运行链路 + +### 6.1 典型智能体对话流程 + +``` +1. AgentView 收集输入、附件、配置 + ↓ +2. web/src/apis/agent_api.js 调用 /api/chat/* + ↓ +3. server/routers/chat_router.py → chat_service / agent_run_service + ↓ +4. 服务层读取 conversation、agent_config、tools、skills、knowledge + ↓ +5a. 同步模式: 直接调用 stream_agent_chat() → 返回 SSE 流 +5b. 异步模式: create_agent_run_view() → ARQ 入队 → worker 执行 + ↓ +6. worker 执行 LangGraph 智能体 + - 中间件挂载知识库、工具、Skills、MCP、附件、沙盒 + - 运行事件写入 Redis + ↓ +7. 最终状态和业务记录写入 Postgres + 文件和产物落到 saves/、MinIO 或沙盒映射目录 + ↓ +8. 前端通过 SSE/轮询消费运行事件 + 渲染消息、工具调用、引用来源、产物卡片、文件预览 +``` + +### 6.2 Agent Run 异步流程 + +``` +用户请求 + ↓ +create_agent_run_view() + - 创建 AgentRun 记录（status=pending） + - ARQ enqueue_job("process_agent_run", run.id) + ↓ +worker process_agent_run() + - 加载用户和配置 + - mark_run_running() + - stream_agent_chat() 消费流 + - ChunkedEventWriter 缓冲写入 Redis + - 处理取消信号 + - mark_run_terminal(status=completed/failed/cancelled) + ↓ +前端 SSE 连接 /api/chat/runs/{run_id}/events + - 轮询 Redis 事件 + - 心跳保活 + - 终端状态返回 close 事件 +``` + +### 6.3 知识库文档处理流程 + +``` +上传文件 + ↓ +add_file_record() → status=UPLOADED + ↓ +parse_file() → 调用 Parser.aparse() + - status=PARSING + - 解析为 Markdown + - 保存到 MinIO + - status=PARSED / ERROR_PARSING + ↓ +index_file() → 向量化和图谱构建 + - status=INDEXING + - 分块 → Embedding → 向量库 + - status=INDEXED / ERROR_INDEXING + ↓ +aquery() → 检索 → Rerank → 返回结果 +``` + +--- + +## 7. 依赖关系 + +### 7.1 后端核心依赖 + +``` +fastapi>=0.121 # Web 框架 +uvicorn[standard]>=0.34 # ASGI 服务器 +arq>=0.26.3 # 异步任务队列 +langgraph>=1.0.1 # 智能体编排 +langchain>=1.2.0 # LLM 框架 +langchain-openai>=1.0 # OpenAI 兼容层 +langchain-mcp-adapters # MCP 适配 +lightrag-hku>=1.4.6 # 知识图谱 +neo4j>=5.28 # 图数据库 +pymilvus>=2.5 # 向量数据库 +asyncpg>=0.30 # PostgreSQL 异步驱动 +psycopg[binary,pool] # PostgreSQL 连接池 +redis>=5.2 # 缓存/消息 +minio>=7.2 # 对象存储 +sqlalchemy[asyncio]>=2 # ORM +langfuse>=4.0 # 可观测性 +pydantic # 数据验证 +pyjwt>=2.8 # JWT 认证 +``` + +### 7.2 前端核心依赖 + +``` +vue@^3.5 # 框架 +vue-router@^4.6 # 路由 +pinia@^3.0 # 状态管理 +pinia-plugin-persistedstate # 状态持久化 +ant-design-vue@^4.2 # UI 组件库 +lucide-vue-next # 图标库 +@antv/g6@^5.0 # 图谱可视化 +sigma@^3.0 # 图谱渲染 +@vueuse/core # 组合式工具 +vite@^7.3 # 构建工具 +less@^4.5 # CSS 预处理器 +marked@^16 # Markdown 渲染 +highlight.js # 代码高亮 +``` + +### 7.3 模块依赖图 + +``` +server/main.py + ├── server/routers/* + │ └── yuxi.services.* + │ ├── yuxi.agents.* (BaseAgent, agent_manager) + │ ├── yuxi.knowledge.* (KnowledgeBaseManager) + │ ├── yuxi.repositories.* (SQLAlchemy CRUD) + │ ├── yuxi.storage.* (Postgres, MinIO) + │ └── yuxi.channels.* (多渠道网关) + └── server/utils/* (lifespan, auth, middleware) + +yuxi.agents.base.BaseAgent + ├── yuxi.agents.context.BaseContext + ├── yuxi.agents.models.load_chat_model + └── yuxi.storage.postgres.manager.pg_manager + +yuxi.services.chat_service + ├── yuxi.agents.buildin.agent_manager + ├── yuxi.repositories.conversation_repository + ├── yuxi.services.langfuse_service + └── yuxi.plugins.guard.content_guard +``` + +--- + +## 8. 项目运行方式 + +### 8.1 环境要求 + +- Docker & Docker Compose +- Python 3.12+（后端开发） +- Node.js 20+ + pnpm（前端开发） +- Git + +### 8.2 快速启动 + +```bash +# 1. 克隆项目 +git clone --branch v0.6.2 --depth 1 https://github.com/xerrors/Yuxi.git +cd Yuxi + +# 2. 初始化（创建 .env 等） +# Linux/macOS +./scripts/init.sh +# Windows PowerShell +.\scripts\init.ps1 + +# 3. 启动全部服务 +docker compose up -d + +# 或使用 Makefile +make up + +# 4. 访问前端 +open http://localhost:5173 +``` + +### 8.3 LITE 模式（轻量启动） + +跳过知识库、图谱、评估等重依赖： + +```bash +make up-lite +# 等效于: +# LITE_MODE=true VITE_USE_RUNS_API=false docker compose up -d postgres redis minio api web +``` + +### 8.4 开发常用命令 + +```bash +# 查看日志 +make logs +docker logs api-dev --tail 100 +docker logs worker-dev --tail 100 + +# 代码格式化 +make format +# 等效于: +cd backend && uv run ruff format package +cd backend && uv run ruff check package --fix +cd web && pnpm run format +cd web && pnpm run lint + +# 运行测试 +cd backend && uv run --group test pytest + +# 停止服务 +make down +``` + +### 8.5 开发环境特性 + +- **热重载**：`api-dev` 和 `web-dev` 服务均配置热重载，本地修改代码后无需重启容器 +- **代码挂载**：`backend/server`、`backend/package`、`web/src` 等目录挂载到容器 +- **调试**：使用 `docker logs` 查看实时日志 + +--- + +## 9. 测试体系 + +测试代码放在 `backend/test/`，按三层组织： + +| 层级 | 目录 | 职责 | +|------|------|------| +| 单元测试 | `test/unit/` | 不依赖外部服务的纯逻辑测试 | +| 集成测试 | `test/integration/` | API 路由测试（需运行中的服务） | +| 端到端测试 | `test/e2e/` | 完整业务流程测试 | + +### 9.1 运行测试 + +```bash +cd backend +uv run --group test pytest + +# 带覆盖率 +uv run --group test pytest --cov=yuxi --cov-report=html + +# 仅运行单元测试 +uv run --group test pytest -m unit + +# 仅运行集成测试 +uv run --group test pytest -m integration + +# 仅运行 e2e 测试 +uv run --group test pytest -m e2e +``` + +### 9.2 测试配置 + +- `pytest.ini` 配置在 `backend/pyproject.toml` 中 +- 全局 fixtures 在 `test/conftest.py` +- 各层独立 `conftest.py` 管理层级 fixtures + +--- + +## 10. 附录：目录结构总览 + +``` +ForcePilot/ +├── .github/ # GitHub 配置（Issue 模板、工作流） +│ ├── ISSUE_TEMPLATE/ +│ └── workflows/ +├── backend/ # 后端代码 +│ ├── package/ # 可复用业务包 +│ │ └── yuxi/ # 核心包 +│ │ ├── agents/ # 智能体体系 +│ │ │ ├── backends/ # 执行后端（沙盒、Skills） +│ │ │ ├── buildin/ # 内置智能体 +│ │ │ ├── middlewares/# 中间件 +│ │ │ ├── toolkits/ # 工具注册与实现 +│ │ │ ├── base.py # BaseAgent +│ │ │ ├── context.py # BaseContext +│ │ │ ├── models.py # 模型加载 +│ │ │ └── state.py # 状态定义 +│ │ ├── channels/ # 多渠道网关 +│ │ │ └── adapters/ # 各渠道适配器 +│ │ ├── config/ # 应用配置 +│ │ ├── gateway/ # 网关协议 +│ │ ├── knowledge/ # 知识库领域 +│ │ │ ├── chunking/ # 分块策略 +│ │ │ ├── graphs/ # 图谱适配 +│ │ │ └── implementations/ # 具体实现 +│ │ ├── models/ # 模型适配（chat/embed/rerank） +│ │ ├── plugins/ # 插件 +│ │ │ └── parser/ # 文档解析 +│ │ ├── repositories/ # 数据访问层 +│ │ ├── services/ # 业务服务层 +│ │ ├── storage/ # 存储基础设施 +│ │ │ ├── minio/ # 对象存储 +│ │ │ └── postgres/ # PostgreSQL +│ │ └── utils/ # 通用工具 +│ ├── server/ # Web 应用入口 +│ │ ├── routers/ # HTTP 路由 +│ │ ├── utils/ # Web 层工具 +│ │ ├── main.py # FastAPI 入口 +│ │ └── worker_main.py # Worker 入口 +│ ├── test/ # 测试代码 +│ │ ├── unit/ # 单元测试 +│ │ ├── integration/ # 集成测试 +│ │ └── e2e/ # 端到端测试 +│ └── pyproject.toml # 后端项目配置 +├── web/ # 前端代码 +│ ├── src/ +│ │ ├── apis/ # API 封装 +│ │ ├── components/ # 可复用组件 +│ │ ├── composables/ # 可组合逻辑 +│ │ ├── layouts/ # 布局组件 +│ │ ├── router/ # 路由配置 +│ │ ├── stores/ # Pinia 状态 +│ │ ├── utils/ # 前端工具 +│ │ ├── views/ # 页面级视图 +│ │ ├── App.vue # 根组件 +│ │ └── main.js # 入口 +│ ├── public/ +│ ├── index.html +│ ├── package.json +│ └── vite.config.js +├── docker/ # Docker 配置 +│ ├── api.Dockerfile +│ ├── web.Dockerfile +│ ├── sandbox_provisioner/ # 沙盒供应器 +│ └── volumes/ # 数据卷挂载点 +├── docs/ # 项目文档 +│ ├── .vitepress/ # VitePress 配置 +│ ├── agents/ # Agent 开发文档 +│ ├── develop-guides/ # 开发指南 +│ └── vibe/ # 开发者笔记 +├── docker-compose.yml # Docker Compose 配置 +├── Makefile # 快捷命令 +├── README.md # 项目说明 +├── ARCHITECTURE.md # 架构文档 +├── AGENTS.md # 开发准则 +└── .env.template # 环境变量模板 +``` + +--- + +## 参考文档 + +- [ARCHITECTURE.md](ARCHITECTURE.md) - 架构代码地图 +- [AGENTS.md](AGENTS.md) - 开发准则与行为规范 +- [README.md](README.md) - 项目快速开始 +- [docker-compose.yml](docker-compose.yml) - 服务拓扑与配置 +- [backend/package/pyproject.toml](backend/package/pyproject.toml) - Python 依赖 +- [web/package.json](web/package.json) - Node.js 依赖 diff --git a/backend/package/yuxi/repositories/channel_message_record_repository.py b/backend/package/yuxi/repositories/channel_message_record_repository.py index f5139c58..6eabe221 100644 --- a/backend/package/yuxi/repositories/channel_message_record_repository.py +++ b/backend/package/yuxi/repositories/channel_message_record_repository.py @@ -95,9 +95,7 @@ class ChannelMessageRecordRepository: cutoff = utc_now_naive() - timedelta(seconds=timeout_seconds) return [r for r in records if r.created_at and r.created_at < cutoff] - async def get_recent_records( - self, channel_id: str, chat_id: str, limit: int = 10 - ) -> list[ChannelMsgRecord]: + async def get_recent_records(self, channel_id: str, chat_id: str, limit: int = 10) -> list[ChannelMsgRecord]: result = await self.db.execute( select(ChannelMsgRecord) .where( @@ -109,12 +107,12 @@ class ChannelMessageRecordRepository: ) return list(result.scalars().all()) - async def get_chat_message_count( - self, channel_id: str, chat_id: str, since_hours: int = 24 - ) -> int: + async def get_chat_message_count(self, channel_id: str, chat_id: str, since_hours: int = 24) -> int: cutoff = utc_now_naive() - timedelta(hours=since_hours) result = await self.db.execute( - select(func.count()).select_from(ChannelMsgRecord).where( + select(func.count()) + .select_from(ChannelMsgRecord) + .where( ChannelMsgRecord.channel_id == channel_id, ChannelMsgRecord.chat_id == chat_id, ChannelMsgRecord.created_at >= cutoff, diff --git a/backend/package/yuxi/storage/postgres/models_channels.py b/backend/package/yuxi/storage/postgres/models_channels.py index 470752ab..b6648313 100644 --- a/backend/package/yuxi/storage/postgres/models_channels.py +++ b/backend/package/yuxi/storage/postgres/models_channels.py @@ -4,6 +4,7 @@ from typing import Any from sqlalchemy import ( BigInteger, + Boolean, Column, DateTime, Index, @@ -17,6 +18,31 @@ from yuxi.storage.postgres.models_business import Base from yuxi.utils.datetime_utils import format_utc_datetime, utc_now_naive +class ChannelConfig(Base): + __tablename__ = "channels_configs" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + channel_id = Column(String(32), unique=True, nullable=False, index=True) + config_json = Column(JSONB, nullable=False, default=dict) + enabled = Column(Boolean, nullable=False, default=True) + registered_by = Column(String(64), nullable=True) + created_at = Column(DateTime, nullable=False, default=utc_now_naive) + updated_at = Column(DateTime, nullable=False, default=utc_now_naive, onupdate=utc_now_naive) + + __table_args__ = (Index("idx_channels_configs_channel_id", "channel_id"),) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_id": self.channel_id, + "config_json": self.config_json or {}, + "enabled": self.enabled, + "registered_by": self.registered_by, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + class ChannelUserMapping(Base): """渠道用户映射表 - 渠道用户与内部用户的绑定关系""" @@ -64,6 +90,18 @@ class ChannelThreadMapping(Base): "internal_user_id", name="uq_channel_thread", ), + Index( + "idx_thread_mapping_user_recent", + "channel_id", + "internal_user_id", + text("last_active_at DESC"), + ), + Index( + "idx_thread_mapping_chat_lookup", + "channel_id", + "channel_chat_id", + text("last_active_at DESC"), + ), ) def to_dict(self) -> dict[str, Any]: @@ -115,6 +153,12 @@ class ChannelMsgRecord(Base): postgresql_where=(status == "processing"), ), Index("idx_msg_records_chat", "channel_id", "chat_id"), + Index( + "idx_msg_records_channel_rt", + "channel_id", + "response_time_ms", + postgresql_where=(response_time_ms.isnot(None)), + ), ) def to_dict(self) -> dict[str, Any]: @@ -148,7 +192,7 @@ class ChannelPolicyConfig(Base): id = Column(BigInteger, primary_key=True, autoincrement=True) channel_id = Column(String(32), nullable=False, unique=True) - group_chat_mode = Column(String(16), nullable=False, default="all") + group_chat_mode = Column(String(16), nullable=False, default="mention_only") whitelist_ids = Column(JSONB, nullable=False, default=list) welcome_message = Column(String(500), nullable=True) work_hours_start = Column(String(5), nullable=True, default="09:00") @@ -183,7 +227,6 @@ class ChannelRoutingRule(Base): channel_id = Column(String(32), nullable=False) command = Column(String(64), nullable=False) agent_config_id = Column(String(64), nullable=False) - priority = Column(Integer, nullable=False, default=0) created_at = Column(DateTime, nullable=False, default=utc_now_naive) updated_at = Column(DateTime, nullable=False, default=utc_now_naive, onupdate=utc_now_naive) @@ -198,7 +241,6 @@ class ChannelRoutingRule(Base): "channel_id": self.channel_id, "command": self.command, "agent_config_id": self.agent_config_id, - "priority": self.priority, "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), } diff --git a/backend/server/main.py b/backend/server/main.py index 1af66310..8fe61472 100644 --- a/backend/server/main.py +++ b/backend/server/main.py @@ -23,7 +23,7 @@ from starlette.middleware.base import BaseHTTPMiddleware from server.routers import router from server.routers.ws_chat_router import ws_chat from yuxi.channels.adapters.slack.http_handler import slack_webhook -from yuxi.channels.adapters.nostr.profile_api import profile_router +from yuxi.channels.message_actions import ChannelNotFoundError from server.utils.lifespan import lifespan from server.utils.auth_middleware import is_public_path from server.utils.common_utils import setup_logging @@ -47,8 +47,15 @@ app.include_router(router, prefix="/api") app.include_router(ws_chat) # Slack HTTP Webhook (单独注册以处理 URL verification challenge) app.include_router(slack_webhook) -# Nostr Profile API -app.include_router(profile_router) + + +@app.exception_handler(ChannelNotFoundError) +async def channel_not_found_handler(request: Request, exc: ChannelNotFoundError): + return JSONResponse( + status_code=status.HTTP_404_NOT_FOUND, + content={"code": -1, "data": None, "message": str(exc)}, + ) + # CORS 设置 app.add_middleware( diff --git a/backend/server/routers/channels_router.py b/backend/server/routers/channels_router.py index ac38a28e..f675d085 100644 --- a/backend/server/routers/channels_router.py +++ b/backend/server/routers/channels_router.py @@ -1,14 +1,37 @@ from __future__ import annotations -from typing import Any +import csv +import codecs +import json +import re +from datetime import datetime +from io import StringIO +from typing import Any, Literal -from fastapi import APIRouter, Depends, HTTPException, Query, status -from sqlalchemy import select +from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, status +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field +from sqlalchemy import asc, desc, func, select +from sqlalchemy.exc import DBAPIError, IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from starlette.responses import JSONResponse from yuxi.channels.manager import get_channel_manager +from yuxi.channels.services.plugin_state_store import PluginStateStore, PostgresPluginStateStore +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, + ChannelException, + ChannelNotConnectedError, + ChannelRateLimitError, + ChannelTimeoutError, +) from yuxi.storage.postgres.models_business import User -from yuxi.storage.postgres.models_channels import ChannelThreadMapping, ChannelUserMapping +from yuxi.storage.postgres.models_channels import ( + ChannelMsgRecord, + ChannelPolicyConfig, + ChannelThreadMapping, + ChannelUserMapping, +) from yuxi.utils.logging_config import logger from server.utils.auth_middleware import get_admin_user, get_db @@ -20,6 +43,11 @@ def _make_response(data: Any = None, code: int = 0, message: str = "ok") -> dict return {"code": code, "data": data, "message": message} +class UpdateStateEntryBody(BaseModel): + value: Any = Field(..., description="状态值（必填）") + ttl_seconds: int | None = Field(None, gt=0, description="过期时间（秒），必须为正整数") + + async def _check_rate_limit(user_id: str, key: str, max_req: int, window: int) -> None: rate_key = f"{key}:{user_id}" allowed = await get_channel_manager().check_rate_limit(rate_key, max_req, window) @@ -30,10 +58,292 @@ async def _check_rate_limit(user_id: str, key: str, max_req: int, window: int) - ) +_POLICY_FIELDS = [ + "group_chat_mode", + "whitelist_ids", + "welcome_message", + "work_hours_start", + "work_hours_end", + "timezone_offset", + "off_hours_reply", +] + +_POLICY_DEFAULTS = { + "group_chat_mode": "mention_only", + "whitelist_ids": [], + "welcome_message": "", + "work_hours_start": "09:00", + "work_hours_end": "18:00", + "timezone_offset": 8, + "off_hours_reply": "", +} + + +class ChannelPolicyUpdate(BaseModel): + group_chat_mode: Literal["mention_only", "all", "whitelist", "blacklist"] | None = None + whitelist_ids: list[str] | None = None + welcome_message: str | None = Field(None, max_length=500) + work_hours_start: str | None = Field(None, pattern=r"^\d{2}:\d{2}$") + work_hours_end: str | None = Field(None, pattern=r"^\d{2}:\d{2}$") + timezone_offset: int | None = Field(None, ge=-12, le=14) + off_hours_reply: str | None = Field(None, max_length=500) + + +def _get_adapter(channel_id: str): + adapter = get_channel_manager()._adapters.get(channel_id) + if adapter is None: + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found or not running") + return adapter + + +async def _check_channel_rate_limit(channel_id: str, user_id: str, action: str = "message") -> None: + result = await get_channel_manager().check_per_channel_rate_limit(channel_id, user_id, action) + if not result.allowed: + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail=f"请求频率超限，请在 {result.retry_after_seconds} 秒后重试", + ) + + +VALID_MESSAGE_STATUSES = frozenset({"processing", "success", "error", "timeout"}) +VALID_CONTENT_TYPES = frozenset({"text", "image", "file", "voice", "video", "location", "event"}) +MIN_SEARCH_LENGTH = 2 +MAX_SEARCH_LENGTH = 100 + +SORTABLE_FIELDS = frozenset({"created_at", "response_time_ms", "id"}) + + +def _escape_like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def _build_message_conditions( + channel_id: str, + status_value: str | None = None, + content_type: str | None = None, + date_from: str | None = None, + date_to: str | None = None, + search: str | None = None, +) -> list: + conditions = [ChannelMsgRecord.channel_id == channel_id] + + if status_value: + if status_value not in VALID_MESSAGE_STATUSES: + raise HTTPException( + status_code=400, + detail=f"无效的 status 值: '{status_value}'，可选: {', '.join(sorted(VALID_MESSAGE_STATUSES))}", + ) + conditions.append(ChannelMsgRecord.status == status_value) + + if content_type: + if content_type not in VALID_CONTENT_TYPES: + raise HTTPException( + status_code=400, + detail=f"无效的 content_type 值: '{content_type}'，可选: {', '.join(sorted(VALID_CONTENT_TYPES))}", + ) + conditions.append(ChannelMsgRecord.content_type == content_type) + + dt_from = None + dt_to = None + + if date_from: + try: + dt_from = datetime.strptime(date_from, "%Y-%m-%d") + conditions.append(ChannelMsgRecord.created_at >= dt_from) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"日期格式无效: '{date_from}'，正确格式: YYYY-MM-DD", + ) + + if date_to: + try: + dt_to = datetime.strptime(date_to, "%Y-%m-%d") + conditions.append(ChannelMsgRecord.created_at < dt_to) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"日期格式无效: '{date_to}'，正确格式: YYYY-MM-DD", + ) + + if dt_from is not None and dt_to is not None: + if dt_from >= dt_to: + raise HTTPException( + status_code=400, + detail=f"date_from ({date_from}) 必须早于 date_to ({date_to})", + ) + + if search: + search = search.strip() + if len(search) < MIN_SEARCH_LENGTH: + raise HTTPException( + status_code=400, + detail=f"搜索关键词至少 {MIN_SEARCH_LENGTH} 个字符", + ) + if len(search) > MAX_SEARCH_LENGTH: + raise HTTPException( + status_code=400, + detail=f"搜索关键词最多 {MAX_SEARCH_LENGTH} 个字符", + ) + escaped = _escape_like(search) + conditions.append(ChannelMsgRecord.content_preview.ilike(f"%{escaped}%", escape="\\")) + + return conditions + + +_CSV_RISKY_PREFIXES = frozenset(("=", "+", "-", "@")) + +_CSV_HEADER = [ + "id", + "channel_type", + "message_id", + "chat_id", + "chat_type", + "content_type", + "sender_user_id", + "content_preview", + "reply_to_message_id", + "agent_config_id", + "status", + "error_message", + "response_time_ms", + "reply_message_id", + "reply_content_preview", + "created_at", + "replied_at", + "extra_metadata", +] + + +def _sanitize_csv_cell(value: str | None) -> str: + if not value: + return "" + if value[0] in _CSV_RISKY_PREFIXES: + return f"'{value}" + return value + + +class _UTF8SigEncoder: + def __init__(self, generator): + self._gen = generator + self._sent_bom = False + + def __iter__(self): + return self + + def __next__(self): + if not self._sent_bom: + self._sent_bom = True + return codecs.BOM_UTF8 + return next(self._gen) + + +def _csv_line_generator(rows): + output = StringIO() + writer = csv.writer(output) + + writer.writerow(_CSV_HEADER) + yield output.getvalue() + output.truncate(0) + output.seek(0) + + for r in rows: + writer.writerow( + [ + r.id, + r.channel_type or "", + r.message_id or "", + r.chat_id, + r.chat_type, + r.content_type, + _sanitize_csv_cell(r.sender_user_id), + _sanitize_csv_cell(r.content_preview), + r.reply_to_message_id or "", + r.agent_config_id or "", + r.status, + _sanitize_csv_cell(r.error_message), + r.response_time_ms or "", + r.reply_message_id or "", + _sanitize_csv_cell(r.reply_content_preview), + r.created_at.strftime("%Y-%m-%dT%H:%M:%SZ") if r.created_at else "", + r.replied_at.strftime("%Y-%m-%dT%H:%M:%SZ") if r.replied_at else "", + json.dumps(r.extra_metadata, ensure_ascii=False) if r.extra_metadata else "", + ] + ) + yield output.getvalue() + output.truncate(0) + output.seek(0) + + +def _get_nostr_adapter(channel_id: str): + adapter = _get_adapter(channel_id) + profile_api = getattr(adapter, "_profile_api", None) + if profile_api is None: + raise HTTPException(status_code=501, detail=f"渠道 '{channel_id}' 不支持 Profile 操作") + return adapter, profile_api + + +def get_state_store() -> PluginStateStore: + return PostgresPluginStateStore() + + +@channels.get("/nostr/profile") +async def get_nostr_profile( + channel_id: str = Query(..., description="Nostr 渠道 ID"), + account_id: str = Query("default"), + current_user: User = Depends(get_admin_user), +): + adapter, profile_api = _get_nostr_adapter(channel_id) + profile = await profile_api.get_profile(account_id) + return _make_response(data=profile) + + +@channels.post("/nostr/profile/publish") +async def publish_nostr_profile( + channel_id: str = Query(..., description="Nostr 渠道 ID"), + profile_data: dict = Body(..., description="Profile 数据"), + account_id: str = Body("default"), + current_user: User = Depends(get_admin_user), +): + adapter, profile_api = _get_nostr_adapter(channel_id) + result = await profile_api.publish_profile(profile_data, account_id) + return _make_response(data=result) + + +@channels.get("/nostr/profile/import") +async def import_nostr_profile( + pubkey: str = Query(..., description="目标 Nostr 公钥 (hex)"), + channel_id: str = Query(..., description="Nostr 渠道 ID"), + current_user: User = Depends(get_admin_user), +): + adapter, profile_api = _get_nostr_adapter(channel_id) + result = await profile_api.import_profile(pubkey) + return _make_response(data=result) + + +@channels.post("/nostr/profile/merge") +async def merge_nostr_profile( + channel_id: str = Query(..., description="Nostr 渠道 ID"), + body: dict = Body(...), + current_user: User = Depends(get_admin_user), +): + local_data = body.get("local", {}) + imported_pubkey = body.get("imported_pubkey", "") + adapter, profile_api = _get_nostr_adapter(channel_id) + result = await profile_api.merge_profile(local_data, imported_pubkey) + return _make_response(data=result) + + @channels.get("/status") async def list_channels_status(current_user: User = Depends(get_admin_user)): status_data = await get_channel_manager().get_channel_status() - return _make_response(data=status_data) + meta = status_data.get("_meta", {}) + code = 0 + message = "ok" + if meta.get("failed", 0) > 0: + code = 2001 + message = f"部分渠道状态获取失败 ({meta['failed']}/{meta['total_channels']})" + return _make_response(code=code, data=status_data, message=message) @channels.get("/actions") @@ -120,10 +430,18 @@ async def get_channel_actions( @channels.get("/{channel_id}/status") async def get_channel_status( - channel_id: str, + channel_id: str = Path( + ..., + min_length=1, + max_length=64, + pattern=r"^[a-zA-Z0-9_-]+$", + description="渠道标识符", + ), + include_stats: bool = Query(False), current_user: User = Depends(get_admin_user), ): - status_data = await get_channel_manager().get_channel_status(channel_id) + await _check_rate_limit(str(current_user.id), "channel_status", 30, 60) + status_data = await get_channel_manager().get_channel_status(channel_id, include_stats=include_stats) if status_data.get("status") == "not_found": raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") return _make_response(data=status_data) @@ -140,21 +458,47 @@ async def start_channel( if not manager.is_registered(channel_id): raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") + config = manager._channels_config.get(channel_id, {}) + if not config.get("enabled", False): + raise HTTPException( + status_code=400, + detail=f"Channel '{channel_id}' is disabled in configuration", + ) + if manager.is_running(channel_id): return _make_response(code=409, message=f"Channel '{channel_id}' is already running") try: await manager.start_channel(channel_id) - return _make_response( - data={"channel_id": channel_id, "status": "connecting"}, - message="Channel start initiated", - ) + except ChannelAuthenticationError as e: + raise HTTPException(status_code=401, detail=f"凭证无效: {e}") + except ChannelNotConnectedError as e: + raise HTTPException(status_code=503, detail=f"渠道不可达: {e}") + except ChannelTimeoutError as e: + raise HTTPException(status_code=504, detail=f"连接超时: {e}") + except ChannelRateLimitError as e: + raise HTTPException(status_code=429, detail=f"上游限流: {e}") except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Failed to start channel {channel_id}: {e}") raise HTTPException(status_code=500, detail=f"Failed to start channel: {e}") + adapter = manager._adapters.get(channel_id) + current_status = manager._adapter_status(adapter) if adapter else "connecting" + + response_data = {"channel_id": channel_id, "status": current_status} + if adapter: + try: + snapshot = adapter.snapshot() + response_data.update( + {k: v for k, v in snapshot.items() if k in ("channel_type", "display_name", "capabilities", "health")} + ) + except Exception: + logger.warning(f"Failed to get snapshot for {channel_id}") + + return _make_response(data=response_data, message="Channel start initiated") + @channels.post("/{channel_id}/stop") async def stop_channel( @@ -185,16 +529,39 @@ async def restart_channel( if not manager.is_registered(channel_id): raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") + config = manager._channels_config.get(channel_id, {}) + if not config.get("enabled", False): + raise HTTPException( + status_code=400, + detail=f"Channel '{channel_id}' is disabled in configuration", + ) + + if not manager.is_running(channel_id): + raise HTTPException(status_code=409, detail=f"Channel '{channel_id}' is not running") + try: await manager.restart_channel(channel_id) - return _make_response( - data={"channel_id": channel_id, "status": "connecting"}, - message="Channel restart initiated", - ) + except ChannelAuthenticationError as e: + raise HTTPException(status_code=401, detail=f"凭证无效: {e}") + except ChannelNotConnectedError as e: + raise HTTPException(status_code=503, detail=f"渠道不可达: {e}") + except ChannelTimeoutError as e: + raise HTTPException(status_code=504, detail=f"连接超时: {e}") + except ChannelRateLimitError as e: + raise HTTPException(status_code=429, detail=f"上游限流: {e}") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Failed to restart channel {channel_id}: {e}") raise HTTPException(status_code=500, detail=f"Failed to restart channel: {e}") + adapter = manager._adapters.get(channel_id) + current_status = manager._adapter_status(adapter) if adapter else "connecting" + return _make_response( + data={"channel_id": channel_id, "status": current_status}, + message="Channel restart initiated", + ) + @channels.put("/{channel_id}/config") async def update_channel_config( @@ -202,18 +569,30 @@ async def update_channel_config( config_updates: dict[str, Any], current_user: User = Depends(get_admin_user), ): + await _check_rate_limit(str(current_user.id), "channel_config", 5, 10) + manager = get_channel_manager() if not manager.is_registered(channel_id): raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") try: - result = await manager.update_channel_config(channel_id, config_updates) - return _make_response( - data=result, - message="Configuration updated. Restart channel to apply changes.", - ) - except Exception as e: + result = await manager.update_channel_config(channel_id, config_updates, user_id=str(current_user.id)) + + if result["adapter_status"] == "stopped": + message = "配置已保存。启动渠道后生效。" + elif result.get("needs_restart"): + message = "配置已更新。建议重启渠道以确保所有内部状态同步。" + else: + message = "配置已实时生效。" + + return _make_response(data=result, message=message) + except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) + except ChannelException as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to update config for channel {channel_id}: {e}") + raise HTTPException(status_code=500, detail=f"Failed to update channel config: {e}") @channels.post("/{channel_id}/test") @@ -240,9 +619,8 @@ async def get_channel_stats( ): from datetime import timedelta - from sqlalchemy import Integer, func + from sqlalchemy import Integer - from yuxi.storage.postgres.models_channels import ChannelMsgRecord from yuxi.utils.datetime_utils import utc_now_naive if period: @@ -444,44 +822,48 @@ async def get_channel_messages( date_from: str | None = Query(None), date_to: str | None = Query(None), search: str | None = Query(None), + sort_by: str = Query("created_at"), + sort_order: str = Query("desc", pattern="^(asc|desc)$"), current_user: User = Depends(get_admin_user), db: AsyncSession = Depends(get_db), ): - from datetime import datetime + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException( + status_code=404, + detail=f"渠道 '{channel_id}' 未注册", + ) - from sqlalchemy import func, select, desc + await _check_channel_rate_limit( + channel_id=channel_id, + user_id=str(current_user.id), + ) - from yuxi.storage.postgres.models_channels import ChannelMsgRecord + conditions = _build_message_conditions( + channel_id=channel_id, + status_value=status, + content_type=content_type, + date_from=date_from, + date_to=date_to, + search=search, + ) - conditions = [ChannelMsgRecord.channel_id == channel_id] - - if status: - conditions.append(ChannelMsgRecord.status == status) - if content_type: - conditions.append(ChannelMsgRecord.content_type == content_type) - if date_from: - try: - dt_from = datetime.strptime(date_from, "%Y-%m-%d") - conditions.append(ChannelMsgRecord.created_at >= dt_from) - except ValueError: - pass - if date_to: - try: - dt_to = datetime.strptime(date_to, "%Y-%m-%d") - conditions.append(ChannelMsgRecord.created_at < dt_to) - except ValueError: - pass - if search: - conditions.append(ChannelMsgRecord.content_preview.ilike(f"%{search}%")) - - count_query = select(func.count()).where(*conditions) + count_query = select(func.count()).select_from(ChannelMsgRecord).where(*conditions) total_result = await db.execute(count_query) total = total_result.scalar() or 0 + sort_column = getattr(ChannelMsgRecord, sort_by, None) + if sort_column is None or sort_by not in SORTABLE_FIELDS: + raise HTTPException( + status_code=400, + detail=f"不支持的排序字段: '{sort_by}'，可选: {', '.join(sorted(SORTABLE_FIELDS))}", + ) + order_func = desc if sort_order == "desc" else asc + query = ( select(ChannelMsgRecord) .where(*conditions) - .order_by(desc(ChannelMsgRecord.created_at)) + .order_by(order_func(sort_column)) .offset((page - 1) * page_size) .limit(page_size) ) @@ -490,6 +872,8 @@ async def get_channel_messages( messages = [ { "id": r.id, + "message_id": r.message_id, + "channel_type": r.channel_type, "created_at": r.created_at.strftime("%Y-%m-%dT%H:%M:%SZ") if r.created_at else None, "sender_user_id": r.sender_user_id, "content_preview": r.content_preview, @@ -497,6 +881,8 @@ async def get_channel_messages( "status": r.status, "response_time_ms": r.response_time_ms, "reply_content_preview": r.reply_content_preview, + "reply_message_id": r.reply_message_id, + "replied_at": r.replied_at.strftime("%Y-%m-%dT%H:%M:%SZ") if r.replied_at else None, "error_message": r.error_message, "chat_id": r.chat_id, "chat_type": r.chat_type, @@ -518,79 +904,48 @@ async def export_channel_messages( current_user: User = Depends(get_admin_user), db: AsyncSession = Depends(get_db), ): - import csv - from datetime import datetime - from io import StringIO + await _check_rate_limit(str(current_user.id), "channel_messages_export", 5, 60) - from fastapi.responses import StreamingResponse - - from yuxi.storage.postgres.models_channels import ChannelMsgRecord - - conditions = [ChannelMsgRecord.channel_id == channel_id] - if status: - conditions.append(ChannelMsgRecord.status == status) - if content_type: - conditions.append(ChannelMsgRecord.content_type == content_type) - if date_from: - try: - dt_from = datetime.strptime(date_from, "%Y-%m-%d") - conditions.append(ChannelMsgRecord.created_at >= dt_from) - except ValueError: - pass - if date_to: - try: - dt_to = datetime.strptime(date_to, "%Y-%m-%d") - conditions.append(ChannelMsgRecord.created_at < dt_to) - except ValueError: - pass - if search: - conditions.append(ChannelMsgRecord.content_preview.ilike(f"%{search}%")) - - query = select(ChannelMsgRecord).where(*conditions).order_by(ChannelMsgRecord.created_at.desc()).limit(10000) - result = await db.execute(query) - rows = result.scalars().all() - - output = StringIO() - writer = csv.writer(output) - writer.writerow( - [ - "id", - "created_at", - "sender_user_id", - "content_preview", - "content_type", - "status", - "response_time_ms", - "reply_content_preview", - "error_message", - "chat_id", - "chat_type", - ] - ) - for r in rows: - writer.writerow( - [ - r.id, - r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else "", - r.sender_user_id, - r.content_preview, - r.content_type, - r.status, - r.response_time_ms or "", - r.reply_content_preview or "", - r.error_message or "", - r.chat_id, - r.chat_type, - ] + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException( + status_code=404, + detail=f"渠道 '{channel_id}' 未注册", ) - output.seek(0) - return StreamingResponse( - iter([output.getvalue()]), - media_type="text/csv; charset=utf-8-sig", - headers={"Content-Disposition": f"attachment; filename=messages_{channel_id}.csv"}, + conditions = _build_message_conditions( + channel_id=channel_id, + status_value=status, + content_type=content_type, + date_from=date_from, + date_to=date_to, + search=search, ) + try: + count_query = select(func.count()).select_from(ChannelMsgRecord).where(*conditions) + total = (await db.execute(count_query)).scalar() or 0 + + query = select(ChannelMsgRecord).where(*conditions).order_by(ChannelMsgRecord.created_at.desc()).limit(10000) + result = await db.execute(query) + rows = result.scalars().all() + + headers = { + "Content-Disposition": f"attachment; filename=messages_{channel_id}.csv", + "X-Total-Count": str(total), + } + if total > 10000: + headers["X-Truncated"] = "true" + + return StreamingResponse( + _UTF8SigEncoder(_csv_line_generator(rows)), + media_type="text/csv; charset=utf-8", + headers=headers, + ) + except Exception: + logger.exception(f"Failed to export messages for channel {channel_id}") + raise HTTPException(status_code=500, detail="导出消息失败，请稍后重试") + @channels.get("/{channel_id}/policy") async def get_channel_policy( @@ -598,70 +953,43 @@ async def get_channel_policy( current_user: User = Depends(get_admin_user), db: AsyncSession = Depends(get_db), ): - from yuxi.storage.postgres.models_channels import ChannelPolicyConfig + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") + + try: + result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) + policy = result.scalar_one_or_none() + except Exception: + logger.error(f"Failed to query policy for channel {channel_id}", exc_info=True) + raise HTTPException(status_code=500, detail="查询策略配置失败") - result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) - policy = result.scalar_one_or_none() if not policy: - return _make_response( - data={ - "channel_id": channel_id, - "group_chat_mode": "all", - "whitelist_ids": [], - "welcome_message": "", - "work_hours_start": "09:00", - "work_hours_end": "18:00", - "timezone_offset": 8, - "off_hours_reply": "", - } - ) + return _make_response(data={"channel_id": channel_id, **_POLICY_DEFAULTS}) return _make_response(data=policy.to_dict()) @channels.put("/{channel_id}/policy") async def update_channel_policy( channel_id: str, - policy_updates: dict[str, Any], + policy_updates: ChannelPolicyUpdate, current_user: User = Depends(get_admin_user), db: AsyncSession = Depends(get_db), ): - from yuxi.storage.postgres.models_channels import ChannelPolicyConfig - from yuxi.utils.datetime_utils import utc_now_naive + await _check_rate_limit(str(current_user.id), "channel_policy_update", 5, 5) + + updates_dict = policy_updates.model_dump(exclude_none=True) result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) policy = result.scalar_one_or_none() if policy: - for key in [ - "group_chat_mode", - "whitelist_ids", - "welcome_message", - "work_hours_start", - "work_hours_end", - "timezone_offset", - "off_hours_reply", - ]: - if key in policy_updates: - setattr(policy, key, policy_updates[key]) - policy.updated_at = utc_now_naive() + for key in _POLICY_FIELDS: + if key in updates_dict: + setattr(policy, key, updates_dict[key]) else: - policy = ChannelPolicyConfig( - channel_id=channel_id, - **{ - k: v - for k, v in policy_updates.items() - if k - in [ - "group_chat_mode", - "whitelist_ids", - "welcome_message", - "work_hours_start", - "work_hours_end", - "timezone_offset", - "off_hours_reply", - ] - }, - ) + policy_data = {k: v for k, v in updates_dict.items() if k in _POLICY_FIELDS} + policy = ChannelPolicyConfig(channel_id=channel_id, **policy_data) db.add(policy) await db.commit() @@ -675,13 +1003,13 @@ async def get_channel_routing( current_user: User = Depends(get_admin_user), db: AsyncSession = Depends(get_db), ): + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"渠道 '{channel_id}' 不存在") + from yuxi.storage.postgres.models_channels import ChannelRoutingRule - result = await db.execute( - select(ChannelRoutingRule) - .where(ChannelRoutingRule.channel_id == channel_id) - .order_by(ChannelRoutingRule.priority.desc()) - ) + result = await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id)) rules = result.scalars().all() return _make_response(data={"channel_id": channel_id, "routes": [r.to_dict() for r in rules]}) @@ -695,6 +1023,26 @@ async def update_channel_routing( ): from yuxi.storage.postgres.models_channels import ChannelRoutingRule + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"渠道 '{channel_id}' 不存在") + + if not routing_updates: + raise HTTPException(status_code=422, detail="不允许清空全部路由规则") + + from yuxi.repositories.agent_config_repository import AgentConfigRepository + + agent_repo = AgentConfigRepository(db) + for item in routing_updates: + aid = str(item.get("agent_config_id", "0")) + try: + aid_int = int(aid) + except (ValueError, TypeError): + raise HTTPException(status_code=422, detail=f"Agent 配置 ID '{aid}' 无效") + config = await agent_repo.get_by_id(aid_int) + if config is None: + raise HTTPException(status_code=422, detail=f"Agent 配置 '{aid}' 不存在") + existing = ( (await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id))) .scalars() @@ -703,17 +1051,30 @@ async def update_channel_routing( for rule in existing: await db.delete(rule) - for i, item in enumerate(routing_updates): + for item in routing_updates: rule = ChannelRoutingRule( channel_id=channel_id, command=item.get("command", ""), agent_config_id=str(item.get("agent_config_id", "0")), - priority=item.get("priority", len(routing_updates) - i), ) db.add(rule) await db.commit() - return _make_response(data={"channel_id": channel_id, "routes": routing_updates}, message="路由规则已保存") + + result = await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id)) + saved_rules = result.scalars().all() + return _make_response( + data={"channel_id": channel_id, "routes": [r.to_dict() for r in saved_rules]}, + message="路由规则已保存", + ) + + +ALL_CONFIG_GROUPS = { + "credentials": {"key": "credentials", "label": "认证凭据"}, + "general": {"key": "general", "label": "通用设置"}, + "routing": {"key": "routing", "label": "路由配置"}, + "advanced": {"key": "advanced", "label": "高级设置"}, +} @channels.get("/{channel_id}/config-schema") @@ -721,60 +1082,65 @@ async def get_channel_config_schema( channel_id: str, current_user: User = Depends(get_admin_user), ): - from yuxi.channels.registry import BUILTIN_ADAPTERS + if not re.match(r"^[a-z0-9][a-z0-9_-]{0,30}$", channel_id): + raise HTTPException(status_code=400, detail=f"Invalid channel_id: '{channel_id}'") - all_adapters = {**BUILTIN_ADAPTERS} - for cid, adapter_cls in all_adapters.items(): - if cid == channel_id: - try: - default_config = adapter_cls.default_config() if hasattr(adapter_cls, "default_config") else {} - except Exception: - default_config = {} + await _check_rate_limit(str(current_user.id), "config_schema", 10, 5) - fields = [] - for key, value in default_config.items(): - field_type = "text" - if "token" in key or "secret" in key or "key" in key: - field_type = "password" - elif isinstance(value, bool): - field_type = "switch" - elif isinstance(value, (int, float)): - field_type = "number" - is_credential = any(k in key for k in ("token", "secret", "key", "app_id")) - fields.append( - { - "key": key, - "label": key.replace("_", " ").title(), - "type": field_type, - "default": value, - "required": "token" in key or "secret" in key, - "group": "credentials" if is_credential else "general", - } - ) + manager = get_channel_manager() + adapter_cls = manager._registry.get(channel_id) + if adapter_cls is None: + return _make_response(data={"channel_type": channel_id, "fields": [], "groups": []}) - fields.append( - { - "key": "agent_config_id", - "label": "默认 Agent", - "type": "agent_select", - "required": False, - "group": "routing", - } - ) + schema = adapter_cls.get_config_schema() - return _make_response( - data={ - "channel_type": channel_id, - "fields": fields, - "groups": [ - {"key": "credentials", "label": "认证凭据"}, - {"key": "general", "label": "通用设置"}, - {"key": "routing", "label": "路由配置"}, - ], - } - ) + if not schema: + logger.info(f"渠道 {channel_id} 未定义 config_schema，返回仅含 agent_config_id 的基本 Schema") - return _make_response(data={"channel_type": channel_id, "fields": [], "groups": []}) + fields = [] + used_groups: set[str] = set() + + for key, schema_def in schema.items(): + field_type = schema_def.get("type", "text") + group = schema_def.get("group", "general") + used_groups.add(group) + + field_entry: dict[str, Any] = { + "key": key, + "label": schema_def.get("label") or key.replace("_", " ").title(), + "type": field_type, + "required": schema_def.get("required", False), + "group": group, + } + if "default" in schema_def: + field_entry["default"] = schema_def["default"] + if "enum" in schema_def: + field_entry["enum"] = schema_def["enum"] + if schema_def.get("secret"): + field_entry["secret"] = True + + fields.append(field_entry) + + fields.append( + { + "key": "agent_config_id", + "label": "默认 Agent", + "type": "agent_select", + "required": False, + "group": "routing", + } + ) + used_groups.add("routing") + + groups = [ALL_CONFIG_GROUPS[g] for g in sorted(used_groups) if g in ALL_CONFIG_GROUPS] + + return _make_response( + data={ + "channel_type": channel_id, + "fields": fields, + "groups": groups, + } + ) @channels.get("/{channel_id}/chats/{chat_id}") @@ -809,14 +1175,161 @@ async def get_channel_chat_mapping( ) +@channels.post("/state/cleanup") +async def cleanup_expired_state( + store: PluginStateStore = Depends(get_state_store), + current_user: User = Depends(get_admin_user), +): + count = await store.cleanup_expired() + return _make_response(data={"cleaned": count}, message=f"清理了 {count} 条过期状态") + + +@channels.get("/{channel_id}/state") +async def list_channel_state( + channel_id: str, + store: PluginStateStore = Depends(get_state_store), + current_user: User = Depends(get_admin_user), +): + await _check_rate_limit(str(current_user.id), "state_read", 30, 10) + + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found") + + all_state = await store.get_all(channel_id) + + safe_state = {} + for ns, entries in all_state.items(): + if ns == "credentials": + safe_state[ns] = {k: {"redacted": True, "has_credential": True} for k in entries} + else: + safe_state[ns] = entries + + return _make_response(data={"channel_id": channel_id, "state": safe_state}) + + +@channels.get("/{channel_id}/state/{namespace}/{key}") +async def get_channel_state_entry( + channel_id: str, + namespace: str = Path(..., min_length=1, max_length=64), + key: str = Path(..., min_length=1, max_length=128), + store: PluginStateStore = Depends(get_state_store), + current_user: User = Depends(get_admin_user), +): + await _check_rate_limit(str(current_user.id), "state_read", 30, 10) + + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found") + + value = await store.get(channel_id, key, namespace) + if value is None: + raise HTTPException(status_code=404, detail="State entry not found") + + if namespace == "credentials": + return _make_response( + data={ + "channel_id": channel_id, + "namespace": namespace, + "key": key, + "redacted": True, + "has_credential": True, + } + ) + + return _make_response(data={"channel_id": channel_id, "namespace": namespace, "key": key, "value": value}) + + +@channels.put("/{channel_id}/state/{namespace}/{key}") +async def update_channel_state_entry( + body: UpdateStateEntryBody, + channel_id: str = Path(..., min_length=1, max_length=32), + namespace: str = Path(..., min_length=1, max_length=64), + key: str = Path(..., min_length=1, max_length=128), + store: PluginStateStore = Depends(get_state_store), + current_user: User = Depends(get_admin_user), +): + await _check_rate_limit(str(current_user.id), "state_write", 10, 10) + + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found") + + logger.info( + "渠道状态写入: user=%s, channel=%s, namespace=%s, key=%s, has_ttl=%s", + current_user.id, + channel_id, + namespace, + key, + body.ttl_seconds is not None, + ) + + try: + is_new = await store.set( + channel_id, + key, + body.value, + namespace, + ttl_seconds=body.ttl_seconds, + ) + except IntegrityError: + raise HTTPException(status_code=409, detail="状态条目写入冲突，请重试") + except DBAPIError as e: + logger.error( + "更新渠道状态失败 (channel=%s, ns=%s, key=%s): %s", + channel_id, + namespace, + key, + e, + exc_info=True, + ) + raise HTTPException(status_code=500, detail="数据库操作失败，请稍后重试") + except Exception as e: + logger.error("更新渠道状态失败 (channel=%s): %s", channel_id, e, exc_info=True) + raise HTTPException(status_code=500, detail="更新渠道状态失败") + + status_code = 201 if is_new else 200 + message = "状态已创建" if is_new else "状态已更新" + return JSONResponse( + content=_make_response( + data={"channel_id": channel_id, "namespace": namespace, "key": key}, + message=message, + ), + status_code=status_code, + ) + + +@channels.delete("/{channel_id}/state/{namespace}/{key}") +async def delete_channel_state_entry( + channel_id: str, + namespace: str = Path(..., min_length=1, max_length=64), + key: str = Path(..., min_length=1, max_length=128), + store: PluginStateStore = Depends(get_state_store), + current_user: User = Depends(get_admin_user), +): + await _check_rate_limit(str(current_user.id), "state_write", 10, 10) + + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found") + + await store.delete(channel_id, key, namespace) + return _make_response( + data={"channel_id": channel_id, "namespace": namespace, "key": key}, + message="状态已删除", + ) + + @channels.delete("/{channel_id}") async def unregister_channel( channel_id: str, current_user: User = Depends(get_admin_user), ): + await _check_rate_limit(str(current_user.id), "channel_unregister", 5, 10) + manager = get_channel_manager() if not manager.is_registered(channel_id): - raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' not found") + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") await manager.unregister_channel(channel_id) return _make_response(data={"channel_id": channel_id, "unregistered": True}, message="渠道已注销") diff --git a/backend/server/routers/ws_chat_router.py b/backend/server/routers/ws_chat_router.py index e09edc7d..d642a8a4 100644 --- a/backend/server/routers/ws_chat_router.py +++ b/backend/server/routers/ws_chat_router.py @@ -89,11 +89,13 @@ async def websocket_chat(ws: WebSocket, client_id: str, token: str = Query(None) return await ws.accept() - await ws.send_json({ - "type": "connected", - "payload": {"client_id": client_id, "user_id": user_id}, - "timestamp": _now_iso(), - }) + await ws.send_json( + { + "type": "connected", + "payload": {"client_id": client_id, "user_id": user_id}, + "timestamp": _now_iso(), + } + ) last_msg_time = time.monotonic() rate_bucket: list[float] = [] @@ -139,11 +141,13 @@ async def websocket_chat(ws: WebSocket, client_id: str, token: str = Query(None) now = time.monotonic() rate_bucket = [t for t in rate_bucket if now - t < 1.0] if len(rate_bucket) >= MAX_MSG_PER_SEC: - await ws.send_json({ - "type": "error", - "payload": {"code": 429, "message": "Rate limit exceeded"}, - "timestamp": _now_iso(), - }) + await ws.send_json( + { + "type": "error", + "payload": {"code": 429, "message": "Rate limit exceeded"}, + "timestamp": _now_iso(), + } + ) continue rate_bucket.append(now) @@ -173,31 +177,37 @@ async def websocket_chat(ws: WebSocket, client_id: str, token: str = Query(None) ) request_id = identity.channel_message_id - await ws.send_json({ - "type": "typing", - "payload": {"status": "agent", "request_id": request_id}, - "timestamp": _now_iso(), - }) + await ws.send_json( + { + "type": "typing", + "payload": {"status": "agent", "request_id": request_id}, + "timestamp": _now_iso(), + } + ) try: await get_channel_manager().dispatch_inbound(message) except Exception as e: logger.error(f"WS message processing error: {e}") - await ws.send_json({ - "type": "error", - "payload": { - "request_id": request_id, - "code": 500, - "message": f"处理出错: {str(e)}", - }, - "timestamp": _now_iso(), - }) + await ws.send_json( + { + "type": "error", + "payload": { + "request_id": request_id, + "code": 500, + "message": f"处理出错: {str(e)}", + }, + "timestamp": _now_iso(), + } + ) - await ws.send_json({ - "type": "done", - "payload": {"request_id": request_id}, - "timestamp": _now_iso(), - }) + await ws.send_json( + { + "type": "done", + "payload": {"request_id": request_id}, + "timestamp": _now_iso(), + } + ) except WebSocketDisconnect: logger.info(f"WS client {client_id} disconnected") diff --git a/web/src/apis/channel_api.js b/web/src/apis/channel_api.js index 0efbb948..8088505a 100644 --- a/web/src/apis/channel_api.js +++ b/web/src/apis/channel_api.js @@ -185,7 +185,14 @@ export const channelApi = { getUserMapping: (channelId, userId) => apiGet(`/api/channels/${channelId}/users/${userId}`), - getChatMapping: (channelId, chatId) => apiGet(`/api/channels/${channelId}/chats/${chatId}`), + getChatMapping: (channelId, chatId, internalUserId) => { + const params = internalUserId ? { internal_user_id: internalUserId } : {} + return apiGet(`/api/channels/${channelId}/chats/${chatId}`, { params }) + }, + + getCredentialStatus: (channelId) => apiGet(`/api/channels/${channelId}/credential-status`), + + refreshCredential: (channelId) => apiAdminPost(`/api/channels/${channelId}/refresh-credential`), getDashboardChannelStats: () => apiGet('/api/channels/status') } diff --git a/web/src/components/channels/ChannelMappingBrowser.vue b/web/src/components/channels/ChannelMappingBrowser.vue index 66677067..4b7e6c39 100644 --- a/web/src/components/channels/ChannelMappingBrowser.vue +++ b/web/src/components/channels/ChannelMappingBrowser.vue @@ -15,6 +15,7 @@ const userResult = ref(null) const userLoading = ref(false) const chatSearchId = ref('') +const chatInternalUserId = ref('') const chatResult = ref(null) const chatLoading = ref(false) @@ -50,7 +51,7 @@ async function searchChat() { chatLoading.value = true chatResult.value = null try { - const res = await channelApi.getChatMapping(props.channelId, cid) + const res = await channelApi.getChatMapping(props.channelId, cid, chatInternalUserId.value || undefined) if (seq !== searchChatSeq) return chatResult.value = res.data } catch (e) { @@ -137,6 +138,15 @@ async function searchChat() { 搜索 +
+ +
搜索中...
diff --git a/web/src/components/channels/ChannelStatusCard.vue b/web/src/components/channels/ChannelStatusCard.vue index 721a2efe..ca89588d 100644 --- a/web/src/components/channels/ChannelStatusCard.vue +++ b/web/src/components/channels/ChannelStatusCard.vue @@ -1,6 +1,6 @@