From e2004575199f34466ff35cb4e08b32cbcd14c63a Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Fri, 22 May 2026 11:18:23 +0800 Subject: [PATCH] =?UTF-8?q?refactor(approval):=20=E5=B0=86ApprovalEngine?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=E5=88=B0=E7=8B=AC=E7=AB=8B=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 删除原approval通道下的engine.py文件，将ApprovalEngine导入路径调整到channel_approval_service服务模块，统一审批引擎的代码组织方式 --- .../package/yuxi/channel/approval/__init__.py | 2 +- .../package/yuxi/channel/approval/engine.py | 272 ------------------ 2 files changed, 1 insertion(+), 273 deletions(-) delete mode 100644 backend/package/yuxi/channel/approval/engine.py diff --git a/backend/package/yuxi/channel/approval/__init__.py b/backend/package/yuxi/channel/approval/__init__.py index d3ae27cc..2b4c8aaa 100644 --- a/backend/package/yuxi/channel/approval/__init__.py +++ b/backend/package/yuxi/channel/approval/__init__.py @@ -1,5 +1,5 @@ from yuxi.channel.approval.delivery import DeliveryResult, deliver_approval_request -from yuxi.channel.approval.engine import ApprovalEngine +from yuxi.services.channel_approval_service import ApprovalEngine from yuxi.channel.approval.native_runtime import ( NativeApprovalCard, NativeApprovalRenderer, diff --git a/backend/package/yuxi/channel/approval/engine.py b/backend/package/yuxi/channel/approval/engine.py deleted file mode 100644 index 5cd7376d..00000000 --- a/backend/package/yuxi/channel/approval/engine.py +++ /dev/null @@ -1,272 +0,0 @@ -import asyncio -import logging -import time - -from yuxi.channel.protocols import ( - ApprovalAction, - ApprovalDecision, - ApprovalProtocol, - ApprovalRequest, - ApprovalResult, -) -from yuxi.channel.plugins.registry import ChannelPluginRegistry - -logger = logging.getLogger(__name__) - -_TERMINAL_DECISIONS = frozenset({ - ApprovalDecision.APPROVED, - ApprovalDecision.DENIED, - ApprovalDecision.CANCELLED, - ApprovalDecision.EXPIRED, -}) - - -class ApprovalEngine: - def __init__(self): - self._pending: dict[str, ApprovalRequest] = {} - self._results: dict[str, ApprovalResult] = {} - self._events: dict[str, asyncio.Event] = {} - self._cleanup_task: asyncio.Task | None = None - - @property - def pending_count(self) -> int: - return len(self._pending) - - @property - def results_count(self) -> int: - return len(self._results) - - def get_pending_request(self, request_id: str) -> ApprovalRequest | None: - return self._pending.get(request_id) - - def get_result(self, request_id: str) -> ApprovalResult | None: - return self._results.get(request_id) - - def _is_terminal(self, request_id: str) -> bool: - existing = self._results.get(request_id) - return existing is not None and existing.decision in _TERMINAL_DECISIONS - - def cancel_request(self, request_id: str, reason: str = "已取消") -> bool: - existing = self._results.get(request_id) - if existing is not None: - logger.warning( - "审批请求已达终态，无法取消: req=%s, current=%s", - request_id, - existing.decision.value, - ) - return False - if request_id not in self._pending: - return False - del self._pending[request_id] - result = ApprovalResult( - request_id=request_id, - decision=ApprovalDecision.CANCELLED, - reason=reason, - ) - self._results[request_id] = result - self._signal(request_id) - logger.info("审批请求已取消: req=%s, reason=%s", request_id, reason) - return True - - def register_decision( - self, - request_id: str, - decision: ApprovalDecision, - decided_by: str = "external", - reason: str | None = None, - ) -> bool: - existing = self._results.get(request_id) - if existing is not None: - logger.warning( - "审批请求已达终态，无法重新决策: req=%s, current=%s, attempted=%s", - request_id, - existing.decision.value, - decision.value, - ) - return False - request = self._pending.get(request_id) - if request is None: - return False - if not request.approver_ids: - logger.warning( - "审批身份校验失败: req=%s, 审批人列表为空", - request_id, - ) - return False - if decided_by not in request.approver_ids: - logger.warning( - "审批身份校验失败: req=%s, decided_by=%s 不在审批人列表 %s 中", - request_id, - decided_by, - request.approver_ids, - ) - return False - del self._pending[request_id] - result = ApprovalResult( - request_id=request_id, - decision=decision, - decided_by=decided_by, - reason=reason, - ) - self._results[request_id] = result - self._signal(request_id) - logger.info( - "审批决策已注册: req=%s, decision=%s, decided_by=%s, reason=%s", - request_id, - decision.value, - decided_by, - reason, - ) - return True - - def cleanup_expired(self, max_age_seconds: float = 300) -> int: - removed = 0 - now = time.monotonic() - stale = [ - rid for rid, req in self._pending.items() - if now - req.created_at > max_age_seconds and rid not in self._results - ] - for rid in stale: - del self._pending[rid] - result = ApprovalResult( - request_id=rid, - decision=ApprovalDecision.EXPIRED, - reason="清理过期审批", - ) - self._results[rid] = result - self._signal(rid) - removed += 1 - if removed > 0: - logger.info("清理过期审批请求: count=%d", removed) - return removed - - def cleanup_results(self, max_age_seconds: float = 600) -> int: - removed = 0 - now = time.monotonic() - stale = [rid for rid, res in self._results.items() if now - res.decided_at > max_age_seconds] - for rid in stale: - del self._results[rid] - removed += 1 - if removed > 0: - logger.info("清理过期审批结果: count=%d", removed) - return removed - - def start_cleanup_task(self, interval: float = 60) -> None: - if self._cleanup_task is not None and not self._cleanup_task.done(): - return - self._cleanup_task = asyncio.create_task(self._run_cleanup_loop(interval)) - - def stop_cleanup_task(self) -> None: - if self._cleanup_task is not None and not self._cleanup_task.done(): - self._cleanup_task.cancel() - - async def _run_cleanup_loop(self, interval: float) -> None: - try: - while True: - await asyncio.sleep(interval) - self.cleanup_expired() - self.cleanup_results() - except asyncio.CancelledError: - pass - - def _signal(self, request_id: str) -> None: - event = self._events.pop(request_id, None) - if event is not None: - event.set() - - async def request_approval( - self, - channel_type: str, - account_id: str, - config: dict, - action: ApprovalAction, - initiator_peer_id: str, - description: str, - context: dict, - ) -> ApprovalRequest | None: - plugin = ChannelPluginRegistry.get(channel_type) - if plugin is None or not isinstance(plugin, ApprovalProtocol): - return None - - need = await plugin.check_approval_required(config, action, initiator_peer_id) - if not need: - return None - - request = await plugin.create_approval_request( - config, - action, - initiator_peer_id, - description, - context, - ) - approvers = plugin.get_approver_ids(config) - if not approvers: - logger.warning( - "审批请求创建失败: channel=%s, 审批人列表为空", - channel_type, - ) - return None - request.approver_ids = list(approvers) - sent = await plugin.send_approval_notification(config, request, approvers) - if not sent: - logger.warning( - "审批通知发送失败: req=%s, channel=%s, approvers=%s", - request.id, - channel_type, - approvers, - ) - - self._pending[request.id] = request - return request - - async def wait_for_decision( - self, channel_type: str, config: dict, request_id: str, timeout: float = 300 - ) -> ApprovalResult: - existing = self._results.get(request_id) - if existing is not None: - return existing - - event = self._events.get(request_id) - if event is None: - event = asyncio.Event() - self._events[request_id] = event - - loop = asyncio.get_running_loop() - deadline = loop.time() + timeout - while True: - remaining = deadline - loop.time() - if remaining <= 0: - break - - plugin_result = await self._check_and_update(channel_type, config, request_id) - if plugin_result is not None: - self._pending.pop(request_id, None) - self._results[request_id] = plugin_result - self._events.pop(request_id, None) - return plugin_result - - try: - await asyncio.wait_for(event.wait(), timeout=min(remaining, 5)) - except asyncio.TimeoutError: - continue - - stored = self._results.get(request_id) - if stored is not None: - self._events.pop(request_id, None) - return stored - - self._events.pop(request_id, None) - expired = ApprovalResult( - request_id=request_id, - decision=ApprovalDecision.EXPIRED, - reason="审批超时", - ) - self._pending.pop(request_id, None) - self._results[request_id] = expired - return expired - - async def _check_and_update(self, channel_type: str, config: dict, request_id: str) -> ApprovalResult | None: - plugin = ChannelPluginRegistry.get(channel_type) - if plugin is None or not isinstance(plugin, ApprovalProtocol): - return None - return await plugin.check_approval_status(config, request_id)