From bba17752208e7e8d16eeeda5a9d0aa396efc83d9 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Mon, 6 Jul 2026 20:50:03 +0800 Subject: [PATCH] =?UTF-8?q?refactor(channel-router):=20=E6=B8=85=E7=90=86?= =?UTF-8?q?=E5=86=97=E4=BD=99=E7=A9=BA=E8=A1=8C=E5=B9=B6=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本次提交包含多类优化： 1. 移除多个路由文件中多余的空导入行，统一代码格式 2. 重构Query参数定义，将长参数拆分为多行提升可读性 3. 新增多个业务端点： - 渠道能力画像矩阵查询CAP-03 - 配对审批计数接口用于待办角标 - 批量查询对端目录资料接口 - 向导扫码登录相关端点 - 会话实时事件SSE推送端点 - 工作台待办统计接口 4. 完善异常处理逻辑，补充OperationTimeoutError导入并优化NotImplementedError的细节返回 5. 调整路由导入顺序，修复动态路由路径冲突隐患 6. 更新文档注释与接口清单，修正部分接口描述细节 --- backend/server/routers/channels/__init__.py | 55 +++++++------ .../server/routers/channels/account_router.py | 1 - .../routers/channels/allowlist_router.py | 5 +- .../server/routers/channels/audit_router.py | 1 - .../routers/channels/capability_router.py | 22 +++++- .../server/routers/channels/config_router.py | 1 - .../routers/channels/content_review_router.py | 5 +- .../routers/channels/dashboard_router.py | 38 ++++++++- .../routers/channels/directory_router.py | 56 ++++++++++++- .../server/routers/channels/doctor_router.py | 1 - .../server/routers/channels/health_router.py | 25 ++---- .../server/routers/channels/login_router.py | 1 - .../server/routers/channels/message_router.py | 1 - .../server/routers/channels/outbox_router.py | 18 ++--- .../server/routers/channels/pairing_router.py | 38 ++++++++- .../server/routers/channels/plugin_router.py | 5 +- .../server/routers/channels/session_router.py | 79 +++++++++++++++++-- .../server/routers/channels/wizard_router.py | 58 +++++++++++++- 18 files changed, 325 insertions(+), 85 deletions(-) diff --git a/backend/server/routers/channels/__init__.py b/backend/server/routers/channels/__init__.py index d0e09f91..10da9ce1 100644 --- a/backend/server/routers/channels/__init__.py +++ b/backend/server/routers/channels/__init__.py @@ -17,7 +17,6 @@ from typing import Any, Literal from fastapi import APIRouter, Depends, Request from sqlalchemy.ext.asyncio import AsyncSession - from yuxi.channels.contract.dtos.common import Operator, OperatorRole from yuxi.channels.contract.dtos.control import ControlResult from yuxi.channels.contract.errors import ( @@ -35,6 +34,7 @@ from yuxi.channels.contract.errors import ( LifecycleHookError, NotFoundError, NotImplementedError, + OperationTimeoutError, PairingExpiredError, PermissionDeniedError, PipelineConfigError, @@ -44,7 +44,6 @@ from yuxi.channels.contract.errors import ( RateLimitError, RuleViolationError, SchemaInitializationError, - OperationTimeoutError, ValidationError, ) from yuxi.channels.infrastructure import DependencyInjectionContainer @@ -351,7 +350,11 @@ def raiseOnControlFailure(result: ControlResult) -> None: message=message, ) if code == "NOT_IMPLEMENTED": - raise NotImplementedError(operation=details.get("operation", ""), message=message) + raise NotImplementedError( + operation=details.get("operation", ""), + message=message, + details={k: v for k, v in details.items() if k != "operation"}, + ) raise InternalError(message=message) @@ -406,23 +409,23 @@ channels_router = APIRouter(prefix="/channels", tags=["channels"]) # 注册顺序：静态路径 router（如 health_router）必须先于含 /{channel_type}/... # 动态路径的 router 注册，避免 /channels/health 被捕获为 channel_type="health" # （规范 §6.5）。 -from server.routers.channels.health_router import health_router +from server.routers.channels.health_router import health_router # noqa: E402 channels_router.include_router(health_router) -from server.routers.channels.audit_router import audit_router +from server.routers.channels.audit_router import audit_router # noqa: E402 channels_router.include_router(audit_router) -from server.routers.channels.analytics_router import analytics_router +from server.routers.channels.analytics_router import analytics_router # noqa: E402 channels_router.include_router(analytics_router) -from server.routers.channels.account_router import account_router -from server.routers.channels.session_router import session_router -from server.routers.channels.login_router import login_router -from server.routers.channels.directory_router import directory_router -from server.routers.channels.pairing_router import pairing_router +from server.routers.channels.account_router import account_router # noqa: E402 +from server.routers.channels.directory_router import directory_router # noqa: E402 +from server.routers.channels.login_router import login_router # noqa: E402 +from server.routers.channels.pairing_router import pairing_router # noqa: E402 +from server.routers.channels.session_router import session_router # noqa: E402 channels_router.include_router(account_router) channels_router.include_router(session_router) @@ -430,46 +433,46 @@ channels_router.include_router(login_router) channels_router.include_router(directory_router) channels_router.include_router(pairing_router) -from server.routers.channels.allowlist_router import allowlist_router +from server.routers.channels.allowlist_router import allowlist_router # noqa: E402 channels_router.include_router(allowlist_router) -from server.routers.channels.message_router import message_router +from server.routers.channels.message_router import message_router # noqa: E402 channels_router.include_router(message_router) -from server.routers.channels.outbox_router import outbox_router -from server.routers.channels.webhook_router import webhook_router -from server.routers.channels.config_router import config_router +from server.routers.channels.config_router import config_router # noqa: E402 +from server.routers.channels.outbox_router import outbox_router # noqa: E402 +from server.routers.channels.webhook_router import webhook_router # noqa: E402 channels_router.include_router(outbox_router) channels_router.include_router(webhook_router) channels_router.include_router(config_router) -from server.routers.channels.capability_router import capability_router +from server.routers.channels.capability_router import capability_router # noqa: E402 channels_router.include_router(capability_router) -from server.routers.channels.doctor_router import doctor_router +from server.routers.channels.doctor_router import doctor_router # noqa: E402 channels_router.include_router(doctor_router) -from server.routers.channels.plugin_router import plugin_router +from server.routers.channels.plugin_router import plugin_router # noqa: E402 channels_router.include_router(plugin_router) -from server.routers.channels.content_review_router import content_review_router +from server.routers.channels.content_review_router import content_review_router # noqa: E402 channels_router.include_router(content_review_router) -from server.routers.channels.dashboard_router import dashboard_router +from server.routers.channels.dashboard_router import dashboard_router # noqa: E402 channels_router.include_router(dashboard_router) -from server.routers.channels.reports_router import reports_router - -channels_router.include_router(reports_router) - -from server.routers.channels.wizard_router import wizard_router +from server.routers.channels.wizard_router import wizard_router # noqa: E402 channels_router.include_router(wizard_router) + +from server.routers.channels.route_binding_router import route_binding_router # noqa: E402 + +channels_router.include_router(route_binding_router) diff --git a/backend/server/routers/channels/account_router.py b/backend/server/routers/channels/account_router.py index ea497cf3..b66d07f6 100644 --- a/backend/server/routers/channels/account_router.py +++ b/backend/server/routers/channels/account_router.py @@ -38,7 +38,6 @@ from typing import Any from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field, model_validator - from yuxi.channels.contract.dtos.channel import ( BatchStateChangeCmd, ChannelType, diff --git a/backend/server/routers/channels/allowlist_router.py b/backend/server/routers/channels/allowlist_router.py index 7899820d..c58f0587 100644 --- a/backend/server/routers/channels/allowlist_router.py +++ b/backend/server/routers/channels/allowlist_router.py @@ -36,7 +36,6 @@ from typing import Any from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.whitelist import ( UNSET, @@ -313,9 +312,7 @@ async def export_whitelist( account_id: str, policy_type: str, request: Request, - export_format: str = Query( - default="json", alias="format", description="导出格式：json / csv" - ), + export_format: str = Query(default="json", alias="format", description="导出格式：json / csv"), use_cases=Depends(get_channel_use_cases), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: diff --git a/backend/server/routers/channels/audit_router.py b/backend/server/routers/channels/audit_router.py index 137685b0..9f7d6d43 100644 --- a/backend/server/routers/channels/audit_router.py +++ b/backend/server/routers/channels/audit_router.py @@ -52,7 +52,6 @@ from zoneinfo import ZoneInfo from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.audit import ( AuditOperationType, AuditQuery, diff --git a/backend/server/routers/channels/capability_router.py b/backend/server/routers/channels/capability_router.py index 1e345bbb..2aab2e47 100644 --- a/backend/server/routers/channels/capability_router.py +++ b/backend/server/routers/channels/capability_router.py @@ -1,4 +1,4 @@ -"""能力查询路由（CAP-01~CAP-02）。 +"""能力查询路由（CAP-01~CAP-03）。 统一采用模板 A（控制面端口路由）：鉴权依赖 get_admin_user， 调用 use_cases.capability_query.，raiseOnControlFailure 转译失败， @@ -11,7 +11,6 @@ from __future__ import annotations from typing import Any from fastapi import APIRouter, Depends, Request - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.storage.postgres.models_business import User @@ -45,6 +44,25 @@ async def list_capabilities( return {"success": True, "data": serialize_control_data(result.data)} +@capability_router.get("/capabilities/matrix", response_model=dict) +async def list_capability_matrix( + request: Request, + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """列出所有渠道能力画像矩阵（CAP-03）。对应控制面操作 capability/matrix。 + + 返回每个插件的完整 ``ChannelCapabilityProfile``，包含静态能力声明、 + 运行时可用性、插件元数据与配置 schema，供能力查询页渲染高密度矩阵。 + """ + operator = build_operator(current_user, request) + result = await use_cases.capability_query.listCapabilityMatrix( + operator=operator, + ) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)} + + @capability_router.get("/capabilities/{channel_type}", response_model=dict) async def get_capability( channel_type: ChannelType, diff --git a/backend/server/routers/channels/config_router.py b/backend/server/routers/channels/config_router.py index f41bf989..a0065d77 100644 --- a/backend/server/routers/channels/config_router.py +++ b/backend/server/routers/channels/config_router.py @@ -43,7 +43,6 @@ from typing import Any from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.config import ( BatchUpdateConfigCmd, ConfigExportCmd, diff --git a/backend/server/routers/channels/content_review_router.py b/backend/server/routers/channels/content_review_router.py index 2c1546bc..55ecba29 100644 --- a/backend/server/routers/channels/content_review_router.py +++ b/backend/server/routers/channels/content_review_router.py @@ -64,7 +64,6 @@ from typing import Any, Literal from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.content_review import ( BatchReviewDecisionCmd, @@ -141,9 +140,7 @@ class BatchReviewDecisionRequest(BaseModel): decisions: list[ReviewDecisionItemRequest] = Field( ..., min_length=1, max_length=100, description="决定条目列表（1-100）" ) - apply_to_pending_messages: bool = Field( - default=False, description="是否关联处理待审消息（本期预留）" - ) + apply_to_pending_messages: bool = Field(default=False, description="是否关联处理待审消息（本期预留）") def _parse_verdict(raw: str) -> ContentReviewVerdict: diff --git a/backend/server/routers/channels/dashboard_router.py b/backend/server/routers/channels/dashboard_router.py index 615d7a58..9f01d678 100644 --- a/backend/server/routers/channels/dashboard_router.py +++ b/backend/server/routers/channels/dashboard_router.py @@ -47,15 +47,27 @@ dashboard_router = APIRouter(tags=["channels-dashboard"]) @dashboard_router.get("/dashboard/overview", response_model=dict) async def get_dashboard_overview( request: Request, + channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), + start_time: str | None = Query(default=None, description="消息统计起始时间（ISO 8601）；仅 messages 子域受影响"), + end_time: str | None = Query(default=None, description="消息统计截止时间（ISO 8601）；仅 messages 子域受影响"), use_cases=Depends(get_channel_use_cases), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """查询全局总览（DSB-01）。对应控制面操作 dashboard/overview。 聚合账户 / 消息 / 会话 / 投递四类子域 KPI 计数。 + ``start_time`` / ``end_time`` 仅影响消息统计；账户、会话、投递为当前 + 快照，不受时间范围影响。 """ operator = build_operator(current_user, request) - result = await use_cases.dashboard_view.getDashboardOverview(operator=operator) + start_time_dt = parse_datetime("start_time", start_time) + end_time_dt = parse_datetime("end_time", end_time) + result = await use_cases.dashboard_view.getDashboardOverview( + channel_type=channel_type, + start_time=start_time_dt.isoformat() if start_time_dt else None, + end_time=end_time_dt.isoformat() if end_time_dt else None, + operator=operator, + ) raiseOnControlFailure(result) return {"success": True, "data": serialize_control_data(result.data)} @@ -93,7 +105,9 @@ async def get_dashboard_messages( 返回指定时间范围内的消息按渠道 / 角色 / 投递状态分组的计数快照。 - 时间参数由 parse_datetime 在 Router 层翻译为 UTC datetime，再转回 ISO 8601 字符串传给端口（端口签名为 str）。非法格式抛 ValidationError（400）。 + 时间参数由 parse_datetime 在 Router 层翻译为 UTC datetime，再转回 + ISO 8601 字符串传给端口（端口签名为 str）。非法格式抛 + ValidationError（400）。 """ operator = build_operator(current_user, request) start_time_dt = parse_datetime("start_time", start_time) @@ -158,6 +172,26 @@ async def get_dashboard_delivery( return {"success": True, "data": serialize_control_data(result.data)} +@dashboard_router.get("/dashboard/todos", response_model=dict) +async def get_dashboard_todos( + request: Request, + channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """查询工作台待办计数（DSB-TODOS）。对应控制面操作 dashboard/todos。 + + 聚合待审批配对数、待复核内容审核数与当前死信积压数，按渠道类型过滤。 + """ + operator = build_operator(current_user, request) + result = await use_cases.dashboard_view.getDashboardTodos( + channel_type=channel_type, + operator=operator, + ) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)} + + @dashboard_router.get("/dashboard/realtime", response_model=dict) async def get_dashboard_realtime( request: Request, diff --git a/backend/server/routers/channels/directory_router.py b/backend/server/routers/channels/directory_router.py index d6900bd9..3aaf6274 100644 --- a/backend/server/routers/channels/directory_router.py +++ b/backend/server/routers/channels/directory_router.py @@ -27,10 +27,14 @@ from __future__ import annotations from typing import Any -from fastapi import APIRouter, Depends, Query, Request - +from fastapi import APIRouter, Body, Depends, Query, Request +from pydantic import BaseModel, ConfigDict, Field from yuxi.channels.contract.dtos.channel import ChannelType -from yuxi.channels.contract.dtos.directory import ClearDirectoryCacheCmd, DirectoryQuery +from yuxi.channels.contract.dtos.directory import ( + BatchDirectoryProfilesQuery, + ClearDirectoryCacheCmd, + DirectoryQuery, +) from yuxi.storage.postgres.models_business import User from server.routers.channels import ( @@ -143,6 +147,52 @@ async def get_directory_user_profile( return {"success": True, "data": serialize_control_data(result.data)} +class BatchProfilesRequest(BaseModel): + """批量目录资料查询请求。""" + + model_config = ConfigDict(frozen=True) + + peer_ids: list[str] = Field( + ..., + min_length=1, + description="待查询的对端 ID 列表（可能是用户 ID 或群组 ID）", + ) + + +@directory_router.post( + "/{channel_type}/{account_id}/directory/batch_profiles", + response_model=dict, +) +async def batch_directory_profiles( + channel_type: ChannelType, + account_id: str, + request: Request, + payload: BatchProfilesRequest = Body(...), + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """批量查询对端资料（用户或群组）。 + + 给定一组 peer_id，逐个尝试查询用户资料；若用户不存在，则尝试查询 + 群组详情。结果统一返回为目录条目（type 区分 user/group），用于路由 + 绑定列表等需要把抽象 ID 翻译为可读名称的场景。5min TTL 缓存。 + + 对应控制面操作 ``directory/batch_profiles``（由 + ``ChannelControlService.batchGetDirectoryProfiles`` 内部构造 + ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。 + """ + operator = build_operator(current_user, request) + query = BatchDirectoryProfilesQuery(peer_ids=tuple(payload.peer_ids)) + result = await use_cases.directory.batchGetDirectoryProfiles( + channel_type=channel_type, + account_id=account_id, + query=query, + operator=operator, + ) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)} + + @directory_router.get( "/{channel_type}/{account_id}/directory/groups/{group_id}", response_model=dict, diff --git a/backend/server/routers/channels/doctor_router.py b/backend/server/routers/channels/doctor_router.py index 0e81c35b..c50e0e05 100644 --- a/backend/server/routers/channels/doctor_router.py +++ b/backend/server/routers/channels/doctor_router.py @@ -29,7 +29,6 @@ from __future__ import annotations from typing import Any from fastapi import APIRouter, Depends, Request - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.storage.postgres.models_business import User diff --git a/backend/server/routers/channels/health_router.py b/backend/server/routers/channels/health_router.py index 9d61ab66..a31ca9ea 100644 --- a/backend/server/routers/channels/health_router.py +++ b/backend/server/routers/channels/health_router.py @@ -41,7 +41,6 @@ from typing import Any, Literal from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.health import ( DiagnosticsExportRequest, @@ -110,12 +109,8 @@ async def get_health( @health_router.get("/health/detail", response_model=dict) async def get_health_detail( - channel_type: ChannelType | None = Query( - default=None, description="按渠道过滤（None 表示全渠道）" - ), - with_probe: bool = Query( - default=False, description="是否绕过缓存触发实时聚合（非调用 probeChannel）" - ), + channel_type: ChannelType | None = Query(default=None, description="按渠道过滤（None 表示全渠道）"), + with_probe: bool = Query(default=False, description="是否绕过缓存触发实时聚合（非调用 probeChannel）"), use_cases=Depends(get_channel_use_cases), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: @@ -157,17 +152,11 @@ class DiagnosticsExportPayload(BaseModel): model_config = ConfigDict(frozen=True) - channel_filter: ChannelType | None = Field( - default=None, description="按渠道过滤（None 表示全渠道）" - ) + channel_filter: ChannelType | None = Field(default=None, description="按渠道过滤（None 表示全渠道）") include_audit_logs: bool = Field(default=True, description="是否包含审计日志") - audit_log_count: int = Field( - default=100, ge=1, le=500, description="审计日志条数上限" - ) + audit_log_count: int = Field(default=100, ge=1, le=500, description="审计日志条数上限") include_error_logs: bool = Field(default=True, description="是否包含错误日志") - error_log_count: int = Field( - default=100, ge=1, le=500, description="错误日志条数上限" - ) + error_log_count: int = Field(default=100, ge=1, le=500, description="错误日志条数上限") # ---------------- Endpoints ---------------- @@ -272,9 +261,7 @@ async def get_single_health( channel_type=channel_type, account_id=account_id, ) - result = await use_cases.health_check_control.getSingleHealth( - query, operator=operator - ) + result = await use_cases.health_check_control.getSingleHealth(query, operator=operator) raiseOnControlFailure(result) return {"success": True, "data": serialize_control_data(result.data)} diff --git a/backend/server/routers/channels/login_router.py b/backend/server/routers/channels/login_router.py index eecb1531..ce95ace5 100644 --- a/backend/server/routers/channels/login_router.py +++ b/backend/server/routers/channels/login_router.py @@ -42,7 +42,6 @@ from typing import Any from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.login import ForceLogoutCmd from yuxi.storage.postgres.models_business import User diff --git a/backend/server/routers/channels/message_router.py b/backend/server/routers/channels/message_router.py index e67543ae..fa4d4bb7 100644 --- a/backend/server/routers/channels/message_router.py +++ b/backend/server/routers/channels/message_router.py @@ -37,7 +37,6 @@ from typing import Any, Literal from fastapi import APIRouter, Depends, File, Form, Header, Query, Request, UploadFile from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.admin import AdminSendCmd from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.common import MessageContent diff --git a/backend/server/routers/channels/outbox_router.py b/backend/server/routers/channels/outbox_router.py index a3a8325f..063bdb21 100644 --- a/backend/server/routers/channels/outbox_router.py +++ b/backend/server/routers/channels/outbox_router.py @@ -89,7 +89,9 @@ async def list_outbox_messages( channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 过滤"), created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"), created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"), - retry_count_min: int | None = Query(default=None, ge=0, description="最小重试次数下界（含），用于筛选已发生重试的条目"), + retry_count_min: int | None = Query( + default=None, ge=0, description="最小重试次数下界（含），用于筛选已发生重试的条目" + ), limit: int = Query(default=100, ge=1, le=200, description="分页大小"), offset: int = Query(default=0, ge=0, description="分页偏移"), use_cases=Depends(get_channel_use_cases), @@ -124,6 +126,7 @@ async def list_outbox_messages( async def get_outbox_stats( request: Request, channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), + channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤"), use_cases=Depends(get_channel_use_cases), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: @@ -132,6 +135,7 @@ async def get_outbox_stats( result = await use_cases.outbox_management.getOutboxStats( channel_type, operator=operator, + channel_account_id=channel_account_id, ) raiseOnControlFailure(result) return {"success": True, "data": serialize_control_data(result.data)} @@ -230,9 +234,7 @@ async def batch_retry_dead_letter( """ operator = build_operator(current_user, request) outbox_ids = cmd.outbox_ids if cmd else None - result = await use_cases.outbox_management.batchRetryDeadLetter( - operator=operator, outbox_ids=outbox_ids - ) + result = await use_cases.outbox_management.batchRetryDeadLetter(operator=operator, outbox_ids=outbox_ids) raiseOnControlFailure(result) return {"success": True, "data": serialize_control_data(result.data)} @@ -252,9 +254,7 @@ async def batch_delete_dead_letter( """ operator = build_operator(current_user, request) outbox_ids = cmd.outbox_ids if cmd else None - result = await use_cases.outbox_management.batchDeleteDeadLetter( - operator=operator, outbox_ids=outbox_ids - ) + result = await use_cases.outbox_management.batchDeleteDeadLetter(operator=operator, outbox_ids=outbox_ids) raiseOnControlFailure(result) return {"success": True, "data": serialize_control_data(result.data)} @@ -265,9 +265,7 @@ async def export_dead_letter( channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"), created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"), - export_format: Literal["json", "csv"] = Query( - default="json", alias="format", description="导出格式（json / csv）" - ), + export_format: Literal["json", "csv"] = Query(default="json", alias="format", description="导出格式（json / csv）"), limit: int = Query(default=10000, ge=1, le=100000, description="导出条目上限，防止大规模死信队列 OOM"), use_cases=Depends(get_channel_use_cases), current_user: User = Depends(get_admin_user), diff --git a/backend/server/routers/channels/pairing_router.py b/backend/server/routers/channels/pairing_router.py index 098c4501..4965f435 100644 --- a/backend/server/routers/channels/pairing_router.py +++ b/backend/server/routers/channels/pairing_router.py @@ -23,9 +23,8 @@ from __future__ import annotations from typing import Any -from fastapi import APIRouter, Body, Depends, Path, Query, Request +from fastapi import APIRouter, Body, Depends, Query, Request from pydantic import BaseModel, ConfigDict, Field - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.pairing import ( BatchPairingCmd, @@ -146,6 +145,41 @@ async def list_pairings( return {"success": True, "data": serialize_control_data(result.data)} +@pairing_router.get("/pairings/count", response_model=dict) +async def count_pairings( + request: Request, + channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"), + account_id: str | None = Query(default=None, description="按账户 ID 过滤"), + peer_id: str | None = Query(default=None, description="按对端 ID 过滤"), + status: PairingStatus | None = Query(default=None, description="按状态过滤"), + created_after: str | None = Query(default=None, description="申请时间下界（ISO 8601，含，按 requested_at 过滤）"), + created_before: str | None = Query(default=None, description="申请时间上界（ISO 8601，含，按 requested_at 过滤）"), + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """统计配对审批记录数（FR-33 待办角标）。 + + 支持按渠道/账户/对端/状态过滤，返回单个整数计数。 + 对应控制面操作 ``pairing/count``。 + """ + operator = build_operator(current_user, request) + created_after_dt = parse_datetime("created_after", created_after) + created_before_dt = parse_datetime("created_before", created_before) + query = PairingQuery( + channel_type=channel_type, + account_id=account_id, + peer_id=peer_id, + status=status, + created_after=created_after_dt, + created_before=created_before_dt, + limit=1, + offset=0, + ) + result = await use_cases.pairing_management.countPairings(query, operator) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)} + + @pairing_router.post("/pairings", response_model=dict) async def create_pairing( payload: CreatePairingRequest, diff --git a/backend/server/routers/channels/plugin_router.py b/backend/server/routers/channels/plugin_router.py index c4adca7a..bf2867a1 100644 --- a/backend/server/routers/channels/plugin_router.py +++ b/backend/server/routers/channels/plugin_router.py @@ -102,7 +102,10 @@ class BatchPluginLifecycleRequest(BaseModel): @plugin_router.get("", response_model=dict) async def list_plugins( request: Request, - state: PluginStateLiteral | None = Query(default=None, description="按生命周期状态过滤（discovered/resolved/loaded/initialized/started/paused/stopped/unloaded/failed/not_installed/installed）"), + state: PluginStateLiteral | None = Query( + default=None, + description="按生命周期状态过滤（discovered/resolved/loaded/initialized/started/paused/stopped/unloaded/failed/not_installed/installed）", + ), use_cases=Depends(get_channel_use_cases), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: diff --git a/backend/server/routers/channels/session_router.py b/backend/server/routers/channels/session_router.py index a2b94e59..b87133c9 100644 --- a/backend/server/routers/channels/session_router.py +++ b/backend/server/routers/channels/session_router.py @@ -1,6 +1,6 @@ """会话资源域 Router。 -提供渠道会话管理的 11 个 HTTP 端点，全部由 ``get_admin_user`` 守门，仅管理员 +提供渠道会话管理的 12 个 HTTP 端点，全部由 ``get_admin_user`` 守门，仅管理员 可访问。Router 不含任何业务逻辑，仅做协议翻译：将 HTTP 请求参数组装为契约层 命令/参数，调用 ``use_cases.session_management`` 的类型化方法走控制面管道， 再通过 ``raiseOnControlFailure`` 转译失败、``serialize_control_data`` 序列化 @@ -10,6 +10,7 @@ 端点清单： - GET /sessions SES-QUERY-01 列出渠道会话 - POST /sessions/batch-close SES-BATCH-CLOSE-01 批量关闭会话 + - GET /sessions/events SES-EVENTS-01 渠道会话实时事件 SSE - GET /sessions/{session_id} SES-QUERY-02 查询会话详情 - POST /sessions/{session_id}/merge SES-05 合并会话（FR-07） - POST /sessions/{session_id}/transfer SES-13 转移会话所有者（FR-26） @@ -23,22 +24,32 @@ from __future__ import annotations +import asyncio +import json +from collections.abc import AsyncIterator from typing import Any from fastapi import APIRouter, Depends, Query, Request +from fastapi.responses import StreamingResponse from pydantic import BaseModel, ConfigDict, Field - +from yuxi.channels.application.extension.channel_event_broadcaster import ( + ChannelEventBroadcaster, +) from yuxi.channels.contract.dtos.channel import ChannelType, SessionStatus from yuxi.channels.contract.dtos.conversation import MergeConversationCmd +from yuxi.channels.contract.dtos.plugin import DomainEvent from yuxi.channels.contract.dtos.session import ( BatchCloseSessionsCmd, CloseSessionCmd, OwnerTransferCmd, ) +from yuxi.channels.infrastructure import DependencyInjectionContainer from yuxi.storage.postgres.models_business import User from server.routers.channels import ( build_operator, + dataclass_to_dict, + get_channel_di_container, get_channel_use_cases, parse_datetime, raiseOnControlFailure, @@ -212,9 +223,7 @@ async def batch_close_sessions( if payload.filter is not None: filter_param = { "channel_type": payload.filter.channel_type, - "inactive_before": parse_datetime( - "inactive_before", payload.filter.inactive_before - ), + "inactive_before": parse_datetime("inactive_before", payload.filter.inactive_before), "status": payload.filter.status, } cmd = BatchCloseSessionsCmd( @@ -229,6 +238,66 @@ async def batch_close_sessions( return {"success": True, "data": serialize_control_data(result.data)} +@session_router.get("/sessions/events") +async def stream_session_events( + request: Request, + di_container: DependencyInjectionContainer = Depends(get_channel_di_container), + current_user: User = Depends(get_admin_user), +) -> StreamingResponse: + """渠道会话/消息实时事件 SSE 端点（BE-14）。 + + 返回 ``text/event-stream`` 长连接，将 ``ChannelSessionUpdated`` / + ``ChannelMessageReceived`` / ``ChannelMessageSent`` 三类事件以 SSE 帧 + 形式推送给已认证管理员。每 15 秒无事件时发送心跳注释，客户端断开时 + 自动取消订阅。 + + 本端点静态路径先于 ``/sessions/{session_id}`` 声明，避免被动态参数捕获。 + """ + broadcaster = di_container.resolve(ChannelEventBroadcaster) + + async def event_stream() -> AsyncIterator[str]: + """SSE 事件流异步生成器。""" + queue, subscription_id = await broadcaster.subscribe() + try: + while not await request.is_disconnected(): + try: + event = await asyncio.wait_for(queue.get(), timeout=15.0) + except TimeoutError: + if await request.is_disconnected(): + break + yield ":heartbeat\n\n" + continue + if await request.is_disconnected(): + break + yield _format_sse_event(event) + finally: + await broadcaster.unsubscribe(subscription_id) + + return StreamingResponse( + event_stream(), + media_type="text/event-stream; charset=utf-8", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + +def _format_sse_event(event: DomainEvent) -> str: + """将 ``DomainEvent`` 格式化为 SSE 帧。 + + 输出格式： + event: {event.event_type}\n + id: {event.event_id}\n + data: {json.dumps(event_dict)}\n\n + + 其中 ``event_dict`` 经 ``dataclass_to_dict`` 递归序列化为 JSON 兼容结构。 + """ + event_dict = dataclass_to_dict(event) + return f"event: {event.event_type}\nid: {event.event_id}\ndata: {json.dumps(event_dict, ensure_ascii=False)}\n\n" + + @session_router.get("/sessions/{session_id}", response_model=dict) async def get_session( session_id: str, diff --git a/backend/server/routers/channels/wizard_router.py b/backend/server/routers/channels/wizard_router.py index f5703486..b7939223 100644 --- a/backend/server/routers/channels/wizard_router.py +++ b/backend/server/routers/channels/wizard_router.py @@ -29,7 +29,6 @@ from urllib.parse import urlparse from fastapi import APIRouter, Depends, Request from pydantic import BaseModel, ConfigDict, Field, field_validator - from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.storage.postgres.models_business import User @@ -142,6 +141,15 @@ class OAuthInitiateRequest(BaseModel): return v +class QrLoginWaitRequest(BaseModel): + """扫码登录轮询请求。字段对齐 ``WizardPort.waitWizardQrLogin`` 入参。""" + + model_config = ConfigDict(frozen=True) + + session_id: str = Field(..., min_length=1, description="扫码会话 ID（由 qr/initiate 返回）") + current_qr_data_url: str | None = Field(None, description="当前二维码数据 URL（用于刷新场景）") + + # ---------------- Endpoints ---------------- @@ -310,3 +318,51 @@ async def handle_wizard_oauth_callback( ) raiseOnControlFailure(result) return {"success": True, "data": serialize_control_data(result.data)} + + +@wizard_router.post("/{channel_type}/wizard/qr/initiate", response_model=dict) +async def initiate_wizard_qr_login( + channel_type: ChannelType, + request: Request, + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """发起向导扫码登录（WIZ-07）。 + + 对应控制面操作 ``wizard/qr-initiate``（由 ``ChannelControlService.initiateWizardQrLogin`` + 内部构造 ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。 + 在账户尚未创建时复用 ``LoginAdapter`` 获取二维码，返回 session_id 与 + qr_data_url 供前端展示并轮询。 + """ + operator = build_operator(current_user, request) + result = await use_cases.wizard.initiateWizardQrLogin( + channel_type=channel_type, + operator=operator, + ) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)} + + +@wizard_router.post("/{channel_type}/wizard/qr/wait", response_model=dict) +async def wait_wizard_qr_login( + channel_type: ChannelType, + payload: QrLoginWaitRequest, + request: Request, + use_cases=Depends(get_channel_use_cases), + current_user: User = Depends(get_admin_user), +) -> dict[str, Any]: + """轮询向导扫码登录结果（WIZ-08）。 + + 对应控制面操作 ``wizard/qr-wait``（由 ``ChannelControlService.waitWizardQrLogin`` + 内部构造 ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。 + 扫码成功后返回 credentials，由前端填入当前向导步骤并通过 ``apply`` 提交。 + """ + operator = build_operator(current_user, request) + result = await use_cases.wizard.waitWizardQrLogin( + channel_type=channel_type, + session_id=payload.session_id, + current_qr_data_url=payload.current_qr_data_url, + operator=operator, + ) + raiseOnControlFailure(result) + return {"success": True, "data": serialize_control_data(result.data)}