From fb754d82f136086c223322061c92d763bc149dbd Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 12:08:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=B7=BB=E5=8A=A0=E5=85=A8?= =?UTF-8?q?=E6=B8=A0=E9=81=93=E7=BD=91=E5=85=B3=E6=A0=B8=E5=BF=83=E5=9F=BA?= =?UTF-8?q?=E7=A1=80=E6=9E=B6=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增渠道扩展框架的核心基础层，为所有渠道扩展提供统一的基础设施。 主要变更： - channel/extensions/__init__.py: 渠道扩展包初始化与注册 - channel/extensions/base.py: 渠道插件基类，定义统一接口规范 - channel_service.py: 渠道服务核心业务逻辑 - models_channels.py: 渠道相关数据库模型 - 8 个 channel_*_repo.py: 渠道数据仓库层 - storage/postgres/manager.py: 数据库管理器，新增渠道表 - models_business.py: 业务模型更新 - config/app.py: 应用配置更新 - conversation_repository.py: 会话仓库更新 - oidc_service.py: OIDC 服务更新 - logging_config.py: 日志配置更新 --- .../yuxi/channel/extensions/__init__.py | 61 ++ .../package/yuxi/channel/extensions/base.py | 611 ++++++++++++++++++ backend/package/yuxi/config/app.py | 58 ++ .../repositories/channel_allowlist_repo.py | 72 +++ .../yuxi/repositories/channel_binding_repo.py | 94 +++ .../yuxi/repositories/channel_config_repo.py | 67 ++ .../channel_device_identity_repo.py | 74 +++ .../channel_identity_link_repo.py | 47 ++ .../repositories/channel_msg_record_repo.py | 68 ++ .../yuxi/repositories/channel_pairing_repo.py | 170 +++++ .../channel_thread_mapping_repo.py | 106 +++ .../repositories/channel_user_mapping_repo.py | 124 ++++ .../repositories/conversation_repository.py | 99 ++- .../package/yuxi/services/channel_service.py | 244 +++++++ backend/package/yuxi/services/oidc_service.py | 3 + .../package/yuxi/storage/postgres/manager.py | 175 +++++ .../yuxi/storage/postgres/models_business.py | 38 +- .../yuxi/storage/postgres/models_channels.py | 462 +++++++++++++ backend/package/yuxi/utils/logging_config.py | 19 +- 19 files changed, 2577 insertions(+), 15 deletions(-) create mode 100644 backend/package/yuxi/channel/extensions/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/base.py create mode 100644 backend/package/yuxi/repositories/channel_allowlist_repo.py create mode 100644 backend/package/yuxi/repositories/channel_binding_repo.py create mode 100644 backend/package/yuxi/repositories/channel_config_repo.py create mode 100644 backend/package/yuxi/repositories/channel_device_identity_repo.py create mode 100644 backend/package/yuxi/repositories/channel_identity_link_repo.py create mode 100644 backend/package/yuxi/repositories/channel_msg_record_repo.py create mode 100644 backend/package/yuxi/repositories/channel_pairing_repo.py create mode 100644 backend/package/yuxi/repositories/channel_thread_mapping_repo.py create mode 100644 backend/package/yuxi/repositories/channel_user_mapping_repo.py create mode 100644 backend/package/yuxi/services/channel_service.py create mode 100644 backend/package/yuxi/storage/postgres/models_channels.py diff --git a/backend/package/yuxi/channel/extensions/__init__.py b/backend/package/yuxi/channel/extensions/__init__.py new file mode 100644 index 00000000..1a3bd21b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/__init__.py @@ -0,0 +1,61 @@ +import logging + +try: + from yuxi.channel.extensions import irc # noqa +except ImportError: + pass + +try: + from yuxi.channel.extensions import nostr # noqa +except ImportError: + pass + +try: + from yuxi.channel.extensions import wecom # noqa +except ImportError: + pass + +try: + from yuxi.channel.extensions import activitypub # noqa +except ImportError: + pass + +try: + from yuxi.channel.extensions import kuaishou # noqa +except ImportError: + pass + +try: + from yuxi.channel.extensions import tencent_im # noqa +except ImportError: + pass + +try: + from yuxi.channel.extensions import xiaohongshu # noqa +except ImportError: + pass + +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +logger = logging.getLogger(__name__) + + +def init_builtin_channels(): + logger.info("Initializing built-in channel plugins...") + + from yuxi.channel.plugins.discovery import run_discovery_pipeline + + result = run_discovery_pipeline() + logger.info( + "Plugin discovery complete: %d discovered, %d skipped, %d errors", + sum(1 for d in result.discovered if not d.already_registered), + len(result.skipped), + len(result.errors), + ) + + plugins = ChannelPluginRegistry.list_all() + logger.info( + "Registered %d channel plugins: %s", + len(plugins), + [p.id for p in plugins], + ) \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/base.py b/backend/package/yuxi/channel/extensions/base.py new file mode 100644 index 00000000..1ab7b646 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/base.py @@ -0,0 +1,611 @@ +from yuxi.channel.capabilities import ChannelCapabilities + + +class BaseChannelPlugin: + """可选的渠道插件基类，提供默认实现。渠道不强制继承，满足 Protocol 即可。""" + + id: str = "" + name: str = "" + order: int = 99 + label: str | None = None + aliases: list[str] | None = None + resolve_reply_to_mode: str | None = None + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities() + + async def resolve_account(self, account_id: str) -> dict: + return {} + + async def list_accounts(self) -> list[dict]: + return [] + + def list_account_ids(self, config: dict) -> list[str]: + return ["default"] + + def is_configured(self, account: dict) -> bool: + return bool(account) + + def is_enabled(self, account: dict) -> bool: + return True + + def disabled_reason(self, account: dict) -> str: + return "" + + async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: + return None + + def describe_account(self, account: dict) -> dict: + return {"account_id": account.get("account_id", "")} + + async def start(self, ctx) -> object: + pass + + async def stop(self, ctx) -> None: + pass + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + pass + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + ) -> None: + pass + + async def probe(self, account: dict) -> bool: + return True + + def build_summary(self, snapshot: object) -> dict: + return {} + + async def resolve_thread_id(self, account_id: str, reply_to_id: str | None) -> str | None: + return None + + async def send_typing(self, target_id: str, thread_id: str | None = None) -> None: + pass + + async def clear_typing(self, target_id: str, thread_id: str | None = None) -> None: + pass + + async def check_ready(self, account_id: str) -> bool: + return True + + # ── HeartbeatProtocol ──────────────────────────────── + + @property + def ping_interval_ms(self) -> int: + return 30000 + + async def on_ping(self, account: dict) -> bool: + return True + + async def start_typing(self, target_id: str, account: dict) -> object: + return None + + async def stop_typing(self, handle: object) -> None: + pass + + async def typing_indicator( + self, + target_id: str, + account: dict, + _stop_event, + ) -> None: + pass + + # ── AgentPromptProtocol ────────────────────────────── + + def build_system_prompt(self, context) -> str | None: + return None + + def build_context_note(self, context) -> str: + return "" + + @property + def channel_format_instructions(self) -> str | None: + return None + + # ── ThreadingProtocol ──────────────────────────────── + + def extract_thread_id(self, msg: object) -> str | None: + return None + + def resolve_reply_transport(self, msg: object, thread_id: str | None): + from yuxi.channel.protocols import ReplyTransport + + return ReplyTransport() + + def get_thread_info(self, thread_id: str): + return None + + # ── MessagingProtocol ──────────────────────────────── + + def resolve_session(self, msg: object): + from yuxi.channel.message.models import PeerKind + from yuxi.channel.protocols import SessionResolution + + if hasattr(msg, "sender") and hasattr(msg.sender, "kind"): + if msg.sender.kind == PeerKind.DIRECT: + return SessionResolution(kind="direct", conversation_id=msg.sender.id) + gid = msg.group.id if hasattr(msg, "group") and msg.group and msg.group.id else "unknown" + return SessionResolution(kind="group", conversation_id=gid) + + def parse_explicit_target(self, content: str) -> str | None: + return None + + def build_reply_payload( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + session_key: str | None = None, + agent_config_id: str | None = None, + **kwargs, + ): + from yuxi.channel.protocols import BoundReplyPayload + + return BoundReplyPayload(target_id=target_id, content=content, reply_to_id=reply_to_id, thread_id=thread_id) + + def build_attachment_refs(self, attachments: list[dict]) -> str: + lines = [] + for att in attachments: + ft = att.get("file_type", "file") + fp = att.get("file_path", "") + label = {"image": "图片", "video": "视频"}.get(ft, "文件") + lines.append(f"[{label}: {fp}]") + return "\n".join(lines) + + def resolve_focused_binding(self, msg: object, session, current_binding: object) -> dict | None: + return None + + @property + def supported_explicit_targets(self) -> list[str]: + return [] + + # ── AgentToolProtocol ──────────────────────────────── + + def get_agent_tools(self) -> list: + return [] + + async def execute_agent_tool(self, tool_name: str, params: dict, context: dict) -> dict: + return {"success": False, "error": f"Unknown tool: {tool_name}"} + + # ── MessageActionProtocol ──────────────────────────── + + @property + def actions(self): + """消息动作适配器。子类重写以注册支持的动作。 + + 用法:: + + class MyPlugin(BaseChannelPlugin): + @property + def actions(self): + if self._actions is None: + self._actions = MessageActionRegistry() + self._actions.register( + MessageAction.SEND, + self._handle_send, + description="发送文本消息", + ) + return self._actions + """ + return None + + def get_message_actions(self) -> list: + if self.actions: + return self.actions.get_message_actions() + return [] + + async def execute_message_action(self, action: str, params: dict, context: dict) -> dict: + if self.actions: + result = await self.actions.execute_action(action, params, context) + return result + return {"success": False, "error": f"Unknown action: {action}"} + + # ── SecurityProtocol ──────────────────────────────── + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + return True + + def resolve_dm_policy(self) -> dict: + return {"mode": "open", "allow_from": []} + + # ── PairingProtocol ───────────────────────────────── + + async def generate_code(self, peer_id: str) -> str: + return "" + + async def verify_code(self, peer_id: str, code: str) -> bool: + return True + + # ── GroupsProtocol ────────────────────────────────── + + async def list_groups(self) -> list[dict]: + return [] + + # ── MentionsProtocol ──────────────────────────────── + + def extract_mentions(self, raw_message: dict) -> list[str]: + return [] + + # ── LifecycleProtocol ─────────────────────────────── + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + pass + + async def on_account_removed(self, account_id: str) -> None: + pass + + async def on_retire(self) -> None: + pass + + async def run_startup_maintenance(self, cfg: dict) -> None: + pass + + # ── ApprovalProtocol ──────────────────────────────── + + async def check_approval_required(self, config: dict, action, initiator_peer_id: str) -> bool: + approval_cfg = config.get("approval", {}) + if not approval_cfg.get("enabled", False): + return False + approvers = approval_cfg.get("approvers", []) + if not approvers or initiator_peer_id in approvers: + return False + return True + + async def create_approval_request( + self, config: dict, action, initiator_peer_id: str, description: str, context: dict + ): + from yuxi.channel.protocols import ApprovalRequest + + return ApprovalRequest( + id="", + channel_type="", + account_id="", + initiator_peer_id=initiator_peer_id, + action=action, + description=description, + ) + + async def send_approval_notification(self, config: dict, request, approver_peer_ids: list[str]) -> bool: + return False + + async def check_approval_status(self, config: dict, request_id: str): + return None + + def get_approver_ids(self, config: dict) -> list[str]: + return config.get("approval", {}).get("approvers", []) + + # ── ConfigSchemaProtocol ──────────────────────────── + + def config_schema(self) -> dict: + return {} + + # ── SetupProtocol ─────────────────────────────────── + + async def setup(self, config: dict) -> bool: + return True + + # ── SetupWizardProtocol ───────────────────────────── + + def setup_wizard_steps(self) -> list: + return [] + + async def validate_wizard_input(self, step_key: str, value: str) -> str | None: + return None + + # ── AuthProtocol ──────────────────────────────────── + + async def login(self, config: dict) -> dict: + return {} + + async def logout(self, config: dict) -> None: + pass + + async def refresh_token(self, config: dict) -> dict | None: + return None + + async def qr_auth_start(self, config: dict) -> str: + return "" + + async def qr_auth_wait(self, config: dict) -> dict | None: + return None + + # ── ElevatedProtocol ──────────────────────────────── + + async def elevate(self, account: dict, reason: str) -> bool: + return True + + async def revoke(self, account: dict) -> None: + pass + + @property + def is_elevated(self) -> bool: + return False + + # ── AllowlistManagementProtocol ───────────────────── + + async def list_allow_entries(self, config: dict, scope: str) -> list[str]: + return [] + + async def add_allow_entry(self, config: dict, scope: str, entry: str) -> bool: + return False + + async def remove_allow_entry(self, config: dict, scope: str, entry: str) -> bool: + return False + + # ── ReactionProtocol ───────────────────────────────── + + async def send_reaction( + self, + target_id: str, + message_id: str, + emoji: str, + *, + account_id: str | None = None, + ) -> None: + pass + + async def remove_reaction( + self, + target_id: str, + message_id: str, + emoji: str, + *, + account_id: str | None = None, + ) -> None: + pass + + async def fetch_reactions( + self, + target_id: str, + message_id: str, + *, + account_id: str | None = None, + ) -> list[dict]: + return [] + + # ── WebhookProtocol ────────────────────────────────── + + async def handle_webhook( + self, + request: object, + account_id: str | None = None, + ) -> object: + return None + + def verify_signature(self, body: bytes, signature: str, secret: str) -> bool: + return False + + def resolve_webhook_auth_bypass_paths(self) -> list[str]: + return [] + + # ── InboundHandlerProtocol ─────────────────────────── + + async def handle_raw_event( + self, + event: dict, + account: dict, + ) -> object | None: + return None + + def parse_to_unified( + self, + raw_event: dict, + account_id: str, + ) -> object | None: + return None + + def resolve_event_type(self, raw_event: dict) -> str: + return "unknown" + + # ── MediaProtocol ──────────────────────────────────── + + async def upload_media( + self, + file_path: str, + media_type: str, + *, + account_id: str | None = None, + ) -> dict: + return {} + + async def download_media( + self, + media_url: str, + *, + account_id: str | None = None, + ) -> bytes: + return b"" + + def detect_media_type(self, filename: str, content_type: str = "") -> str: + return "file" + + def validate_media_url(self, url: str) -> bool: + return False + + def resolve_media_size_limit(self, media_type: str) -> int: + return 0 + + # ── FormatProtocol ─────────────────────────────────── + + def markdown_to_native(self, md_text: str) -> dict | str: + return md_text + + def native_to_markdown(self, native_content: dict | str) -> str: + if isinstance(native_content, dict): + return native_content.get("text", "") + return str(native_content) + + # ── DedupeProtocol ─────────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return False + + def mark_seen(self, key: str) -> None: + pass + + def reset(self) -> None: + pass + + @property + def ttl_seconds(self) -> int: + return 300 + + @property + def max_entries(self) -> int: + return 10000 + + # ── CardProtocol ───────────────────────────────────── + + async def send_card( + self, + target_id: str, + card_content: dict, + *, + reply_to_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + return None + + async def update_card( + self, + message_id: str, + card_content: dict, + *, + account_id: str | None = None, + ) -> bool: + return False + + async def edit_card( + self, + target_id: str, + message_id: str, + card_content: dict, + *, + thread_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + ok = await self.update_card(message_id, card_content, account_id=account_id) + return message_id if ok else None + + def build_text_card(self, text: str, title: str | None = None, **kwargs) -> dict: + return {"text": text} + + def supports_card_interactions(self) -> bool: + return False + + # ── ErrorHandlingProtocol ──────────────────────────── + + def classify_error(self, error: BaseException) -> object: + from yuxi.channel.errors import classify_error + + return classify_error(error) + + def is_retryable(self, error: BaseException) -> bool: + from yuxi.channel.errors import is_retryable + + return is_retryable(error) + + def should_backoff(self, error: BaseException, attempt: int) -> int: + from yuxi.channel.errors import should_backoff + + return should_backoff(error, attempt) + + # ── PollingProtocol ────────────────────────────────── + + poll_interval_seconds: int = 60 + + async def poll_once(self, account: dict) -> list[dict]: + return [] + + async def should_poll(self, account: dict) -> bool: + return False + + def get_poll_cursor(self, account_id: str) -> str | None: + return None + + def set_poll_cursor(self, account_id: str, cursor: str) -> None: + pass + + # ── PinProtocol ────────────────────────────────────── + + async def pin_message( + self, + message_id: str, + *, + account_id: str | None = None, + ) -> bool: + return False + + async def unpin_message( + self, + message_id: str, + *, + account_id: str | None = None, + ) -> bool: + return False + + async def list_pins( + self, + chat_id: str, + *, + account_id: str | None = None, + ) -> list[dict]: + return [] + + # ── WebsocketProtocol ──────────────────────────────── + + async def connect_ws(self, account: dict) -> object: + return None + + async def disconnect_ws(self) -> None: + pass + + async def on_ws_message(self, message: str | bytes) -> None: + pass + + @property + def ws_reconnect_interval_ms(self) -> int: + return 5000 + + @property + def ws_heartbeat_interval_ms(self) -> int: + return 30000 + + # ── Hooks ─────────────────────────────────────────── + + async def on_message_sending(self, msg: object, content: str) -> str | None: + return content + + async def on_message_received(self, msg: object) -> object | None: + return msg + + async def before_send_attempt(self, target_id: str, content: str) -> None: + pass + + async def after_send_success(self, target_id: str, receipt: object) -> None: + pass + + async def after_send_failure(self, target_id: str, error: str) -> None: + pass + + async def after_commit(self, receipt: object) -> None: + pass diff --git a/backend/package/yuxi/config/app.py b/backend/package/yuxi/config/app.py index 117245fc..63824e85 100644 --- a/backend/package/yuxi/config/app.py +++ b/backend/package/yuxi/config/app.py @@ -28,6 +28,50 @@ from yuxi.config.static.models import ( from yuxi.utils.logging_config import logger +class DependencyWhitelistEntry(BaseModel): + """依赖白名单条目。""" + + name: str + min_version: str | None = None + max_version: str | None = None + allowed_versions: list[str] | None = None + blocked_versions: list[str] | None = None + + +DEFAULT_DEPENDENCY_WHITELIST: list[DependencyWhitelistEntry] = [ + DependencyWhitelistEntry(name="aiofiles", min_version="0.8.0"), + DependencyWhitelistEntry(name="aiohttp", min_version="3.8.0"), + DependencyWhitelistEntry(name="aiosignal", min_version="1.2.0"), + DependencyWhitelistEntry(name="asyncpg", min_version="0.28.0"), + DependencyWhitelistEntry(name="cachetools", min_version="5.0.0"), + DependencyWhitelistEntry(name="cryptography", min_version="41.0.0"), + DependencyWhitelistEntry(name="fastapi", min_version="0.100.0"), + DependencyWhitelistEntry(name="frozenlist", min_version="1.3.0"), + DependencyWhitelistEntry(name="httpx", min_version="0.24.0"), + DependencyWhitelistEntry(name="lxml", min_version="4.9.0"), + DependencyWhitelistEntry(name="multidict", min_version="6.0.0"), + DependencyWhitelistEntry(name="numpy", min_version="1.24.0"), + DependencyWhitelistEntry(name="orjson", min_version="3.9.0"), + DependencyWhitelistEntry(name="pillow", min_version="10.0.0"), + DependencyWhitelistEntry(name="psycopg2-binary", min_version="2.9.0"), + DependencyWhitelistEntry(name="pydantic", min_version="2.0.0"), + DependencyWhitelistEntry(name="pydantic-core", min_version="2.0.0"), + DependencyWhitelistEntry(name="pydantic-settings", min_version="2.0.0"), + DependencyWhitelistEntry(name="python-dotenv", min_version="1.0.0"), + DependencyWhitelistEntry(name="pyyaml", min_version="6.0"), + DependencyWhitelistEntry(name="redis", min_version="4.5.0"), + DependencyWhitelistEntry(name="requests", min_version="2.28.0"), + DependencyWhitelistEntry(name="sqlalchemy", min_version="2.0.0"), + DependencyWhitelistEntry(name="starlette", min_version="0.27.0"), + DependencyWhitelistEntry(name="structlog", min_version="23.0.0"), + DependencyWhitelistEntry(name="tenacity", min_version="8.0.0"), + DependencyWhitelistEntry(name="urllib3", min_version="1.26.0"), + DependencyWhitelistEntry(name="uvicorn", min_version="0.22.0"), + DependencyWhitelistEntry(name="websockets", min_version="10.0"), + DependencyWhitelistEntry(name="yarl", min_version="1.8.0"), +] + + class Config(BaseModel): """应用配置类""" @@ -84,9 +128,23 @@ class Config(BaseModel): sandbox_max_output_bytes: int = Field(default=262144, description="沙箱最大输出字节数") sandbox_keepalive_interval_seconds: int = Field(default=30, description="沙箱保活间隔（秒）") + # ============================================================ + # Media Store 配置 + # ============================================================ + media_store_enabled: bool = Field(default=True, description="是否启用媒体本地存储") + media_store_max_total_bytes: int = Field(default=500 * 1024 * 1024, description="媒体存储总上限（字节），默认500MB") + media_store_max_file_bytes: int = Field(default=10 * 1024 * 1024, description="单文件大小上限（字节），默认10MB") + media_store_ttl_seconds: int = Field(default=7 * 24 * 3600, description="媒体文件TTL（秒），默认7天") + media_cleanup_interval_seconds: int = Field(default=3600, description="清理扫描间隔（秒），默认1小时") + # ============================================================ # 模型信息（只读，不持久化） # ============================================================ + dependency_whitelist: list[DependencyWhitelistEntry] = Field( + default_factory=lambda: [DependencyWhitelistEntry(**e.model_dump()) for e in DEFAULT_DEPENDENCY_WHITELIST], + description="渠道插件依赖白名单", + exclude=True, + ) model_names: dict[str, ChatModelProvider] = Field( default_factory=lambda: DEFAULT_CHAT_MODEL_PROVIDERS.copy(), description="聊天模型提供商配置", diff --git a/backend/package/yuxi/repositories/channel_allowlist_repo.py b/backend/package/yuxi/repositories/channel_allowlist_repo.py new file mode 100644 index 00000000..e3971d20 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_allowlist_repo.py @@ -0,0 +1,72 @@ +from datetime import UTC, datetime + +from sqlalchemy import delete, select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelAllowlistEntry + + +class ChannelAllowlistRepository: + async def list_by_channel(self, channel_id: str, allow_type: str | None = None) -> list[ChannelAllowlistEntry]: + async with pg_manager.get_async_session_context() as session: + query = select(ChannelAllowlistEntry).where( + ChannelAllowlistEntry.channel_id == channel_id, + ChannelAllowlistEntry.enabled == True, + ) + if allow_type: + query = query.where(ChannelAllowlistEntry.allow_type == allow_type) + result = await session.execute(query) + return list(result.scalars().all()) + + async def add_entry(self, channel_id: str, allow_type: str, allow_id: str) -> ChannelAllowlistEntry | None: + now = datetime.now(UTC).replace(tzinfo=None) + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelAllowlistEntry).where( + ChannelAllowlistEntry.channel_id == channel_id, + ChannelAllowlistEntry.allow_type == allow_type, + ChannelAllowlistEntry.allow_id == allow_id, + ) + ) + entry = result.scalar_one_or_none() + if entry is not None: + entry.enabled = True + entry.updated_at = now + else: + entry = ChannelAllowlistEntry( + channel_id=channel_id, + allow_type=allow_type, + allow_id=allow_id, + ) + session.add(entry) + await session.commit() + await session.refresh(entry) + return entry + + async def remove_entry(self, channel_id: str, allow_type: str, allow_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelAllowlistEntry).where( + ChannelAllowlistEntry.channel_id == channel_id, + ChannelAllowlistEntry.allow_type == allow_type, + ChannelAllowlistEntry.allow_id == allow_id, + ) + ) + entry = result.scalar_one_or_none() + if entry is None: + return False + entry.enabled = False + await session.commit() + return True + + async def delete_entry(self, channel_id: str, allow_type: str, allow_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + delete(ChannelAllowlistEntry).where( + ChannelAllowlistEntry.channel_id == channel_id, + ChannelAllowlistEntry.allow_type == allow_type, + ChannelAllowlistEntry.allow_id == allow_id, + ) + ) + await session.commit() + return result.rowcount > 0 diff --git a/backend/package/yuxi/repositories/channel_binding_repo.py b/backend/package/yuxi/repositories/channel_binding_repo.py new file mode 100644 index 00000000..6476e203 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_binding_repo.py @@ -0,0 +1,94 @@ +from typing import Any + +from sqlalchemy import select + +from yuxi.channel.routing.models import BindingMatch, PeerConstraint, PeerKind, RouteBinding +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelBinding + + +def to_route_binding(binding: ChannelBinding) -> RouteBinding: + peer_constraint = None + if binding.peer_kind and binding.peer_id: + peer_constraint = PeerConstraint( + kind=PeerKind(binding.peer_kind), + peer_id=binding.peer_id, + ) + + match = BindingMatch( + channel=binding.channel_type, + channel_config_id=binding.channel_config_id, + account_id=binding.account_id, + peer=peer_constraint, + guild_id=binding.guild_id, + team_id=binding.team_id, + roles=list(binding.roles or []), + ) + + return RouteBinding( + agent_config_id=binding.agent_config_id, + match=match, + dm_scope=binding.dm_scope, + priority=binding.priority, + session_dm_scope=binding.session_dm_scope, + ) + + +class ChannelBindingRepository: + async def get_by_id(self, binding_id: str) -> ChannelBinding | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelBinding).where(ChannelBinding.id == binding_id)) + return result.scalar_one_or_none() + + async def list_by_agent(self, agent_config_id: int) -> list[ChannelBinding]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelBinding) + .where(ChannelBinding.agent_config_id == agent_config_id) + .order_by(ChannelBinding.priority.desc()) + ) + return list(result.scalars().all()) + + async def list_all(self) -> list[ChannelBinding]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelBinding).order_by(ChannelBinding.priority.desc())) + return list(result.scalars().all()) + + async def list_by_channel_type(self, channel_type: str) -> list[ChannelBinding]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelBinding) + .where(ChannelBinding.channel_type == channel_type) + .order_by(ChannelBinding.priority.desc()) + ) + return list(result.scalars().all()) + + async def create(self, data: dict[str, Any]) -> ChannelBinding: + async with pg_manager.get_async_session_context() as session: + binding = ChannelBinding(**data) + session.add(binding) + await session.commit() + await session.refresh(binding) + return binding + + async def update(self, binding_id: str, data: dict[str, Any]) -> ChannelBinding | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelBinding).where(ChannelBinding.id == binding_id)) + binding = result.scalar_one_or_none() + if binding is None: + return None + for key, value in data.items(): + if key != "id": + setattr(binding, key, value) + await session.commit() + await session.refresh(binding) + return binding + + async def delete(self, binding_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelBinding).where(ChannelBinding.id == binding_id)) + binding = result.scalar_one_or_none() + if binding is None: + return False + await session.delete(binding) + return True diff --git a/backend/package/yuxi/repositories/channel_config_repo.py b/backend/package/yuxi/repositories/channel_config_repo.py new file mode 100644 index 00000000..c19c3ab9 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_config_repo.py @@ -0,0 +1,67 @@ +from typing import Any + +from sqlalchemy import select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelConfig + + +class ChannelConfigRepository: + async def get_by_id(self, config_id: str) -> ChannelConfig | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelConfig).where(ChannelConfig.id == config_id)) + return result.scalar_one_or_none() + + async def list_by_type(self, channel_type: str) -> list[ChannelConfig]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelConfig) + .where(ChannelConfig.channel_type == channel_type) + .order_by(ChannelConfig.created_at.desc()) + ) + return list(result.scalars().all()) + + async def list_enabled(self, channel_type: str | None = None) -> list[ChannelConfig]: + async with pg_manager.get_async_session_context() as session: + query = select(ChannelConfig).where(ChannelConfig.enabled.is_(True)) + if channel_type: + query = query.where(ChannelConfig.channel_type == channel_type) + query = query.order_by(ChannelConfig.created_at.desc()) + result = await session.execute(query) + return list(result.scalars().all()) + + async def list_all(self) -> list[ChannelConfig]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelConfig).order_by(ChannelConfig.created_at.desc())) + return list(result.scalars().all()) + + async def create(self, data: dict[str, Any]) -> ChannelConfig: + async with pg_manager.get_async_session_context() as session: + config = ChannelConfig(**data) + session.add(config) + await session.commit() + await session.refresh(config) + return config + + async def update(self, config_id: str, data: dict[str, Any]) -> ChannelConfig | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelConfig).where(ChannelConfig.id == config_id)) + config = result.scalar_one_or_none() + if config is None: + return None + for key, value in data.items(): + if key != "id": + setattr(config, key, value) + await session.commit() + await session.refresh(config) + return config + + async def delete(self, config_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelConfig).where(ChannelConfig.id == config_id)) + config = result.scalar_one_or_none() + if config is None: + return False + await session.delete(config) + await session.commit() + return True diff --git a/backend/package/yuxi/repositories/channel_device_identity_repo.py b/backend/package/yuxi/repositories/channel_device_identity_repo.py new file mode 100644 index 00000000..53479d82 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_device_identity_repo.py @@ -0,0 +1,74 @@ +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import DeviceIdentity + + +class DeviceIdentityRepository: + async def get_by_device_id(self, device_id: str) -> DeviceIdentity | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(DeviceIdentity).where(DeviceIdentity.device_id == device_id)) + return result.scalar_one_or_none() + + async def create(self, data: dict[str, Any]) -> DeviceIdentity: + async with pg_manager.get_async_session_context() as session: + record = DeviceIdentity(**data) + session.add(record) + await session.commit() + await session.refresh(record) + return record + + async def update(self, device_id: str, data: dict[str, Any]) -> DeviceIdentity | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(DeviceIdentity).where(DeviceIdentity.device_id == device_id)) + record = result.scalar_one_or_none() + if record is None: + return None + for key, value in data.items(): + setattr(record, key, value) + await session.commit() + await session.refresh(record) + return record + + async def mark_used(self, device_id: str) -> bool: + now = datetime.now(UTC).replace(tzinfo=None) + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(DeviceIdentity).where(DeviceIdentity.device_id == device_id)) + record = result.scalar_one_or_none() + if record is None: + return False + record.last_used_at = now + await session.commit() + return True + + async def revoke(self, device_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(DeviceIdentity).where(DeviceIdentity.device_id == device_id)) + record = result.scalar_one_or_none() + if record is None: + return False + record.status = "revoked" + await session.commit() + return True + + async def list_active(self) -> list[DeviceIdentity]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(DeviceIdentity) + .where(DeviceIdentity.status == "active") + .order_by(DeviceIdentity.created_at.desc()) + ) + return list(result.scalars().all()) + + async def delete(self, device_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(DeviceIdentity).where(DeviceIdentity.device_id == device_id)) + record = result.scalar_one_or_none() + if record is None: + return False + await session.delete(record) + await session.commit() + return True diff --git a/backend/package/yuxi/repositories/channel_identity_link_repo.py b/backend/package/yuxi/repositories/channel_identity_link_repo.py new file mode 100644 index 00000000..89b44b70 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_identity_link_repo.py @@ -0,0 +1,47 @@ +from sqlalchemy import delete, select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelIdentityLink + + +class ChannelIdentityLinkRepository: + + async def find_all(self) -> dict[str, list[str]]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelIdentityLink)) + rows = result.scalars().all() + links: dict[str, list[str]] = {} + for row in rows: + links.setdefault(row.identity, []).append(f"{row.channel_type}:{row.peer_id}") + return links + + async def find_by_identity(self, identity: str) -> list[str]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelIdentityLink).where(ChannelIdentityLink.identity == identity) + ) + rows = result.scalars().all() + return [f"{row.channel_type}:{row.peer_id}" for row in rows] + + async def add_link(self, identity: str, channel_type: str, peer_id: str) -> None: + async with pg_manager.get_async_session_context() as session: + link = ChannelIdentityLink(identity=identity, channel_type=channel_type, peer_id=peer_id) + session.add(link) + await session.commit() + + async def remove_link(self, channel_type: str, peer_id: str) -> None: + async with pg_manager.get_async_session_context() as session: + await session.execute( + delete(ChannelIdentityLink).where( + ChannelIdentityLink.channel_type == channel_type, + ChannelIdentityLink.peer_id == peer_id, + ) + ) + await session.commit() + + async def remove_identity(self, identity: str) -> None: + async with pg_manager.get_async_session_context() as session: + await session.execute( + delete(ChannelIdentityLink).where(ChannelIdentityLink.identity == identity) + ) + await session.commit() diff --git a/backend/package/yuxi/repositories/channel_msg_record_repo.py b/backend/package/yuxi/repositories/channel_msg_record_repo.py new file mode 100644 index 00000000..f1205261 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_msg_record_repo.py @@ -0,0 +1,68 @@ +from typing import Any + +from sqlalchemy import select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelMsgRecord +from yuxi.utils.datetime_utils import utc_now_naive + + +class ChannelMsgRecordRepository: + async def get_by_id(self, record_id: int) -> ChannelMsgRecord | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelMsgRecord).where(ChannelMsgRecord.id == record_id)) + return result.scalar_one_or_none() + + async def create_record(self, data: dict[str, Any]) -> ChannelMsgRecord: + async with pg_manager.get_async_session_context() as session: + record = ChannelMsgRecord(**data) + session.add(record) + await session.commit() + await session.refresh(record) + return record + + async def mark_success( + self, record_id: int, reply_message_id: str, reply_content_preview: str, response_time_ms: int + ) -> ChannelMsgRecord | None: + now = utc_now_naive() + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelMsgRecord).where(ChannelMsgRecord.id == record_id)) + record = result.scalar_one_or_none() + if record is None: + return None + record.status = "success" + record.reply_message_id = reply_message_id + record.reply_content_preview = reply_content_preview + record.response_time_ms = response_time_ms + record.replied_at = now + await session.commit() + return record + + async def mark_error( + self, record_id: int, error_message: str, response_time_ms: int | None = None + ) -> ChannelMsgRecord | None: + now = utc_now_naive() + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelMsgRecord).where(ChannelMsgRecord.id == record_id)) + record = result.scalar_one_or_none() + if record is None: + return None + record.status = "error" + record.error_message = error_message + record.response_time_ms = response_time_ms + record.replied_at = now + await session.commit() + return record + + async def list_by_chat(self, channel_id: str, chat_id: str, limit: int = 50) -> list[ChannelMsgRecord]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelMsgRecord) + .where( + ChannelMsgRecord.channel_id == channel_id, + ChannelMsgRecord.chat_id == chat_id, + ) + .order_by(ChannelMsgRecord.created_at.desc()) + .limit(limit) + ) + return list(result.scalars().all()) diff --git a/backend/package/yuxi/repositories/channel_pairing_repo.py b/backend/package/yuxi/repositories/channel_pairing_repo.py new file mode 100644 index 00000000..f36cf7fd --- /dev/null +++ b/backend/package/yuxi/repositories/channel_pairing_repo.py @@ -0,0 +1,170 @@ +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelPairingRecord + + +class ChannelPairingRecordRepository: + async def get_by_id(self, record_id: str) -> ChannelPairingRecord | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelPairingRecord).where(ChannelPairingRecord.id == record_id)) + return result.scalar_one_or_none() + + async def find_by_channel_peer( + self, channel_type: str, peer_id: str, account_id: str = "default", status: str | None = None + ) -> list[ChannelPairingRecord]: + async with pg_manager.get_async_session_context() as session: + query = select(ChannelPairingRecord).where( + ChannelPairingRecord.channel_type == channel_type, + ChannelPairingRecord.account_id == account_id, + ChannelPairingRecord.peer_id == peer_id, + ) + if status: + query = query.where(ChannelPairingRecord.status == status) + result = await session.execute(query.order_by(ChannelPairingRecord.created_at.desc())) + return list(result.scalars().all()) + + async def find_pending( + self, channel_type: str, peer_id: str, account_id: str = "default" + ) -> list[ChannelPairingRecord]: + return await self.find_by_channel_peer(channel_type, peer_id, account_id, status="pending") + + async def find_pending_by_account(self, channel_type: str, account_id: str) -> list[ChannelPairingRecord]: + async with pg_manager.get_async_session_context() as session: + now = datetime.now(UTC).replace(tzinfo=None) + result = await session.execute( + select(ChannelPairingRecord) + .where( + ChannelPairingRecord.channel_type == channel_type, + ChannelPairingRecord.account_id == account_id, + ChannelPairingRecord.status == "pending", + ChannelPairingRecord.expires_at > now, + ) + .order_by(ChannelPairingRecord.created_at.desc()) + ) + return list(result.scalars().all()) + + async def mark_expired(self) -> int: + now = datetime.now(UTC).replace(tzinfo=None) + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelPairingRecord).where( + ChannelPairingRecord.status == "pending", + ChannelPairingRecord.expires_at < now, + ) + ) + records = result.scalars().all() + for record in records: + record.status = "expired" + if records: + await session.commit() + return len(records) + + async def upsert_pending( + self, + channel_type: str, + account_id: str, + peer_id: str, + code: str, + token: str, + expires_at: datetime, + ) -> ChannelPairingRecord: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelPairingRecord).where( + ChannelPairingRecord.channel_type == channel_type, + ChannelPairingRecord.account_id == account_id, + ChannelPairingRecord.peer_id == peer_id, + ChannelPairingRecord.status == "pending", + ) + ) + record = result.scalar_one_or_none() + if record is not None: + record.pairing_code = code + record.pairing_token = token + record.expires_at = expires_at + else: + record = ChannelPairingRecord( + channel_type=channel_type, + account_id=account_id, + peer_id=peer_id, + pairing_code=code, + pairing_token=token, + expires_at=expires_at, + ) + session.add(record) + await session.commit() + await session.refresh(record) + return record + + async def create(self, data: dict[str, Any]) -> ChannelPairingRecord: + async with pg_manager.get_async_session_context() as session: + record = ChannelPairingRecord(**data) + session.add(record) + await session.commit() + await session.refresh(record) + return record + + async def update_status( + self, record_id: str, status: str, paired_at: datetime | None = None + ) -> ChannelPairingRecord | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelPairingRecord).where(ChannelPairingRecord.id == record_id)) + record = result.scalar_one_or_none() + if record is None: + return None + record.status = status + if paired_at is not None: + record.paired_at = paired_at + await session.commit() + await session.refresh(record) + return record + + async def approve(self, record_id: str) -> ChannelPairingRecord | None: + return await self.update_status(record_id, "paired", datetime.now(UTC).replace(tzinfo=None)) + + async def reject(self, record_id: str) -> ChannelPairingRecord | None: + return await self.update_status(record_id, "expired") + + async def remove_pending( + self, channel_type: str, account_id: str, peer_id: str, target_status: str = "paired" + ) -> bool: + async with pg_manager.get_async_session_context() as session: + now = datetime.now(UTC).replace(tzinfo=None) + result = await session.execute( + select(ChannelPairingRecord).where( + ChannelPairingRecord.channel_type == channel_type, + ChannelPairingRecord.account_id == account_id, + ChannelPairingRecord.peer_id == peer_id, + ChannelPairingRecord.status == "pending", + ) + ) + record = result.scalar_one_or_none() + if record is None: + return False + record.status = target_status + if target_status == "paired": + record.paired_at = now + await session.commit() + return True + + async def delete(self, record_id: str) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelPairingRecord).where(ChannelPairingRecord.id == record_id)) + record = result.scalar_one_or_none() + if record is None: + return False + await session.delete(record) + await session.commit() + return True + + async def list_all(self, status: str | None = None) -> list[ChannelPairingRecord]: + async with pg_manager.get_async_session_context() as session: + query = select(ChannelPairingRecord) + if status: + query = query.where(ChannelPairingRecord.status == status) + result = await session.execute(query.order_by(ChannelPairingRecord.created_at.desc())) + return list(result.scalars().all()) diff --git a/backend/package/yuxi/repositories/channel_thread_mapping_repo.py b/backend/package/yuxi/repositories/channel_thread_mapping_repo.py new file mode 100644 index 00000000..fe04d0ed --- /dev/null +++ b/backend/package/yuxi/repositories/channel_thread_mapping_repo.py @@ -0,0 +1,106 @@ +import uuid +from datetime import datetime +from typing import Any + +from sqlalchemy import select + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelThreadMapping + + +class ChannelThreadMappingRepository: + async def get_by_id(self, mapping_id: int) -> ChannelThreadMapping | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelThreadMapping).where(ChannelThreadMapping.id == mapping_id)) + return result.scalar_one_or_none() + + async def find_by_channel_chat( + self, channel_id: str, channel_chat_id: str, internal_user_id: str + ) -> ChannelThreadMapping | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelThreadMapping).where( + ChannelThreadMapping.channel_id == channel_id, + ChannelThreadMapping.channel_chat_id == channel_chat_id, + ChannelThreadMapping.internal_user_id == internal_user_id, + ) + ) + return result.scalar_one_or_none() + + async def find_by_thread_id(self, thread_id: str) -> ChannelThreadMapping | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelThreadMapping).where(ChannelThreadMapping.thread_id == thread_id) + ) + return result.scalar_one_or_none() + + async def resolve_thread( + self, + channel_id: str, + channel_chat_id: str, + internal_user_id: str, + agent_id: str | None = None, + thread_id: str | None = None, + ) -> ChannelThreadMapping: + now = datetime.now() + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelThreadMapping).where( + ChannelThreadMapping.channel_id == channel_id, + ChannelThreadMapping.channel_chat_id == channel_chat_id, + ChannelThreadMapping.internal_user_id == internal_user_id, + ) + ) + mapping = result.scalar_one_or_none() + if mapping is not None: + mapping.last_active_at = now + mapping.agent_id = agent_id or mapping.agent_id + await session.commit() + await session.refresh(mapping) + return mapping + + mapping = ChannelThreadMapping( + channel_id=channel_id, + channel_chat_id=channel_chat_id, + internal_user_id=internal_user_id, + thread_id=thread_id or uuid.uuid4().hex, + agent_id=agent_id, + last_active_at=now, + ) + session.add(mapping) + await session.commit() + await session.refresh(mapping) + return mapping + + async def touch_active(self, thread_id: str) -> bool: + from datetime import datetime + + now = datetime.now() + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelThreadMapping).where(ChannelThreadMapping.thread_id == thread_id) + ) + mapping = result.scalar_one_or_none() + if mapping is None: + return False + mapping.last_active_at = now + await session.commit() + return True + + async def create(self, data: dict[str, Any]) -> ChannelThreadMapping: + async with pg_manager.get_async_session_context() as session: + mapping = ChannelThreadMapping(**data) + session.add(mapping) + await session.commit() + await session.refresh(mapping) + return mapping + + async def delete(self, mapping_id: int) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelThreadMapping).where(ChannelThreadMapping.id == mapping_id)) + mapping = result.scalar_one_or_none() + if mapping is None: + return False + await session.delete(mapping) + await session.commit() + return True diff --git a/backend/package/yuxi/repositories/channel_user_mapping_repo.py b/backend/package/yuxi/repositories/channel_user_mapping_repo.py new file mode 100644 index 00000000..4e47edd6 --- /dev/null +++ b/backend/package/yuxi/repositories/channel_user_mapping_repo.py @@ -0,0 +1,124 @@ +import hashlib +import secrets +from typing import Any + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError + +from yuxi.storage.postgres.manager import pg_manager +from yuxi.storage.postgres.models_channels import ChannelUserMapping + + +def _make_channel_user_id(channel_id: str, channel_user_id: str) -> str: + unique_key = f"{channel_id}:{channel_user_id}" + uid_hash = hashlib.sha256(unique_key.encode()).hexdigest()[:16] + return f"ch_{uid_hash}" + + +class ChannelUserMappingRepository: + async def get_by_id(self, mapping_id: int) -> ChannelUserMapping | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelUserMapping).where(ChannelUserMapping.id == mapping_id)) + return result.scalar_one_or_none() + + async def find_by_channel(self, channel_id: str, channel_user_id: str) -> ChannelUserMapping | None: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelUserMapping).where( + ChannelUserMapping.channel_id == channel_id, + ChannelUserMapping.channel_user_id == channel_user_id, + ) + ) + return result.scalar_one_or_none() + + async def resolve_user(self, channel_id: str, channel_user_id: str) -> ChannelUserMapping: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelUserMapping).where( + ChannelUserMapping.channel_id == channel_id, + ChannelUserMapping.channel_user_id == channel_user_id, + ) + ) + mapping = result.scalar_one_or_none() + if mapping is not None: + return mapping + + from server.utils.auth_utils import AuthUtils + from yuxi.storage.postgres.models_business import User + + user_id = _make_channel_user_id(channel_id, channel_user_id) + username = user_id + + existing = await session.execute(select(User.id).where(User.user_id == user_id)) + if existing.scalar_one_or_none() is not None: + username = f"{user_id}_{secrets.token_hex(4)}" + + random_password = secrets.token_urlsafe(32) + password_hash = AuthUtils.hash_password(random_password) + + user = User( + username=username, + user_id=user_id, + password_hash=password_hash, + role="user", + source=f"channel:{channel_id}", + ) + session.add(user) + try: + await session.flush() + except IntegrityError: + await session.rollback() + existing_user = await session.execute(select(User).where(User.user_id == user_id)) + user = existing_user.scalar_one() + + mapping = ChannelUserMapping( + channel_id=channel_id, + channel_user_id=channel_user_id, + internal_user_id=str(user.id), + ) + session.add(mapping) + await session.commit() + await session.refresh(mapping) + return mapping + + async def find_by_internal_user(self, internal_user_id: str) -> list[ChannelUserMapping]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelUserMapping).where(ChannelUserMapping.internal_user_id == internal_user_id) + ) + return list(result.scalars().all()) + + async def list_all(self) -> list[ChannelUserMapping]: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelUserMapping)) + return list(result.scalars().all()) + + async def create(self, data: dict[str, Any]) -> ChannelUserMapping: + channel_id = data["channel_id"] + channel_user_id = data["channel_user_id"] + async with pg_manager.get_async_session_context() as session: + result = await session.execute( + select(ChannelUserMapping).where( + ChannelUserMapping.channel_id == channel_id, + ChannelUserMapping.channel_user_id == channel_user_id, + ) + ) + existing = result.scalar_one_or_none() + if existing is not None: + return existing + + mapping = ChannelUserMapping(**data) + session.add(mapping) + await session.commit() + await session.refresh(mapping) + return mapping + + async def delete(self, mapping_id: int) -> bool: + async with pg_manager.get_async_session_context() as session: + result = await session.execute(select(ChannelUserMapping).where(ChannelUserMapping.id == mapping_id)) + mapping = result.scalar_one_or_none() + if mapping is None: + return False + await session.delete(mapping) + await session.commit() + return True diff --git a/backend/package/yuxi/repositories/conversation_repository.py b/backend/package/yuxi/repositories/conversation_repository.py index 81d3b42e..f1623078 100644 --- a/backend/package/yuxi/repositories/conversation_repository.py +++ b/backend/package/yuxi/repositories/conversation_repository.py @@ -4,7 +4,8 @@ import uuid as uuid_lib -from sqlalchemy import select +from sqlalchemy import select, update +from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -72,6 +73,52 @@ class ConversationRepository: result = await self.db.execute(select(Conversation).where(Conversation.thread_id == thread_id)) return result.scalar_one_or_none() + async def upsert_conversation( + self, + thread_id: str, + user_id: str, + agent_id: str, + title: str | None = None, + metadata: dict | None = None, + update_metadata: dict | None = None, + ) -> Conversation: + metadata = (metadata or {}).copy() + metadata.setdefault("attachments", []) + + normalized_title = self._normalize_title(title) + + insert_stmt = insert(Conversation).values( + thread_id=thread_id, + user_id=str(user_id), + agent_id=agent_id, + title=normalized_title or "New Conversation", + status="active", + extra_metadata=metadata, + ) + + update_values: dict = {"updated_at": utc_now_naive()} + if update_metadata is not None: + current_meta = Conversation.extra_metadata + update_values["extra_metadata"] = current_meta.op("||")(update_metadata) + + upsert_stmt = insert_stmt.on_conflict_do_update( + index_elements=["thread_id"], + set_=update_values, + ).returning(Conversation) + + result = await self.db.execute(upsert_stmt) + conversation = result.scalar_one() + + stats_stmt = insert(ConversationStats).values( + conversation_id=conversation.id, + ).on_conflict_do_nothing() + await self.db.execute(stats_stmt) + await self.db.commit() + await self.db.refresh(conversation) + + logger.info(f"Upserted conversation: {conversation.thread_id} for user {user_id}") + return conversation + async def _get_conversation_by_id(self, conversation_id: int) -> Conversation | None: result = await self.db.execute(select(Conversation).where(Conversation.id == conversation_id)) return result.scalar_one_or_none() @@ -285,24 +332,30 @@ class ConversationRepository: metadata: dict | None = None, is_pinned: bool | None = None, ) -> Conversation | None: - conversation = await self.get_conversation_by_thread_id(thread_id) - if not conversation: - return None + values: dict = {"updated_at": utc_now_naive()} normalized_title = self._normalize_title(title) if normalized_title is not None: - conversation.title = normalized_title + values["title"] = normalized_title if status is not None: - conversation.status = status + values["status"] = status if is_pinned is not None: - conversation.is_pinned = is_pinned - + values["is_pinned"] = is_pinned if metadata is not None: - current_metadata = conversation.extra_metadata or {} - current_metadata.update(metadata) - conversation.extra_metadata = current_metadata + values["extra_metadata"] = Conversation.extra_metadata.op("||")(metadata) + + stmt = ( + update(Conversation) + .where(Conversation.thread_id == thread_id) + .values(**values) + .returning(Conversation) + ) + + result = await self.db.execute(stmt) + conversation = result.scalar_one_or_none() + if conversation is None: + return None - conversation.updated_at = utc_now_naive() await self.db.commit() await self.db.refresh(conversation) @@ -461,3 +514,25 @@ class ConversationRepository: metadata["attachments"] = new_attachments await self._save_metadata(conversation, metadata) return True + + async def get_conversations_by_metadata_key(self, key: str, value: str) -> list[Conversation]: + result = await self.db.execute( + select(Conversation) + .where(Conversation.extra_metadata[key].astext == value) + .order_by(Conversation.updated_at.desc()) + ) + return list(result.scalars().all()) + + async def get_conversations_by_channel( + self, channel_type: str, account_id: str | None = None + ) -> list[Conversation]: + conditions = [Conversation.extra_metadata["channel_type"].astext == channel_type] + if account_id: + conditions.append(Conversation.extra_metadata["account_id"].astext == account_id) + + result = await self.db.execute( + select(Conversation) + .where(*conditions) + .order_by(Conversation.updated_at.desc()) + ) + return list(result.scalars().all()) diff --git a/backend/package/yuxi/services/channel_service.py b/backend/package/yuxi/services/channel_service.py new file mode 100644 index 00000000..33e15a6e --- /dev/null +++ b/backend/package/yuxi/services/channel_service.py @@ -0,0 +1,244 @@ +from typing import Any + +from yuxi.channel.runtime.manager import gateway +from yuxi.channel.security.pairing import PairingManager +from yuxi.repositories.channel_binding_repo import ChannelBindingRepository +from yuxi.repositories.channel_config_repo import ChannelConfigRepository +from yuxi.repositories.channel_pairing_repo import ChannelPairingRecordRepository +from yuxi.repositories.channel_thread_mapping_repo import ChannelThreadMappingRepository +from yuxi.repositories.channel_user_mapping_repo import ChannelUserMappingRepository + +_PAIRING_PEER_ID_PREFIX = "admin:pairing" + +_config_repo = ChannelConfigRepository() +_binding_repo = ChannelBindingRepository() +_pairing_repo = ChannelPairingRecordRepository() +_user_mapping_repo = ChannelUserMappingRepository() +_thread_mapping_repo = ChannelThreadMappingRepository() + + +async def list_channel_configs( + channel_type: str | None = None, + enabled: bool | None = None, +) -> list[dict[str, Any]]: + if channel_type and enabled is not None: + configs = await _config_repo.list_enabled(channel_type) + elif enabled is not None: + configs = await _config_repo.list_enabled() + elif channel_type: + configs = await _config_repo.list_by_type(channel_type) + else: + configs = await _config_repo.list_all() + return [c.to_dict(mask_secrets=True) for c in configs] + + +async def get_channel_config(config_id: str) -> dict[str, Any] | None: + config = await _config_repo.get_by_id(config_id) + return config.to_dict(mask_secrets=True) if config else None + + +async def create_channel_config(data: dict[str, Any]) -> dict[str, Any]: + config = await _config_repo.create(data) + return config.to_dict(mask_secrets=True) + + +async def update_channel_config(config_id: str, data: dict[str, Any]) -> dict[str, Any] | None: + config = await _config_repo.update(config_id, data) + return config.to_dict(mask_secrets=True) if config else None + + +async def delete_channel_config(config_id: str) -> bool: + return await _config_repo.delete(config_id) + + +async def list_channel_bindings(agent_config_id: int | None = None) -> list[dict[str, Any]]: + if agent_config_id is not None: + bindings = await _binding_repo.list_by_agent(agent_config_id) + else: + bindings = await _binding_repo.list_all() + return [b.to_dict() for b in bindings] + + +async def get_channel_binding(binding_id: str) -> dict[str, Any] | None: + binding = await _binding_repo.get_by_id(binding_id) + return binding.to_dict() if binding else None + + +async def create_channel_binding(data: dict[str, Any]) -> dict[str, Any]: + binding = await _binding_repo.create(data) + return binding.to_dict() + + +async def update_channel_binding(binding_id: str, data: dict[str, Any]) -> dict[str, Any] | None: + binding = await _binding_repo.update(binding_id, data) + return binding.to_dict() if binding else None + + +async def delete_channel_binding(binding_id: str) -> bool: + return await _binding_repo.delete(binding_id) + + +async def list_channel_bindings_by_type(channel_type: str) -> list[dict[str, Any]]: + bindings = await _binding_repo.list_by_channel_type(channel_type) + return [b.to_dict() for b in bindings] + + +async def list_channel_pairing_records(status: str | None = None) -> list[dict[str, Any]]: + records = await _pairing_repo.list_all(status=status) + return [r.to_dict() for r in records] + + +async def approve_channel_pairing(record_id: str) -> dict[str, Any] | None: + record = await _pairing_repo.approve(record_id) + return record.to_dict() if record else None + + +async def reject_channel_pairing(record_id: str) -> dict[str, Any] | None: + record = await _pairing_repo.reject(record_id) + return record.to_dict() if record else None + + +async def get_channel_pairing_record(record_id: str) -> dict[str, Any] | None: + record = await _pairing_repo.get_by_id(record_id) + return record.to_dict() if record else None + + +async def delete_channel_pairing_record(record_id: str) -> bool: + return await _pairing_repo.delete(record_id) + + +async def generate_agent_pairing_code(agent_config_id: int) -> dict[str, Any]: + bindings = await _binding_repo.list_by_agent(agent_config_id) + if not bindings: + raise ValueError(f"Agent config {agent_config_id} has no channel bindings") + + binding = bindings[0] + mgr = PairingManager() + peer_id = f"{_PAIRING_PEER_ID_PREFIX}:{agent_config_id}" + + req = await mgr.upsert_code( + channel_type=binding.channel_type, + peer_id=peer_id, + account_id=binding.account_id or "default", + ) + + return { + "code": req.code, + "channel_type": binding.channel_type, + "account_id": binding.account_id or "default", + "agent_config_id": agent_config_id, + "record_id": req._record_id, + } + + +async def list_user_mappings() -> list[dict[str, Any]]: + mappings = await _user_mapping_repo.list_all() + return [m.to_dict() for m in mappings] + + +async def get_user_mapping(mapping_id: int) -> dict[str, Any] | None: + mapping = await _user_mapping_repo.get_by_id(mapping_id) + return mapping.to_dict() if mapping else None + + +async def find_user_mapping_by_channel(channel_id: str, channel_user_id: str) -> dict[str, Any] | None: + mapping = await _user_mapping_repo.find_by_channel(channel_id, channel_user_id) + return mapping.to_dict() if mapping else None + + +async def resolve_channel_user(channel_id: str, channel_user_id: str) -> dict[str, Any]: + mapping = await _user_mapping_repo.resolve_user(channel_id, channel_user_id) + return mapping.to_dict() + + +async def find_user_mappings_by_internal_user(internal_user_id: str) -> list[dict[str, Any]]: + mappings = await _user_mapping_repo.find_by_internal_user(internal_user_id) + return [m.to_dict() for m in mappings] + + +async def create_user_mapping(data: dict[str, Any]) -> dict[str, Any]: + mapping = await _user_mapping_repo.create(data) + return mapping.to_dict() + + +async def delete_user_mapping(mapping_id: int) -> bool: + return await _user_mapping_repo.delete(mapping_id) + + +async def resolve_channel_thread( + channel_id: str, + channel_chat_id: str, + internal_user_id: str, + agent_id: str | None = None, +) -> dict[str, Any]: + mapping = await _thread_mapping_repo.resolve_thread( + channel_id, + channel_chat_id, + internal_user_id, + agent_id, + ) + return mapping.to_dict() + + +async def get_thread_mapping(mapping_id: int) -> dict[str, Any] | None: + mapping = await _thread_mapping_repo.get_by_id(mapping_id) + return mapping.to_dict() if mapping else None + + +async def find_thread_mapping_by_channel_chat( + channel_id: str, + channel_chat_id: str, + internal_user_id: str, +) -> dict[str, Any] | None: + mapping = await _thread_mapping_repo.find_by_channel_chat( + channel_id, + channel_chat_id, + internal_user_id, + ) + return mapping.to_dict() if mapping else None + + +async def find_thread_mapping_by_thread_id(thread_id: str) -> dict[str, Any] | None: + mapping = await _thread_mapping_repo.find_by_thread_id(thread_id) + return mapping.to_dict() if mapping else None + + +async def create_thread_mapping(data: dict[str, Any]) -> dict[str, Any]: + mapping = await _thread_mapping_repo.create(data) + return mapping.to_dict() + + +async def delete_thread_mapping(mapping_id: int) -> bool: + return await _thread_mapping_repo.delete(mapping_id) + + +async def get_gateway_health() -> dict[str, Any]: + report = gateway.get_health() + return { + "status": report.status, + "channels": report.channels, + "summary": report.summary, + "boot": gateway.get_boot_status().to_dict(), + } + + +async def get_gateway_healthz() -> dict[str, Any]: + return {"status": "ok"} + + +async def get_hook_status() -> dict[str, Any]: + result = gateway.get_hook_status() + return result.to_dict() + + +async def reload_hooks() -> dict[str, Any]: + result = gateway.reload_hooks() + return result.to_dict() + + +async def install_hook(hook_id: str, hook_md_content: str, target: str = "workspace") -> dict[str, Any]: + return gateway.install_hook(hook_id, hook_md_content, target) + + +async def uninstall_hook(source: str, hook_id: str) -> dict[str, Any]: + return gateway.uninstall_hook(source, hook_id) diff --git a/backend/package/yuxi/services/oidc_service.py b/backend/package/yuxi/services/oidc_service.py index a57ff60f..a630372b 100644 --- a/backend/package/yuxi/services/oidc_service.py +++ b/backend/package/yuxi/services/oidc_service.py @@ -590,6 +590,7 @@ async def _create_oidc_binding_placeholder(db, sub: str, target_user: User) -> N password_hash=password_hash, role=target_user.role, department_id=target_user.department_id, + source="oidc", is_deleted=1, # 标记为deleted，不参与实际登录 last_login=utc_now_naive(), ) @@ -692,6 +693,7 @@ async def create_oidc_user(db, user_info: dict, department_id: int | None = None "role": oidc_config.default_role, "department_id": department_id, "last_login": utc_now_naive(), + "source": "oidc", } ) logger.info(f"Created OIDC user: {new_user.username} ({user_id})") @@ -880,6 +882,7 @@ async def oidc_callback_handler(code: str, state: str, db, request: Request | No "role": user.role, "department_id": user.department_id, "department_name": department_name, + "source": user.source, } exchange_code = OIDCUtils.generate_login_code(response_data) diff --git a/backend/package/yuxi/storage/postgres/manager.py b/backend/package/yuxi/storage/postgres/manager.py index 36e853f5..a8219319 100644 --- a/backend/package/yuxi/storage/postgres/manager.py +++ b/backend/package/yuxi/storage/postgres/manager.py @@ -4,6 +4,7 @@ import json import os from contextlib import asynccontextmanager +import yuxi.storage.postgres.models_channels as _ # 触发 SQLAlchemy Base.metadata 注册 from psycopg_pool import AsyncConnectionPool from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine @@ -249,6 +250,180 @@ class PostgresManager(metaclass=SingletonMeta): for stmt in stmts: await conn.execute(text(stmt)) + async def ensure_channels_schema(self): + self._check_initialized() + stmts = [ + """CREATE TABLE IF NOT EXISTS channel_configs ( + id VARCHAR(64) PRIMARY KEY, + channel_type VARCHAR(50) NOT NULL, + name VARCHAR(100) NOT NULL, + config JSONB NOT NULL DEFAULT '{}', + enabled BOOLEAN DEFAULT TRUE, + created_by VARCHAR(64), + updated_by VARCHAR(64), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() + )""", + """CREATE INDEX IF NOT EXISTS idx_channel_configs_type + ON channel_configs (channel_type)""", + """CREATE INDEX IF NOT EXISTS idx_channel_configs_type_enabled + ON channel_configs (channel_type, enabled)""", + """CREATE TABLE IF NOT EXISTS channel_bindings ( + id VARCHAR(64) PRIMARY KEY, + channel_type VARCHAR(50) NOT NULL, + channel_config_id VARCHAR(64) REFERENCES channel_configs(id) ON DELETE SET NULL, + account_id VARCHAR(100), + peer_kind VARCHAR(20), + peer_id VARCHAR(200), + agent_config_id INTEGER NOT NULL REFERENCES agent_configs(id) ON DELETE CASCADE, + priority INTEGER DEFAULT 0, + dm_scope VARCHAR(50) DEFAULT 'per-channel-peer', + created_by VARCHAR(64), + updated_by VARCHAR(64), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() + )""", + """CREATE INDEX IF NOT EXISTS idx_channel_bindings_agent + ON channel_bindings (agent_config_id)""", + """CREATE INDEX IF NOT EXISTS idx_channel_bindings_lookup + ON channel_bindings (channel_type, account_id, peer_kind, peer_id)""", + """CREATE INDEX IF NOT EXISTS idx_channel_bindings_type + ON channel_bindings (channel_type)""", + """CREATE INDEX IF NOT EXISTS idx_channel_bindings_config + ON channel_bindings (channel_config_id)""", + """ALTER TABLE channel_bindings + ADD COLUMN IF NOT EXISTS guild_id VARCHAR(100)""", + """ALTER TABLE channel_bindings + ADD COLUMN IF NOT EXISTS team_id VARCHAR(100)""", + """ALTER TABLE channel_bindings + ADD COLUMN IF NOT EXISTS roles JSONB DEFAULT '[]'""", + """ALTER TABLE channel_bindings + ADD COLUMN IF NOT EXISTS session_dm_scope VARCHAR(50)""", + """CREATE TABLE IF NOT EXISTS channel_pairing_records ( + id VARCHAR(64) PRIMARY KEY, + channel_type VARCHAR(50) NOT NULL, + account_id VARCHAR(100) NOT NULL DEFAULT 'default', + peer_id VARCHAR(200) NOT NULL, + agent_config_id INTEGER REFERENCES agent_configs(id) ON DELETE SET NULL, + pairing_code VARCHAR(8) NOT NULL, + pairing_token VARCHAR(64) NOT NULL, + status VARCHAR(20) DEFAULT 'pending', + expires_at TIMESTAMP WITH TIME ZONE NOT NULL, + paired_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() + )""", + """CREATE INDEX IF NOT EXISTS idx_channel_pairing_records_lookup + ON channel_pairing_records (channel_type, account_id, peer_id, status)""", + """CREATE INDEX IF NOT EXISTS idx_channel_pairing_records_expiry + ON channel_pairing_records (expires_at)""", + """CREATE INDEX IF NOT EXISTS idx_channel_pairing_records_agent + ON channel_pairing_records (agent_config_id)""", + """CREATE TABLE IF NOT EXISTS channel_user_mappings ( + id BIGSERIAL PRIMARY KEY, + channel_id VARCHAR(32) NOT NULL, + channel_user_id VARCHAR(128) NOT NULL, + internal_user_id VARCHAR(64) NOT NULL, + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + CONSTRAINT uq_channel_user UNIQUE (channel_id, channel_user_id) + )""", + """CREATE INDEX IF NOT EXISTS idx_channel_user_internal + ON channel_user_mappings (internal_user_id)""", + """CREATE TABLE IF NOT EXISTS channel_thread_mappings ( + id BIGSERIAL PRIMARY KEY, + channel_id VARCHAR(32) NOT NULL, + channel_chat_id VARCHAR(128) NOT NULL, + internal_user_id VARCHAR(64) NOT NULL, + thread_id VARCHAR(64) NOT NULL, + agent_id VARCHAR(64), + last_active_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + CONSTRAINT uq_channel_thread UNIQUE (channel_id, channel_chat_id, internal_user_id) + )""", + """CREATE INDEX IF NOT EXISTS idx_channel_thread_thread + ON channel_thread_mappings (thread_id)""", + """CREATE INDEX IF NOT EXISTS idx_channel_thread_active + ON channel_thread_mappings (last_active_at)""", + """CREATE TABLE IF NOT EXISTS channel_msg_records ( + id BIGSERIAL PRIMARY KEY, + channel_id VARCHAR(32) NOT NULL, + channel_type VARCHAR(32) NOT NULL, + message_id VARCHAR(128) NOT NULL, + chat_id VARCHAR(128) NOT NULL, + chat_type VARCHAR(32) NOT NULL DEFAULT 'direct', + content_type VARCHAR(32) NOT NULL DEFAULT 'text', + sender_user_id VARCHAR(128) NOT NULL, + content_preview VARCHAR(500) NOT NULL, + reply_to_message_id VARCHAR(128), + agent_config_id INTEGER, + status VARCHAR(32) NOT NULL DEFAULT 'processing', + error_message VARCHAR(500), + response_time_ms INTEGER, + reply_message_id VARCHAR(128), + reply_content_preview VARCHAR(500), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + replied_at TIMESTAMP WITH TIME ZONE, + extra_data JSONB + )""", + """CREATE INDEX IF NOT EXISTS idx_msg_records_channel_status + ON channel_msg_records (channel_id, status)""", + """CREATE INDEX IF NOT EXISTS idx_msg_records_created + ON channel_msg_records (created_at)""", + """CREATE INDEX IF NOT EXISTS idx_msg_records_chat + ON channel_msg_records (channel_id, chat_id)""", + """ALTER TABLE IF EXISTS users + ADD COLUMN IF NOT EXISTS source VARCHAR(32) NOT NULL DEFAULT 'local'""", + """CREATE TABLE IF NOT EXISTS channel_device_identities ( + id VARCHAR(64) PRIMARY KEY, + device_id VARCHAR(64) NOT NULL, + name VARCHAR(255) NOT NULL, + public_key_pem TEXT NOT NULL, + status VARCHAR(20) DEFAULT 'active', + created_by VARCHAR(64), + last_used_at TIMESTAMP WITH TIME ZONE, + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() + )""", + """CREATE INDEX IF NOT EXISTS idx_channel_device_identities_device_id + ON channel_device_identities (device_id)""", + """CREATE INDEX IF NOT EXISTS idx_channel_device_identities_status + ON channel_device_identities (status)""", + """CREATE TABLE IF NOT EXISTS channel_identity_links ( + id VARCHAR(64) PRIMARY KEY, + identity VARCHAR(255) NOT NULL, + channel_type VARCHAR(64) NOT NULL, + peer_id VARCHAR(255) NOT NULL, + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + CONSTRAINT uq_identity_link_channel_peer UNIQUE (channel_type, peer_id) + )""", + "CREATE INDEX IF NOT EXISTS idx_identity_links_identity ON channel_identity_links(identity)", + """CREATE TABLE IF NOT EXISTS pay_notifications ( + id BIGSERIAL PRIMARY KEY, + notification_id VARCHAR(128) NOT NULL, + mch_id VARCHAR(32) NOT NULL, + event_type VARCHAR(64) NOT NULL, + out_trade_no VARCHAR(64) DEFAULT '', + transaction_id VARCHAR(64) DEFAULT '', + resource_json JSONB DEFAULT '{}', + status VARCHAR(32) DEFAULT 'pending', + error_message TEXT, + retry_count INTEGER DEFAULT 0, + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + completed_at TIMESTAMP WITH TIME ZONE, + CONSTRAINT uk_pay_notification_id UNIQUE (notification_id) + )""", + "CREATE INDEX IF NOT EXISTS idx_pay_notifications_mch_id ON pay_notifications(mch_id)", + "CREATE INDEX IF NOT EXISTS idx_pay_notifications_out_trade_no ON pay_notifications(out_trade_no)", + "CREATE INDEX IF NOT EXISTS idx_pay_notifications_status ON pay_notifications(status)", + "CREATE INDEX IF NOT EXISTS idx_pay_notifications_created_at ON pay_notifications(created_at)", + ] + + async with self.async_engine.begin() as conn: + for stmt in stmts: + await conn.execute(text(stmt)) + @property def is_postgresql(self) -> bool: """检查是否是 PostgreSQL 数据库""" diff --git a/backend/package/yuxi/storage/postgres/models_business.py b/backend/package/yuxi/storage/postgres/models_business.py index ef329965..ac2fd382 100644 --- a/backend/package/yuxi/storage/postgres/models_business.py +++ b/backend/package/yuxi/storage/postgres/models_business.py @@ -1,8 +1,11 @@ """PostgreSQL 业务数据模型 - 用户、部门、对话等相关表""" +import logging from datetime import timedelta from typing import Any +_logger = logging.getLogger(__name__) + from sqlalchemy import ( JSON, Boolean, @@ -17,6 +20,7 @@ from sqlalchemy import ( UniqueConstraint, ) from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy.orm import relationship from yuxi.utils.datetime_utils import format_utc_datetime, utc_now_naive @@ -61,6 +65,9 @@ class User(Base): password_hash = Column(String, nullable=False) role = Column(String, nullable=False, default="user") # 角色: superadmin, admin, user department_id = Column(Integer, ForeignKey("departments.id"), nullable=True) # 部门ID + source = Column( + String(32), nullable=False, default="local", comment="用户来源: local / channel:telegram / channel:feishu / ..." + ) created_at = Column(DateTime, default=utc_now_naive) last_login = Column(DateTime, nullable=True) @@ -91,6 +98,7 @@ class User(Base): "avatar": self.avatar, "role": self.role, "department_id": self.department_id, + "source": self.source, "created_at": format_utc_datetime(self.created_at), "last_login": format_utc_datetime(self.last_login), "login_failed_count": self.login_failed_count, @@ -543,7 +551,7 @@ class ModelProvider(Base): embedding_models_endpoint = Column(String(200), nullable=True, comment="Embedding 模型列表端点") rerank_models_endpoint = Column(String(200), nullable=True, comment="Rerank 模型列表端点") api_key_env = Column(String(128), nullable=True, comment="API Key 环境变量名") - api_key = Column(String(500), nullable=True, comment="直接配置的 API Key") + _api_key = Column("api_key", String(1000), nullable=True, comment="直接配置的 API Key (加密存储)") capabilities = Column(JSON, nullable=False, default=list, comment="支持能力：chat/embedding/rerank") enabled_models = Column(JSON, nullable=False, default=list, comment="已启用模型配置对象") @@ -558,6 +566,34 @@ class ModelProvider(Base): created_at = Column(DateTime, default=utc_now_naive, comment="创建时间") updated_at = Column(DateTime, default=utc_now_naive, onupdate=utc_now_naive, comment="更新时间") + @hybrid_property + def api_key(self) -> str | None: + from yuxi.channel.secrets.crypto import CryptoError, decrypt_secret + + if self._api_key: + try: + return decrypt_secret(self._api_key) + except CryptoError: + _logger.warning( + "Failed to decrypt api_key for provider %s", + self.provider_id, + ) + return None + return None + + @api_key.setter # type: ignore[no-redef] + def api_key(self, value: str | None) -> None: + from yuxi.channel.secrets.crypto import CryptoError, encrypt_secret, is_encrypted + + if value and not is_encrypted(value): + try: + self._api_key = encrypt_secret(value) + except CryptoError as e: + _logger.error("Failed to encrypt api_key: %s", e) + self._api_key = value + else: + self._api_key = value + def to_dict(self) -> dict[str, Any]: return { "id": self.id, diff --git a/backend/package/yuxi/storage/postgres/models_channels.py b/backend/package/yuxi/storage/postgres/models_channels.py new file mode 100644 index 00000000..4a54d297 --- /dev/null +++ b/backend/package/yuxi/storage/postgres/models_channels.py @@ -0,0 +1,462 @@ +import logging +import uuid +from typing import Any + +from sqlalchemy import ( + JSON, + BigInteger, + Boolean, + Column, + DateTime, + ForeignKey, + Index, + Integer, + String, + Text, + UniqueConstraint, +) +from yuxi.storage.postgres.models_business import Base +from yuxi.utils.datetime_utils import format_utc_datetime, utc_now_naive + +_logger = logging.getLogger(__name__) + +_SECRET_FIELD_NAMES = frozenset( + { + "app_secret", + "bot_token", + "signing_secret", + "encrypt_key", + "verification_token", + "api_key", + "api_secret", + "access_token", + "refresh_token", + "webhook_token", + "client_secret", + "secret", + "password", + "passwd", + "auth_token", + "app_key", + "license_key", + "private_key", + "webhook_secret", + } +) + +_SECRET_MARKER = "$secret" + + +class ChannelConfig(Base): + __tablename__ = "channel_configs" + + id = Column(String(64), primary_key=True, default=lambda: uuid.uuid4().hex) + channel_type = Column(String(50), nullable=False, comment="渠道类型: feishu/dingtalk/wecom_bot/telegram/slack/imessage") + name = Column(String(100), nullable=False, comment="渠道账户名称") + config = Column(JSON, nullable=False, default=dict, comment="渠道特定配置") + enabled = Column(Boolean, nullable=False, default=True, comment="启用状态") + created_by = Column(String(64), nullable=True, comment="创建人") + updated_by = Column(String(64), nullable=True, comment="更新人") + created_at = Column(DateTime, default=utc_now_naive) + updated_at = Column(DateTime, default=utc_now_naive, onupdate=utc_now_naive) + + __table_args__ = ( + Index("idx_channel_configs_type", "channel_type"), + Index("idx_channel_configs_type_enabled", "channel_type", "enabled"), + ) + + def to_dict(self, mask_secrets: bool = True) -> dict[str, Any]: + config = dict(self.config or {}) + if mask_secrets: + _mask_secret_values(config) + return { + "id": self.id, + "channel_type": self.channel_type, + "name": self.name, + "config": config, + "enabled": bool(self.enabled), + "created_by": self.created_by, + "updated_by": self.updated_by, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + +class ChannelBinding(Base): + __tablename__ = "channel_bindings" + + id = Column(String(64), primary_key=True, default=lambda: uuid.uuid4().hex) + channel_type = Column(String(50), nullable=False, comment="渠道类型") + channel_config_id = Column( + String(64), + ForeignKey("channel_configs.id", ondelete="SET NULL"), + nullable=True, + comment="关联 channel_configs; NULL=渠道级通配绑定", + ) + account_id = Column(String(100), nullable=True, comment="渠道账户ID(对端视角), '*'=通配") + peer_kind = Column(String(20), nullable=True, comment="direct/group/channel") + peer_id = Column(String(200), nullable=True, comment="对端ID, '*'=通配") + agent_config_id = Column( + Integer, ForeignKey("agent_configs.id", ondelete="CASCADE"), nullable=False, comment="路由目标 AgentConfig" + ) + priority = Column(Integer, nullable=False, default=0, comment="优先级, 数值越大越优先") + dm_scope = Column( + String(50), + nullable=False, + default="per-channel-peer", + comment="DM Session 隔离粒度: main/per-peer/per-channel-peer/per-account-channel-peer", + ) + guild_id = Column(String(100), nullable=True, comment="Discord/Slack Guild ID") + team_id = Column(String(100), nullable=True, comment="Slack Team ID") + roles = Column(JSON, nullable=True, default=list, comment="角色列表，匹配 guild+roles 场景") + session_dm_scope = Column( + String(50), + nullable=True, + comment="单条绑定级 DM 隔离策略覆盖（覆盖 agent 级默认值）", + ) + created_by = Column(String(64), nullable=True, comment="创建人") + updated_by = Column(String(64), nullable=True, comment="更新人") + created_at = Column(DateTime, default=utc_now_naive) + updated_at = Column(DateTime, default=utc_now_naive, onupdate=utc_now_naive) + + __table_args__ = ( + Index("idx_channel_bindings_agent", "agent_config_id"), + Index("idx_channel_bindings_lookup", "channel_type", "account_id", "peer_kind", "peer_id"), + Index("idx_channel_bindings_type", "channel_type"), + Index("idx_channel_bindings_config", "channel_config_id"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_type": self.channel_type, + "channel_config_id": self.channel_config_id, + "account_id": self.account_id, + "peer_kind": self.peer_kind, + "peer_id": self.peer_id, + "agent_config_id": self.agent_config_id, + "priority": self.priority, + "dm_scope": self.dm_scope, + "guild_id": self.guild_id, + "team_id": self.team_id, + "roles": self.roles or [], + "session_dm_scope": self.session_dm_scope, + "created_by": self.created_by, + "updated_by": self.updated_by, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + +class ChannelPairingRecord(Base): + __tablename__ = "channel_pairing_records" + + id = Column(String(64), primary_key=True, default=lambda: uuid.uuid4().hex) + channel_type = Column(String(50), nullable=False, comment="渠道类型") + account_id = Column(String(100), nullable=False, default="default", comment="渠道账户ID") + peer_id = Column(String(200), nullable=False, comment="对端用户ID") + agent_config_id = Column( + Integer, + ForeignKey("agent_configs.id", ondelete="SET NULL"), + nullable=True, + comment="配对归属的 AgentConfig（路由解析后回填）", + ) + pairing_code = Column(String(8), nullable=False, comment="8位配对码") + pairing_token = Column(String(64), nullable=False, comment="配对令牌") + status = Column(String(20), nullable=False, default="pending", comment="pending/paired/expired") + expires_at = Column(DateTime, nullable=False, comment="过期时间") + paired_at = Column(DateTime, nullable=True, comment="配对完成时间") + created_at = Column(DateTime, default=utc_now_naive) + + __table_args__ = ( + Index("idx_channel_pairing_records_lookup", "channel_type", "account_id", "peer_id", "status"), + Index("idx_channel_pairing_records_expiry", "expires_at"), + Index("idx_channel_pairing_records_agent", "agent_config_id"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_type": self.channel_type, + "account_id": self.account_id, + "peer_id": self.peer_id, + "agent_config_id": self.agent_config_id, + "pairing_code": self.pairing_code, + "pairing_token": self.pairing_token, + "status": self.status, + "expires_at": format_utc_datetime(self.expires_at), + "paired_at": format_utc_datetime(self.paired_at) if self.paired_at else None, + "created_at": format_utc_datetime(self.created_at), + } + + +class ChannelUserMapping(Base): + __tablename__ = "channel_user_mappings" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + channel_id = Column(String(32), nullable=False, comment="渠道标识") + channel_user_id = Column(String(128), nullable=False, comment="渠道侧用户ID") + internal_user_id = Column(String(64), nullable=False, comment="内部用户ID") + 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__ = ( + UniqueConstraint("channel_id", "channel_user_id", name="uq_channel_user"), + Index("idx_channel_user_internal", "internal_user_id"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_id": self.channel_id, + "channel_user_id": self.channel_user_id, + "internal_user_id": self.internal_user_id, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + +class ChannelIdentityLink(Base): + __tablename__ = "channel_identity_links" + + id = Column(String(64), primary_key=True, default=lambda: uuid.uuid4().hex) + identity = Column(String(255), nullable=False, index=True, comment="统一身份标识") + channel_type = Column(String(64), nullable=False, comment="渠道类型") + peer_id = Column(String(255), nullable=False, comment="渠道侧用户/群组ID") + created_at = Column(DateTime, default=utc_now_naive) + updated_at = Column(DateTime, default=utc_now_naive, onupdate=utc_now_naive) + + __table_args__ = ( + UniqueConstraint("channel_type", "peer_id", name="uq_identity_link_channel_peer"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "identity": self.identity, + "channel_type": self.channel_type, + "peer_id": self.peer_id, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + +class ChannelAllowlistEntry(Base): + __tablename__ = "channel_allowlist_entries" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + channel_id = Column(String(32), nullable=False, comment="渠道标识") + allow_type = Column(String(32), nullable=False, default="user", comment="允许类型: user/group") + allow_id = Column(String(128), nullable=False, comment="允许的用户ID或群组ID") + enabled = Column(Boolean, nullable=False, default=True, comment="是否启用") + 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__ = ( + UniqueConstraint("channel_id", "allow_type", "allow_id", name="uq_channel_allowlist"), + Index("idx_channel_allowlist_channel", "channel_id"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_id": self.channel_id, + "allow_type": self.allow_type, + "allow_id": self.allow_id, + "enabled": self.enabled, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + +class ChannelThreadMapping(Base): + __tablename__ = "channel_thread_mappings" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + channel_id = Column(String(32), nullable=False, comment="渠道标识") + channel_chat_id = Column(String(128), nullable=False, comment="渠道侧会话/群组ID") + internal_user_id = Column(String(64), nullable=False, comment="内部用户ID") + thread_id = Column(String(64), nullable=False, comment="LangGraph thread_id (UUID)") + agent_id = Column(String(64), nullable=True, comment="绑定的 Agent ID") + last_active_at = Column(DateTime, nullable=False, default=utc_now_naive, comment="最后活跃时间") + 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__ = ( + UniqueConstraint("channel_id", "channel_chat_id", "internal_user_id", name="uq_channel_thread"), + Index("idx_channel_thread_thread", "thread_id"), + Index("idx_channel_thread_active", "last_active_at"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_id": self.channel_id, + "channel_chat_id": self.channel_chat_id, + "internal_user_id": self.internal_user_id, + "thread_id": self.thread_id, + "agent_id": self.agent_id, + "last_active_at": format_utc_datetime(self.last_active_at), + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } + + +class ChannelMsgRecord(Base): + __tablename__ = "channel_msg_records" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + channel_id = Column(String(32), nullable=False, comment="渠道标识") + channel_type = Column(String(32), nullable=False, comment="渠道类型枚举值") + message_id = Column(String(128), nullable=False, comment="渠道侧消息ID") + chat_id = Column(String(128), nullable=False, comment="渠道侧会话ID") + chat_type = Column(String(32), nullable=False, default="direct", comment="会话类型: direct/group/thread/forum") + content_type = Column(String(32), nullable=False, default="text", comment="消息类型: text/image/file/audio/video") + sender_user_id = Column(String(128), nullable=False, comment="发送者渠道用户ID") + content_preview = Column(String(500), nullable=False, comment="消息内容预览(截断500字符)") + reply_to_message_id = Column(String(128), nullable=True, comment="被回复的消息ID") + agent_config_id = Column(Integer, nullable=True, comment="处理消息的 AgentConfig ID") + status = Column(String(32), nullable=False, default="processing", comment="processing/success/error/timeout") + error_message = Column(String(500), nullable=True, comment="错误信息") + response_time_ms = Column(Integer, nullable=True, comment="响应时间(毫秒)") + reply_message_id = Column(String(128), nullable=True, comment="回复消息ID") + reply_content_preview = Column(String(500), nullable=True, comment="回复内容预览") + created_at = Column(DateTime, nullable=False, default=utc_now_naive) + replied_at = Column(DateTime, nullable=True, comment="回复时间") + extra_data = Column(JSON, nullable=True, comment="扩展元数据") + + __table_args__ = ( + Index("idx_msg_records_channel_status", "channel_id", "status"), + Index("idx_msg_records_created", "created_at"), + Index("idx_msg_records_chat", "channel_id", "chat_id"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_id": self.channel_id, + "channel_type": self.channel_type, + "message_id": self.message_id, + "chat_id": self.chat_id, + "chat_type": self.chat_type, + "content_type": self.content_type, + "sender_user_id": self.sender_user_id, + "content_preview": self.content_preview, + "reply_to_message_id": self.reply_to_message_id, + "agent_config_id": self.agent_config_id, + "status": self.status, + "error_message": self.error_message, + "response_time_ms": self.response_time_ms, + "reply_message_id": self.reply_message_id, + "reply_content_preview": self.reply_content_preview, + "created_at": format_utc_datetime(self.created_at), + "replied_at": format_utc_datetime(self.replied_at) if self.replied_at else None, + "extra_data": self.extra_data, + } + + +def _mask_secret_values(config: dict, _depth: int = 0) -> None: + if _depth > 10: + return + for key, value in config.items(): + if isinstance(value, dict): + if _SECRET_MARKER in value: + inner = dict(value[_SECRET_MARKER]) + inner["ref"] = "***" + config[key] = {_SECRET_MARKER: inner} + else: + _mask_secret_values(value, _depth + 1) + elif key.lower() in _SECRET_FIELD_NAMES and value: + config[key] = "***" + + +def _encrypt_config_sensitive_fields(config: dict, _depth: int = 0) -> dict: + if _depth > 10: + return config + from yuxi.channel.secrets.crypto import encrypt_secret + + result: dict = {} + for key, value in config.items(): + if isinstance(value, dict): + if _SECRET_MARKER in value: + result[key] = value + else: + result[key] = _encrypt_config_sensitive_fields(value, _depth + 1) + elif isinstance(value, str) and key.lower() in _SECRET_FIELD_NAMES and value: + result[key] = encrypt_secret(value) + else: + result[key] = value + return result + + +def _decrypt_config_sensitive_fields(config: dict, _depth: int = 0) -> dict: + if _depth > 10: + return config + from yuxi.channel.secrets.crypto import CryptoError, decrypt_secret, is_encrypted + + result: dict = {} + for key, value in config.items(): + if isinstance(value, dict): + if _SECRET_MARKER in value: + result[key] = value + else: + result[key] = _decrypt_config_sensitive_fields(value, _depth + 1) + elif isinstance(value, str) and is_encrypted(value): + try: + result[key] = decrypt_secret(value) + except CryptoError: + _logger.warning( + "Failed to decrypt field %s in channel config, using redacted", + key, + ) + result[key] = "***" + else: + result[key] = value + return result + + +def _install_channel_config_crypto_hooks(): + from sqlalchemy import event as sa_event + + @sa_event.listens_for(ChannelConfig, "before_insert") + @sa_event.listens_for(ChannelConfig, "before_update") + def _before_write(_mapper, _connection, target): + if target.config and isinstance(target.config, dict): + target.config = _encrypt_config_sensitive_fields(dict(target.config)) + + @sa_event.listens_for(ChannelConfig, "load") + def _after_load(target, _context): + if target.config and isinstance(target.config, dict): + target.config = _decrypt_config_sensitive_fields(dict(target.config)) + + +_install_channel_config_crypto_hooks() + + +class DeviceIdentity(Base): + __tablename__ = "channel_device_identities" + + id = Column(String(64), primary_key=True, default=lambda: uuid.uuid4().hex) + device_id = Column(String(64), unique=True, nullable=False, index=True) + name = Column(String(255), nullable=False) + public_key_pem = Column(Text, nullable=False) + status = Column(String(20), default="active") + created_by = Column(String(64), nullable=True) + last_used_at = Column(DateTime, nullable=True) + created_at = Column(DateTime, default=utc_now_naive) + updated_at = Column(DateTime, default=utc_now_naive, onupdate=utc_now_naive) + + __table_args__ = (Index("idx_channel_device_identities_status", "status"),) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "device_id": self.device_id, + "name": self.name, + "public_key_pem": self.public_key_pem, + "status": self.status, + "created_by": self.created_by, + "last_used_at": format_utc_datetime(self.last_used_at) if self.last_used_at else None, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + } diff --git a/backend/package/yuxi/utils/logging_config.py b/backend/package/yuxi/utils/logging_config.py index d76f9c99..9e5e50a1 100644 --- a/backend/package/yuxi/utils/logging_config.py +++ b/backend/package/yuxi/utils/logging_config.py @@ -12,7 +12,7 @@ LOG_FILE = f"{SAVE_DIR}/logs/yuxi-{DATETIME}.log" class LoguruHandler(logging.Handler): - """将 Python logging 桥接到 loguru 的 handler""" + """将 Python logging 桥接到 loguru 的 handler，自动注入 trace_id""" def emit(self, record: logging.LogRecord): level_map = { @@ -27,11 +27,18 @@ class LoguruHandler(logging.Handler): msg = self.format(record) except Exception: msg = record.getMessage() + trace_id = getattr(record, "trace_id", "-") or "-" + msg = f"[{trace_id}] {msg}" loguru_logger.opt(depth=1, exception=record.exc_info).log(level, msg) def _setup_logging_bridge(): - """配置 logging 到 loguru 的桥接，捕获第三方库日志（如 LightRAG）""" + """配置 logging 到 loguru 的桥接，捕获第三方库日志（如 LightRAG）， + + 同时为 yuxi.channel 模块日志安装 trace_id filter 实现全链路追踪。 + """ + from yuxi.channel.runtime.trace_ctx import TraceIdFilter + loguru_handler = LoguruHandler() loguru_handler.setLevel(logging.DEBUG) @@ -51,6 +58,14 @@ def _setup_logging_bridge(): lib_logger.setLevel(logging.WARNING) lib_logger.propagate = False + # 桥接 yuxi.channel 模块日志，并安装 trace_id filter + channel_logger = logging.getLogger("yuxi.channel") + channel_logger.addHandler(loguru_handler) + channel_logger.setLevel(logging.DEBUG) + channel_logger.propagate = False + trace_filter = TraceIdFilter() + channel_logger.addFilter(trace_filter) + def setup_logger(name, level="DEBUG", console=True): """使用 loguru 设置日志记录器"""