From 6ca611fead8690a4a917eda193e1c1652b2a71a0 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 14 May 2026 09:24:50 +0800 Subject: [PATCH] =?UTF-8?q?refactor(channel):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E5=B9=B6=E6=96=B0=E5=A2=9E=E5=A4=9A=E9=A1=B9=E6=B8=A0=E9=81=93?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 简化message_actions.py中获取适配器的逻辑 2. 新增适配器合法性校验工具方法 3. 新增会话映射过期清理功能 4. 重构渠道状态机与基础适配器实现 5. 统一渠道操作异常处理逻辑 6. 新增凭证状态查询与刷新接口 7. 优化健康检查与自动重连逻辑 8. 新增统计数据缓存与批量查询优化 9. 修复部分数据库操作的异常处理逻辑 --- backend/package/yuxi/channels/base.py | 31 ++- backend/package/yuxi/channels/manager.py | 205 +++++++++++++----- .../package/yuxi/channels/message_actions.py | 2 - backend/package/yuxi/channels/registry.py | 35 +++ .../package/yuxi/channels/session_mapper.py | 28 ++- backend/server/routers/channels_router.py | 146 +++++++++---- 6 files changed, 337 insertions(+), 110 deletions(-) diff --git a/backend/package/yuxi/channels/base.py b/backend/package/yuxi/channels/base.py index e4993a09..41a8982c 100644 --- a/backend/package/yuxi/channels/base.py +++ b/backend/package/yuxi/channels/base.py @@ -9,6 +9,7 @@ from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelMessage, ChannelResponse, + ChannelStatus, ChannelType, DeliveryResult, HealthStatus, @@ -39,6 +40,23 @@ class BaseChannelAdapter(ChannelLifecycleProtocol, ChannelGatewayProtocol, ABC): config_schema: ClassVar[dict[str, dict[str, Any]]] = {} + VALID_TRANSITIONS: ClassVar[dict[str, set[str]]] = { + ChannelStatus.DISCONNECTED.value: {ChannelStatus.CONNECTING.value}, + ChannelStatus.CONNECTING.value: { + ChannelStatus.CONNECTED.value, + ChannelStatus.ERROR.value, + ChannelStatus.DISCONNECTED.value, + }, + ChannelStatus.CONNECTED.value: {"disconnecting", ChannelStatus.RECONNECTING.value, ChannelStatus.ERROR.value}, + "disconnecting": {ChannelStatus.DISCONNECTED.value}, + ChannelStatus.RECONNECTING.value: { + ChannelStatus.CONNECTED.value, + ChannelStatus.ERROR.value, + ChannelStatus.DISCONNECTED.value, + }, + ChannelStatus.ERROR.value: {ChannelStatus.CONNECTING.value, ChannelStatus.DISCONNECTED.value}, + } + _state_store: PluginStateStore | None = None def __init__(self, config: dict[str, Any] | None = None): @@ -147,7 +165,7 @@ class BaseChannelAdapter(ChannelLifecycleProtocol, ChannelGatewayProtocol, ABC): async def send(self, response: ChannelResponse) -> DeliveryResult: ... async def receive(self) -> AsyncIterator[ChannelMessage]: - raise NotImplementedError + return yield # type: ignore[misc] @abstractmethod @@ -303,6 +321,17 @@ class BaseChannelAdapter(ChannelLifecycleProtocol, ChannelGatewayProtocol, ABC): def is_configured(self) -> bool: return bool(self.config) + def _transition(self, new_status: str) -> None: + current = self.status + valid_targets = self.VALID_TRANSITIONS.get(current) + if valid_targets is None: + raise ValueError(f"Unknown current status: {current}") + if new_status not in valid_targets: + raise ValueError( + f"Illegal state transition: {current} -> {new_status}. " + f"Allowed transitions from {current}: {valid_targets}" + ) + @property def status(self) -> str: return getattr(self, "_status", "unknown") or "unknown" diff --git a/backend/package/yuxi/channels/manager.py b/backend/package/yuxi/channels/manager.py index 02ef4ed4..088df13e 100644 --- a/backend/package/yuxi/channels/manager.py +++ b/backend/package/yuxi/channels/manager.py @@ -19,6 +19,7 @@ from yuxi.channels.infra.config_watcher import ConfigWatcher from yuxi.channels.models import ChannelStatus from yuxi.channels.registry import _BUILTIN_ADAPTERS, ChannelRegistry, _load_builtin_adapters from yuxi.channels.router import MessageRouter +from yuxi.channels.session_mapper import SessionMapper from yuxi.channels.services.context import GatewayRequestContext from yuxi.channels.services.doctor import ConfigDoctor, DiagnosisIssue from yuxi.channels.services.maintenance import MaintenanceRunner @@ -35,6 +36,7 @@ from yuxi.utils.logging_config import logger HEALTH_CHECK_INTERVAL = 60 HEALTH_CHECK_MAX_AGE = 65 CONFIG_WATCH_INTERVAL = 30.0 +STATS_CACHE_TTL = 60.0 _DEFAULT_RATE_LIMIT = { "enabled": True, @@ -125,6 +127,10 @@ class ChannelManager: self._prev_statuses: dict[str, str] = {} self._config_lock: asyncio.Lock = asyncio.Lock() self._cached_health: dict[str, tuple[float, Any]] = {} + self._stats_cache: dict[str, tuple[float, dict]] = {} + self._batch_stats_cache: tuple[float, dict[str, dict]] = (0, {}) + self._start_stop_locks: dict[str, asyncio.Lock] = {} + self._health_failure_counts: dict[str, int] = defaultdict(int) @property def phase(self) -> str: @@ -484,6 +490,11 @@ class ChannelManager: return await self._watcher.reload_now(changed_keys) async def start_channel(self, channel_id: str, config: dict[str, Any] | None = None) -> None: + lock = self._start_stop_locks.setdefault(channel_id, asyncio.Lock()) + async with lock: + return await self._start_channel_impl(channel_id, config) + + async def _start_channel_impl(self, channel_id: str, config: dict[str, Any] | None = None) -> None: if channel_id in self._adapters: logger.warning(f"Channel {channel_id} already running") return @@ -493,6 +504,8 @@ class ChannelManager: raise ValueError(f"No adapter registered for channel {channel_id}") config = config or self._channels_config.get(channel_id, {}) + _validate_channel_config(channel_id, config) + adapter = adapter_cls(config=config) adapter._state_store = self._state_store adapter.on_message(self._handle_inbound_message) @@ -532,6 +545,11 @@ class ChannelManager: logger.info(f"Channel {channel_id} started") async def stop_channel(self, channel_id: str) -> None: + lock = self._start_stop_locks.setdefault(channel_id, asyncio.Lock()) + async with lock: + return await self._stop_channel_impl(channel_id) + + async def _stop_channel_impl(self, channel_id: str) -> None: adapter = self._adapters.get(channel_id) if not adapter: return @@ -551,6 +569,7 @@ class ChannelManager: finally: self._adapters.pop(channel_id, None) self._prev_statuses.pop(channel_id, None) + self._health_failure_counts.pop(channel_id, None) if channel_id in self._circuit_breakers: self._circuit_breakers[channel_id] = CircuitBreaker(channel_id=channel_id) @@ -733,8 +752,6 @@ class ChannelManager: "status": "error", "total_messages": 0, "today_messages": 0, - "total_count": 0, - "today_count": 0, "active_connections": 0, "health": None, } @@ -749,8 +766,6 @@ class ChannelManager: "status": info.get("status", "not_found"), "total_messages": stats.get("total_messages", 0), "today_messages": stats.get("today_messages", 0), - "total_count": stats.get("total_messages", 0), - "today_count": stats.get("today_messages", 0), "active_connections": 1 if info.get("status") == ChannelStatus.CONNECTED.value else 0, "health": info.get("health"), } @@ -771,38 +786,34 @@ class ChannelManager: _validate_channel_config(channel_id, config_updates) adapter = self._adapters.get(channel_id) - old_config: dict[str, Any] | None = dict(adapter.config) if adapter else None - try: - if adapter: - await adapter.reload_config(config_updates) - - async with pg_manager.get_async_session_context() as db: - result = await db.execute(select(ChannelConfig).where(ChannelConfig.channel_id == channel_id)) - db_config = result.scalar_one_or_none() - if db_config: - existing = db_config.config_json or {} - db_config.config_json = {**existing, **config_updates} - db_config.updated_at = _utc_now() - else: - db_config = ChannelConfig( - channel_id=channel_id, - config_json=config_updates, - enabled=True, - ) - db.add(db_config) - await db.commit() - - except Exception: - if adapter and old_config is not None: - await adapter.reload_config(old_config) - raise + async with pg_manager.get_async_session_context() as db: + result = await db.execute(select(ChannelConfig).where(ChannelConfig.channel_id == channel_id)) + db_config = result.scalar_one_or_none() + if db_config: + existing = db_config.config_json or {} + db_config.config_json = {**existing, **config_updates} + db_config.updated_at = _utc_now() + else: + db_config = ChannelConfig( + channel_id=channel_id, + config_json=config_updates, + enabled=True, + ) + db.add(db_config) + await db.commit() async with self._config_lock: if channel_id not in self._channels_config: self._channels_config[channel_id] = {} self._channels_config[channel_id].update(config_updates) + if adapter: + try: + await adapter.reload_config(config_updates) + except Exception: + logger.exception(f"Failed to reload config for running adapter {channel_id}") + needs_restart = adapter is None logger.info( @@ -916,6 +927,9 @@ class ChannelManager: return {"channel_id": channel_id, "test_result": "failure", "error": str(e)} def is_registered(self, channel_id: str) -> bool: + return channel_id in self._channels_config + + def is_enabled(self, channel_id: str) -> bool: if channel_id not in self._channels_config: return False return self._channels_config[channel_id].get("enabled", True) @@ -939,6 +953,16 @@ class ChannelManager: except Exception: logger.warning("Failed to cleanup expired state entries", exc_info=True) + try: + async with pg_manager.get_async_session_context() as db: + removed = await SessionMapper._cleanup_expired_mappings(db) + if removed > 0: + logger.info(f"[Cleanup] Removed {removed} expired thread mappings") + except asyncio.CancelledError: + break + except Exception: + logger.warning("Failed to cleanup expired thread mappings", exc_info=True) + async def check_rate_limit(self, key: str, max_req: int, window_seconds: int) -> bool: lock = self._rate_limit_locks.setdefault(key, asyncio.Lock()) async with lock: @@ -984,6 +1008,7 @@ class ChannelManager: logger.exception("Error handling inbound message") async def _health_check_loop(self, channel_id: str) -> None: + HEALTH_FAILURE_THRESHOLD = 3 while channel_id in self._adapters: await asyncio.sleep(HEALTH_CHECK_INTERVAL) adapter = self._adapters.get(channel_id) @@ -998,6 +1023,7 @@ class ChannelManager: try: health = await adapter.health_check() self._cached_health[channel_id] = (time.monotonic(), health) + self._health_failure_counts[channel_id] = 0 if cb and health.status == "healthy": try: await cb.record_success() @@ -1017,6 +1043,18 @@ class ChannelManager: await cb.record_failure() except Exception: pass + self._health_failure_counts[channel_id] += 1 + if self._health_failure_counts[channel_id] >= HEALTH_FAILURE_THRESHOLD: + logger.warning( + f"Health check failed {self._health_failure_counts[channel_id]} consecutive times " + f"for {channel_id}, triggering auto-reconnect" + ) + self._health_failure_counts[channel_id] = 0 + try: + await self.restart_channel(channel_id) + except Exception as reconnect_error: + logger.error(f"Auto-reconnect failed for {channel_id}: {reconnect_error}") + continue if current_status != prev_status and self._broadcaster: self._prev_statuses[channel_id] = current_status @@ -1034,11 +1072,16 @@ class ChannelManager: if not channel_ids: return {} + now = time.monotonic() + cache_ts, cached = self._batch_stats_cache + if now - cache_ts < STATS_CACHE_TTL and cached: + return {cid: cached.get(cid, self._empty_stats()) for cid in channel_ids} + try: from yuxi.utils.datetime_utils import utc_now_naive - now = utc_now_naive() - today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) + utc_now = utc_now_naive() + today_start = utc_now.replace(hour=0, minute=0, second=0, microsecond=0) async with pg_manager.get_async_session_context() as session: totals_result = await session.execute( @@ -1084,26 +1127,13 @@ class ChannelManager: "success_rate": round(success / total, 3) if total > 0 else 0, } else: - stats_map[cid] = { - "total_messages": 0, - "today_messages": 0, - "success_count": 0, - "error_count": 0, - "success_rate": 0, - } - return stats_map + stats_map[cid] = self._empty_stats() + + self._batch_stats_cache = (now, stats_map) + return {cid: stats_map.get(cid, self._empty_stats()) for cid in channel_ids} except Exception: logger.warning(f"批量查询渠道统计失败 (channel_ids 数量: {len(channel_ids)}):", exc_info=True) - return { - cid: { - "total_messages": 0, - "today_messages": 0, - "success_count": 0, - "error_count": 0, - "success_rate": 0, - } - for cid in channel_ids - } + return {cid: self._empty_stats() for cid in channel_ids} async def _get_single_channel_status(self, channel_id: str, stats: dict | None = None) -> dict: adapter = self._adapters.get(channel_id) @@ -1193,11 +1223,16 @@ class ChannelManager: } async def _get_channel_stats(self, channel_id: str) -> dict: + now = time.monotonic() + cached = self._stats_cache.get(channel_id) + if cached and now - cached[0] < STATS_CACHE_TTL: + return cached[1] + try: from yuxi.utils.datetime_utils import utc_now_naive - now = utc_now_naive() - today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) + utc_now = utc_now_naive() + today_start = utc_now.replace(hour=0, minute=0, second=0, microsecond=0) async with pg_manager.get_async_session_context() as session: result = await session.execute( @@ -1224,21 +1259,27 @@ class ChannelManager: ) today = today_result.scalar() or 0 - return { + stats = { "total_messages": total, "today_messages": today, "success_count": int(success), "error_count": int(error), "success_rate": round(success / total, 3) if total > 0 else 0, } + self._stats_cache[channel_id] = (now, stats) + return stats except Exception: - return { - "total_messages": 0, - "today_messages": 0, - "success_count": 0, - "error_count": 0, - "success_rate": 0, - } + return self._empty_stats() + + @staticmethod + def _empty_stats() -> dict: + return { + "total_messages": 0, + "today_messages": 0, + "success_count": 0, + "error_count": 0, + "success_rate": 0, + } def _adapter_status(self, adapter: BaseChannelAdapter) -> str: _status = getattr(adapter, "_status", None) @@ -1246,6 +1287,52 @@ class ChannelManager: return "unknown" return _status.value if hasattr(_status, "value") else str(_status) + async def _get_credential_summary(self, channel_id: str, adapter: BaseChannelAdapter) -> dict: + try: + secret_manager = getattr(adapter, "_secret_manager", None) + if secret_manager and hasattr(secret_manager, "get_credential_status"): + return await secret_manager.get_credential_status() + except Exception: + logger.warning(f"获取渠道 {channel_id} 凭证状态失败", exc_info=True) + + config = self._channels_config.get(channel_id, {}) + has_credential = any( + any(pattern in k.lower() for pattern in self.SENSITIVE_KEY_PATTERNS) and v for k, v in config.items() + ) + return { + "has_credential": has_credential, + "is_expired": False, + "source": "config" if has_credential else "none", + } + + async def get_credential_status(self, channel_id: str) -> dict: + adapter = self._adapters.get(channel_id) + if adapter: + return await self._get_credential_summary(channel_id, adapter) + + config = self._channels_config.get(channel_id, {}) + has_credential = any( + any(pattern in k.lower() for pattern in self.SENSITIVE_KEY_PATTERNS) and v for k, v in config.items() + ) + return { + "has_credential": has_credential, + "is_expired": False, + "source": "config" if has_credential else "none", + } + + async def refresh_credential(self, channel_id: str) -> dict: + adapter = self._adapters.get(channel_id) + if adapter and hasattr(adapter, "refresh_credential"): + result = await adapter.refresh_credential() + return result + + if adapter: + logger.info(f"渠道 {channel_id} 适配器不支持凭证刷新，尝试重连") + await self.restart_channel(channel_id) + return {"refreshed": True, "method": "restart", "message": "通过重启渠道刷新凭证"} + + raise ValueError(f"Channel '{channel_id}' is not running, cannot refresh credential") + async def _ensure_virtual_department(self, db) -> None: from yuxi.storage.postgres.models_business import Department diff --git a/backend/package/yuxi/channels/message_actions.py b/backend/package/yuxi/channels/message_actions.py index 84175f3f..e6db081a 100644 --- a/backend/package/yuxi/channels/message_actions.py +++ b/backend/package/yuxi/channels/message_actions.py @@ -309,6 +309,4 @@ class ActionRouter: for adapter in self._manager._adapters.values(): if hasattr(adapter, "_active_chats") and chat_id in adapter._active_chats: return adapter - for adapter in self._manager._adapters.values(): - return adapter return None diff --git a/backend/package/yuxi/channels/registry.py b/backend/package/yuxi/channels/registry.py index 8822f75f..f24b10bf 100644 --- a/backend/package/yuxi/channels/registry.py +++ b/backend/package/yuxi/channels/registry.py @@ -1,5 +1,7 @@ from __future__ import annotations +import inspect + from yuxi.channels.base import BaseChannelAdapter _BUILTIN_ADAPTERS: dict[str, type[BaseChannelAdapter]] = {} @@ -217,6 +219,39 @@ class ChannelRegistry: def is_available(self, channel_id: str) -> bool: return channel_id in self._adapters or channel_id in _BUILTIN_ADAPTERS + @staticmethod + def _validate_adapter(adapter_cls: type[BaseChannelAdapter]) -> list[str]: + CAPABILITY_METHOD_MAP: dict[str, str] = { + "reactions": "send_reaction", + "edit": "edit_message", + "unsend": "delete_message", + "media": "send_media", + } + + caps = getattr(adapter_cls, "capabilities", None) + if caps is None: + return [] + + issues: list[str] = [] + for cap_flag, method_name in CAPABILITY_METHOD_MAP.items(): + if not getattr(caps, cap_flag, False): + continue + method = getattr(adapter_cls, method_name, None) + if method is None: + issues.append(f"Capability '{cap_flag}' declared but method '{method_name}' not found") + continue + try: + source = inspect.getsource(method) + except (OSError, TypeError): + continue + if "raise NotImplementedError" in source or "raise NotImplementedError()" in source: + issues.append( + f"Capability '{cap_flag}' declared but method '{method_name}' is not implemented " + f"(raises NotImplementedError)" + ) + + return issues + def load_builtins(self) -> None: """将 _BUILTIN_ADAPTERS 中的所有内置适配器加载到 self._adapters 中""" _load_builtin_adapters() diff --git a/backend/package/yuxi/channels/session_mapper.py b/backend/package/yuxi/channels/session_mapper.py index a7e28123..4f1cc4f1 100644 --- a/backend/package/yuxi/channels/session_mapper.py +++ b/backend/package/yuxi/channels/session_mapper.py @@ -1,8 +1,9 @@ from __future__ import annotations import uuid as uuid_lib +from datetime import timedelta -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -16,6 +17,9 @@ from yuxi.utils.logging_config import logger VIRTUAL_DEPARTMENT_ID = -1 USER_SOURCE_PREFIX = "channel:" +THREAD_MAPPING_CLEANUP_TTL_DAYS = 7 +THREAD_MAPPING_CLEANUP_LONG_TTL_DAYS = 30 + class SessionMapper: def __init__(self, db: AsyncSession, department_id: int = VIRTUAL_DEPARTMENT_ID): @@ -139,3 +143,25 @@ class SessionMapper: ) ) return result.scalar_one_or_none() + + @staticmethod + async def cleanup_expired(db: AsyncSession, ttl_days: int = THREAD_MAPPING_CLEANUP_TTL_DAYS) -> int: + cutoff = utc_now_naive() - timedelta(days=ttl_days) + result = await db.execute(delete(ChannelThreadMapping).where(ChannelThreadMapping.last_active_at < cutoff)) + await db.commit() + removed = result.rowcount + if removed: + logger.info(f"SessionMapper: cleaned up {removed} expired thread mappings (ttl={ttl_days}d)") + return removed + + @staticmethod + async def _cleanup_expired_mappings(db: AsyncSession) -> int: + cutoff = utc_now_naive() - timedelta(days=THREAD_MAPPING_CLEANUP_LONG_TTL_DAYS) + result = await db.execute(delete(ChannelThreadMapping).where(ChannelThreadMapping.last_active_at < cutoff)) + await db.commit() + removed = result.rowcount + if removed: + logger.info( + f"SessionMapper: cleaned up {removed} expired mappings (ttl={THREAD_MAPPING_CLEANUP_LONG_TTL_DAYS}d)" + ) + return removed diff --git a/backend/server/routers/channels_router.py b/backend/server/routers/channels_router.py index f675d085..61e62607 100644 --- a/backend/server/routers/channels_router.py +++ b/backend/server/routers/channels_router.py @@ -58,6 +58,24 @@ async def _check_rate_limit(user_id: str, key: str, max_req: int, window: int) - ) +async def _handle_channel_operation_exceptions(channel_id: str, operation): + try: + return await operation + except ChannelAuthenticationError as e: + raise HTTPException(status_code=401, detail=f"凭证无效: {e}") + except ChannelNotConnectedError as e: + raise HTTPException(status_code=503, detail=f"渠道不可达: {e}") + except ChannelTimeoutError as e: + raise HTTPException(status_code=504, detail=f"连接超时: {e}") + except ChannelRateLimitError as e: + raise HTTPException(status_code=429, detail=f"上游限流: {e}") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Failed to execute channel operation for {channel_id}: {e}") + raise HTTPException(status_code=500, detail=f"Failed to execute channel operation: {e}") + + _POLICY_FIELDS = [ "group_chat_mode", "whitelist_ids", @@ -468,21 +486,7 @@ async def start_channel( if manager.is_running(channel_id): return _make_response(code=409, message=f"Channel '{channel_id}' is already running") - try: - await manager.start_channel(channel_id) - except ChannelAuthenticationError as e: - raise HTTPException(status_code=401, detail=f"凭证无效: {e}") - except ChannelNotConnectedError as e: - raise HTTPException(status_code=503, detail=f"渠道不可达: {e}") - except ChannelTimeoutError as e: - raise HTTPException(status_code=504, detail=f"连接超时: {e}") - except ChannelRateLimitError as e: - raise HTTPException(status_code=429, detail=f"上游限流: {e}") - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - except Exception as e: - logger.error(f"Failed to start channel {channel_id}: {e}") - raise HTTPException(status_code=500, detail=f"Failed to start channel: {e}") + await _handle_channel_operation_exceptions(channel_id, manager.start_channel(channel_id)) adapter = manager._adapters.get(channel_id) current_status = manager._adapter_status(adapter) if adapter else "connecting" @@ -539,21 +543,7 @@ async def restart_channel( if not manager.is_running(channel_id): raise HTTPException(status_code=409, detail=f"Channel '{channel_id}' is not running") - try: - await manager.restart_channel(channel_id) - except ChannelAuthenticationError as e: - raise HTTPException(status_code=401, detail=f"凭证无效: {e}") - except ChannelNotConnectedError as e: - raise HTTPException(status_code=503, detail=f"渠道不可达: {e}") - except ChannelTimeoutError as e: - raise HTTPException(status_code=504, detail=f"连接超时: {e}") - except ChannelRateLimitError as e: - raise HTTPException(status_code=429, detail=f"上游限流: {e}") - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - except Exception as e: - logger.error(f"Failed to restart channel {channel_id}: {e}") - raise HTTPException(status_code=500, detail=f"Failed to restart channel: {e}") + await _handle_channel_operation_exceptions(channel_id, manager.restart_channel(channel_id)) adapter = manager._adapters.get(channel_id) current_status = manager._adapter_status(adapter) if adapter else "connecting" @@ -980,20 +970,31 @@ async def update_channel_policy( updates_dict = policy_updates.model_dump(exclude_none=True) - result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) - policy = result.scalar_one_or_none() + try: + result = await db.execute(select(ChannelPolicyConfig).where(ChannelPolicyConfig.channel_id == channel_id)) + policy = result.scalar_one_or_none() - if policy: - for key in _POLICY_FIELDS: - if key in updates_dict: - setattr(policy, key, updates_dict[key]) - else: - policy_data = {k: v for k, v in updates_dict.items() if k in _POLICY_FIELDS} - policy = ChannelPolicyConfig(channel_id=channel_id, **policy_data) - db.add(policy) + if policy: + for key in _POLICY_FIELDS: + if key in updates_dict: + setattr(policy, key, updates_dict[key]) + else: + policy_data = {k: v for k, v in updates_dict.items() if k in _POLICY_FIELDS} + policy = ChannelPolicyConfig(channel_id=channel_id, **policy_data) + db.add(policy) + + await db.commit() + await db.refresh(policy) + except IntegrityError: + await db.rollback() + raise HTTPException(status_code=409, detail="策略配置写入冲突，请重试") + except DBAPIError as e: + logger.error(f"数据库操作失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="数据库操作失败，请稍后重试") + except Exception as e: + logger.error(f"更新策略配置失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="更新策略配置失败") - await db.commit() - await db.refresh(policy) return _make_response(data=policy.to_dict(), message="策略配置已保存") @@ -1009,8 +1010,16 @@ async def get_channel_routing( from yuxi.storage.postgres.models_channels import ChannelRoutingRule - result = await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id)) - rules = result.scalars().all() + try: + result = await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id)) + rules = result.scalars().all() + except DBAPIError as e: + logger.error(f"查询路由规则失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="数据库操作失败，请稍后重试") + except Exception as e: + logger.error(f"查询路由规则失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="查询路由规则失败") + return _make_response(data={"channel_id": channel_id, "routes": [r.to_dict() for r in rules]}) @@ -1059,10 +1068,27 @@ async def update_channel_routing( ) db.add(rule) - await db.commit() + try: + await db.commit() + except IntegrityError: + await db.rollback() + raise HTTPException(status_code=409, detail="路由规则写入冲突，请重试") + except DBAPIError as e: + logger.error(f"数据库操作失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="数据库操作失败，请稍后重试") + except Exception as e: + logger.error(f"保存路由规则失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="保存路由规则失败") - result = await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id)) - saved_rules = result.scalars().all() + try: + result = await db.execute(select(ChannelRoutingRule).where(ChannelRoutingRule.channel_id == channel_id)) + saved_rules = result.scalars().all() + except DBAPIError as e: + logger.error(f"查询路由规则失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="数据库操作失败，请稍后重试") + except Exception as e: + logger.error(f"查询路由规则失败 (channel={channel_id}): {e}", exc_info=True) + raise HTTPException(status_code=500, detail="查询路由规则失败") return _make_response( data={"channel_id": channel_id, "routes": [r.to_dict() for r in saved_rules]}, message="路由规则已保存", @@ -1320,6 +1346,32 @@ async def delete_channel_state_entry( ) +@channels.get("/{channel_id}/credential-status") +async def get_channel_credential_status( + channel_id: str, + current_user: User = Depends(get_admin_user), +): + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") + + credential = await manager.get_credential_status(channel_id) + return _make_response(data={"channel_id": channel_id, "credential": credential}) + + +@channels.post("/{channel_id}/refresh-credential") +async def refresh_channel_credential( + channel_id: str, + current_user: User = Depends(get_admin_user), +): + manager = get_channel_manager() + if not manager.is_registered(channel_id): + raise HTTPException(status_code=404, detail=f"Channel '{channel_id}' is not registered") + + result = await manager.refresh_credential(channel_id) + return _make_response(data={"channel_id": channel_id, "credential": result}) + + @channels.delete("/{channel_id}") async def unregister_channel( channel_id: str,