diff --git a/app/application/chain/context.py b/app/application/chain/context.py index 83fe8b313..46ebc8f98 100644 --- a/app/application/chain/context.py +++ b/app/application/chain/context.py @@ -35,6 +35,7 @@ if TYPE_CHECKING: SubscriptionMutationScope, SyncSubscriptionMutationScope, ) + from app.application.system import SystemService from app.application.transfer.execution import TransferExecutionRepository from app.application.transfer.workflow import TransferAdmissionRepository @@ -84,6 +85,7 @@ class ChainRuntimeContext: default_factory=lambda: ChainRuntimeConfig(media_extensions=()) ) stop_state: StopState = field(default_factory=lambda: runtime_stop_state) + system_service: Optional[SystemService] = None def _unconfigured_chain_runtime_context() -> ChainRuntimeContext: diff --git a/app/application/messaging/interaction.py b/app/application/messaging/interaction.py index f0bdb2674..be4288e95 100644 --- a/app/application/messaging/interaction.py +++ b/app/application/messaging/interaction.py @@ -34,7 +34,7 @@ class SlashInteractionManager: _ttl = timedelta(hours=24) - def __init__(self): + def __init__(self) -> None: """初始化按请求和用户索引的 slash 会话表。""" self._by_id: Dict[str, PendingSlashInteraction] = {} self._by_user: Dict[str, str] = {} @@ -140,9 +140,23 @@ class InteractionDispatch: class MessageGateway(Protocol): """声明交互控制器使用的消息发送和编辑能力。""" - def post_message(self, message: Message): ... + def post_message(self, message: Message) -> None: + """发送一条交互消息。""" + ... - def edit_message(self, **kwargs) -> bool: ... + def edit_message( + self, + channel: NotificationChannel, + source: str, + message_id: Union[str, int], + chat_id: Union[str, int], + text: str, + title: Optional[str] = None, + buttons: Optional[List[List[Dict[str, Any]]]] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> bool: + """编辑一条已发送的交互消息。""" + ... def supports_interaction_buttons(channel: Optional[NotificationChannel]) -> bool: diff --git a/app/application/messaging/router.py b/app/application/messaging/router.py index 080fd87d5..e82b963aa 100644 --- a/app/application/messaging/router.py +++ b/app/application/messaging/router.py @@ -1,7 +1,7 @@ """ 交互路由层：统一选择活动文本会话，并按固定顺序派发按钮回调。 -文本会话候选覆盖 Site、Subscribe、Skill、Media 四类， +文本会话候选覆盖 Site、Subscribe、Skill、Media、Update 五类， 按会话创建时间选择最近激活的一条，避免旧会话抢占新会话的输入。 """ @@ -16,6 +16,7 @@ from app.application.messaging.media import media_interaction_manager from app.application.messaging.site import site_interaction_manager from app.application.messaging.skill import skill_interaction_manager from app.application.messaging.subscribe import subscribe_interaction_manager +from app.application.messaging.update import update_interaction_manager @dataclass(frozen=True, slots=True) @@ -117,5 +118,6 @@ def has_pending_interaction(user_id: Union[str, int]) -> bool: subscribe_interaction_manager, skill_interaction_manager, media_interaction_manager, + update_interaction_manager, ) ) diff --git a/app/application/messaging/update.py b/app/application/messaging/update.py new file mode 100644 index 000000000..c48accfde --- /dev/null +++ b/app/application/messaging/update.py @@ -0,0 +1,999 @@ +"""通知渠道中的主程序更新检查、下载进度和重启确认交互。""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable, Coroutine +from dataclasses import dataclass +from threading import Lock +from typing import Any, Optional, Protocol, Union + +from app.application.messaging.interaction import ( + MessageGateway, + PendingSlashInteraction, + SlashInteractionManager, + supports_interaction_buttons, + update_or_post_message, +) +from app.application.system import SystemOperationResult +from app.runtime.log import logger +from app.schemas.message import Message +from app.schemas.notification import ChannelCapabilityManager +from app.schemas.system import SystemUpdateItemStatus, SystemUpdateStatus, SystemUpdateType +from app.schemas.types import NotificationChannel + +update_interaction_manager = SlashInteractionManager() + +_monitor_lock = Lock() +_monitored_requests: set[str] = set() + + +class SystemUpdateInteractionActions(Protocol): + """声明更新交互调用的主程序升级应用用例。""" + + def update_status(self) -> SystemUpdateStatus: + """读取当前后台更新状态。""" + ... + + def check_update(self) -> SystemUpdateStatus: + """立即检查主程序正式版本更新。""" + ... + + def download_update( + self, + target: SystemUpdateType = "application", + ) -> SystemOperationResult: + """启动主程序更新包下载。""" + ... + + def install_update( + self, + target: SystemUpdateType = "application", + ) -> SystemOperationResult: + """确认主程序更新包并请求重启安装。""" + ... + + +UpdateMonitorSubmitter = Callable[[Coroutine[Any, Any, None]], Any] +RestartMarker = Callable[[NotificationChannel, Union[str, int], Optional[str]], None] +RestartMarkerClearer = Callable[[], None] + + +@dataclass(frozen=True, slots=True) +class SystemUpdateInteractionView: + """保存一次渠道无关的更新交互展示内容。""" + + title: str + text: str + buttons: Optional[list[list[dict[str, str]]]] = None + + +class SystemUpdateInteractionHandler: + """编排 `/update` 的检查、下载进度编辑和重启确认流程。""" + + _poll_interval_seconds = 3.0 + _terminal_download_states = {"idle", "available", "ready", "failed", "installing"} + + def __init__( + self, + *, + messenger: MessageGateway, + actions: SystemUpdateInteractionActions, + submit_monitor: UpdateMonitorSubmitter, + mark_restart: RestartMarker, + clear_restart_marker: RestartMarkerClearer, + poll_interval_seconds: float = _poll_interval_seconds, + ) -> None: + """注入消息网关、系统更新用例和受管后台任务提交器。""" + self._messenger = messenger + self._actions = actions + self._submit_monitor = submit_monitor + self._mark_restart = mark_restart + self._clear_restart_marker = clear_restart_marker + self._poll_interval_seconds = max(0.0, poll_interval_seconds) + + def remote_update( + self, + arg_str: str = "", + channel: Optional[NotificationChannel] = None, + userid: Optional[Union[str, int]] = None, + source: Optional[str] = None, + ) -> None: + """执行 `/update`，检查正式版本并创建后续确认会话。""" + if channel is None or userid is None: + return + request = update_interaction_manager.create_or_replace( + user_id=userid, + command="/update", + channel=channel, + source=source, + username=None, + ) + try: + status = self._actions.check_update() + except Exception as error: # noqa: BLE001 交互入口必须回显稳定错误 + logger.warning(f"检查 MoviePilot 更新失败：{error}") + self._render_check_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username="", + error=str(error), + ) + return + + normalized_arg = str(arg_str or "").strip() + if normalized_arg and self.handle_text_interaction( + channel=channel, + source=source, + userid=userid, + username="", + text=normalized_arg, + ): + return + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username="", + ) + + @staticmethod + def parse_callback(callback_data: str) -> Optional[tuple[str, str]]: + """解析 `/update` 的按钮回调。""" + if not str(callback_data or "").startswith("update:"): + return None + parts = str(callback_data).split(":") + if len(parts) != 3 or not parts[1] or not parts[2]: + return None + return parts[1], parts[2] + + def handle_callback_interaction( + self, + callback_data: str, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> bool: + """消费 `/update` 按钮回调并保持原消息作为进度锚点。""" + parsed = self.parse_callback(callback_data) + if not parsed: + return False + request_id, action = parsed + request = update_interaction_manager.get_by_id(request_id, userid) + if request is None: + self._messenger.post_message( + Message( + channel=channel, + source=source, + userid=userid, + username=username, + title="升级交互已失效，请重新发送 /update", + save_history=False, + ) + ) + return True + + request.channel = channel + request.source = source + request.username = username + if action == "close": + update_interaction_manager.remove(request.request_id) + update_or_post_message( + chain=self._messenger, + channel=channel, + source=source, + userid=userid, + username=username, + title="MoviePilot 更新", + text="本次升级交互已结束，已开始的后台下载不会被取消。", + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return True + if action == "refresh": + self._check_and_render( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return True + if action == "download": + self._start_download( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return True + if action == "install": + self._install_and_restart( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return True + return False + + def handle_text_interaction( + self, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + text: str, + ) -> bool: + """消费不支持按钮渠道或用户主动输入的升级确认文本。""" + request = update_interaction_manager.get_by_user(userid) + if request is None: + return False + request.channel = channel + request.source = source + request.username = username + normalized = str(text or "").strip().lower() + + if normalized in {"稍后", "取消", "关闭", "退出", "cancel", "close", "quit", "exit"}: + update_interaction_manager.remove(request.request_id) + self._messenger.post_message( + Message( + channel=channel, + source=source, + userid=userid, + username=username, + title="本次升级交互已结束", + save_history=False, + ) + ) + return True + if normalized in {"刷新", "检查", "状态", "refresh", "check", "status"}: + self._check_and_render( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + ) + return True + + if normalized == "确认": + normalized = "确认重启" if request.awaiting_input == "install" else "确认升级" + if normalized in {"确认升级", "升级", "下载", "重试", "update", "download", "retry"}: + self._start_download( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + ) + return True + if normalized in {"确认重启", "重启", "安装", "restart", "install"}: + self._install_and_restart( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + ) + return True + + try: + status = self._actions.update_status() + except Exception as error: # noqa: BLE001 文本交互错误必须回显 + self._render_check_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=str(error), + ) + return True + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username=username, + ) + return True + + def _check_and_render( + self, + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """重新检查版本并在原交互消息中展示结果。""" + try: + status = self._actions.check_update() + except Exception as error: # noqa: BLE001 交互入口必须回显稳定错误 + logger.warning(f"检查 MoviePilot 更新失败：{error}") + self._render_check_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=str(error), + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def _start_download( + self, + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """启动主程序下载，立即展示状态并登记持续进度监视。""" + try: + result = self._actions.download_update("application") + status = result.data if isinstance(result.data, SystemUpdateStatus) else self._actions.update_status() + except Exception as error: # noqa: BLE001 交互入口必须回显稳定错误 + logger.warning(f"启动 MoviePilot 更新下载失败：{error}") + self._render_operation_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=str(error), + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return + + operation_error = None if result.success else result.message or "无法启动更新包下载" + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username=username, + operation_error=operation_error, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + item = self._application_item(status) + if result.success and item.state == "downloading": + self._schedule_monitor( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + initial_item=item, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def _install_and_restart( + self, + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """先更新提示并记录重启目标，再确认安装和请求受管重启。""" + try: + status = self._actions.update_status() + except Exception as error: # noqa: BLE001 交互入口必须回显稳定错误 + self._render_operation_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=str(error), + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return + item = self._application_item(status) + if item.state != "ready": + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username=username, + operation_error="更新包尚未下载完成", + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return + + installing = item.model_copy(update={"state": "installing", "can_install": False}) + self._render_item( + request=request, + item=installing, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + self._mark_restart(channel, userid, source) + try: + result = self._actions.install_update("application") + except Exception as error: # noqa: BLE001 重启失败必须恢复交互 + self._clear_restart_marker() + logger.warning(f"安装 MoviePilot 更新失败：{error}") + self._render_operation_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=str(error), + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return + if result.success: + update_interaction_manager.remove(request.request_id) + return + + self._clear_restart_marker() + try: + status = self._actions.update_status() + except Exception: # noqa: BLE001 优先保留安装用例的稳定错误 + status = SystemUpdateStatus( + state="ready", + current_version=item.current_version or "unknown", + updates=[item], + ) + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username=username, + operation_error=result.message or "无法重启并安装更新", + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def _schedule_monitor( + self, + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + initial_item: SystemUpdateItemStatus, + original_message_id: Optional[Union[str, int]], + original_chat_id: Optional[str], + ) -> None: + """确保同一交互只登记一个非阻塞下载进度监视任务。""" + with _monitor_lock: + if request.request_id in _monitored_requests: + return + _monitored_requests.add(request.request_id) + monitor = self._monitor_download( + request_id=request.request_id, + channel=channel, + source=source, + userid=userid, + username=username, + initial_item=initial_item, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + try: + self._submit_monitor(monitor) + except Exception as error: # noqa: BLE001 下载继续运行，交互降级为手动刷新 + monitor.close() + with _monitor_lock: + _monitored_requests.discard(request.request_id) + logger.warning(f"登记 MoviePilot 更新进度监视失败：{error}") + self._messenger.post_message( + Message( + channel=channel, + source=source, + userid=userid, + username=username, + title="更新包已开始下载", + text="自动进度更新暂不可用，可重新发送 /update 查看状态。", + original_message_id=original_message_id, + original_chat_id=original_chat_id, + save_history=False, + ) + ) + + async def _monitor_download( + self, + *, + request_id: str, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + initial_item: SystemUpdateItemStatus, + original_message_id: Optional[Union[str, int]], + original_chat_id: Optional[str], + ) -> None: + """按 Web 端三秒节奏轮询状态，并持续编辑原消息直到下载终态。""" + last_fingerprint = self._item_fingerprint(initial_item) + last_progress_bucket = initial_item.progress // 10 + edit_fallback_sent = False + try: + while True: + await asyncio.sleep(self._poll_interval_seconds) + request = update_interaction_manager.get_by_id(request_id, userid) + if request is None: + return + status = await asyncio.to_thread(self._actions.update_status) + item = self._application_item(status) + fingerprint = self._item_fingerprint(item) + if fingerprint != last_fingerprint: + request.awaiting_input = self._awaiting_input(item) + view = self._build_view(request=request, item=item, channel=channel) + if original_message_id and original_chat_id and ChannelCapabilityManager.supports_editing(channel): + edited = await asyncio.to_thread( + self._edit_view, + view=view, + channel=channel, + source=source, + userid=userid, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + if not edited and (not edit_fallback_sent or item.state in self._terminal_download_states): + await asyncio.to_thread( + self._post_view, + view=view, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + edit_fallback_sent = True + else: + progress_bucket = item.progress // 10 + if progress_bucket != last_progress_bucket or item.state in self._terminal_download_states: + await asyncio.to_thread( + self._post_view, + view=view, + channel=channel, + source=source, + userid=userid, + username=username, + ) + last_progress_bucket = progress_bucket + last_fingerprint = fingerprint + + if item.state != "downloading": + if item.state == "idle": + update_interaction_manager.remove(request_id) + return + except Exception as error: # noqa: BLE001 监视错误不能影响实际下载 + logger.warning(f"监视 MoviePilot 更新下载进度失败：{error}") + request = update_interaction_manager.get_by_id(request_id, userid) + if request is not None: + await asyncio.to_thread( + self._render_operation_failure, + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=f"读取下载进度失败：{error}", + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + finally: + with _monitor_lock: + _monitored_requests.discard(request_id) + + def _render_status( + self, + *, + request: PendingSlashInteraction, + status: SystemUpdateStatus, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + operation_error: Optional[str] = None, + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """从聚合更新快照提取主程序状态并更新交互消息。""" + item = self._application_item(status) + self._render_item( + request=request, + item=item, + channel=channel, + source=source, + userid=userid, + username=username, + operation_error=operation_error, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + if item.state == "idle" and not item.error and not operation_error: + update_interaction_manager.remove(request.request_id) + + def _render_item( + self, + *, + request: PendingSlashInteraction, + item: SystemUpdateItemStatus, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + operation_error: Optional[str] = None, + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """更新会话阶段并优先编辑原消息展示指定主程序状态。""" + request.awaiting_input = self._awaiting_input(item) + view = self._build_view( + request=request, + item=item, + channel=channel, + operation_error=operation_error, + ) + update_or_post_message( + chain=self._messenger, + channel=channel, + source=source, + userid=userid, + username=username, + title=view.title, + text=view.text, + buttons=view.buttons, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def _render_check_failure( + self, + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + error: str, + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """展示版本检查失败并保留刷新入口。""" + request.awaiting_input = "refresh" + buttons = self._buttons( + request=request, + channel=channel, + actions=(("重新检查", "refresh"), ("关闭", "close")), + ) + text = f"{error or '无法读取更新状态'}\n\n" + if not buttons: + text += "回复“刷新”重试，回复“关闭”结束。" + update_or_post_message( + chain=self._messenger, + channel=channel, + source=source, + userid=userid, + username=username, + title="检查 MoviePilot 更新失败", + text=text.strip(), + buttons=buttons, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def _render_operation_failure( + self, + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + error: str, + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """读取最新状态后展示下载或安装动作失败。""" + try: + status = self._actions.update_status() + except Exception: # noqa: BLE001 保留原始动作错误 + self._render_check_failure( + request=request, + channel=channel, + source=source, + userid=userid, + username=username, + error=error, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + return + self._render_status( + request=request, + status=status, + channel=channel, + source=source, + userid=userid, + username=username, + operation_error=error, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def _build_view( + self, + *, + request: PendingSlashInteraction, + item: SystemUpdateItemStatus, + channel: NotificationChannel, + operation_error: Optional[str] = None, + ) -> SystemUpdateInteractionView: + """把主程序更新状态转换为与 Web 流程一致的渠道展示。""" + actions: tuple[tuple[str, str], ...] + if item.state == "available": + title = "发现 MoviePilot 主程序更新" + lines = [f"当前版本：{item.current_version or '未知'}", f"目标版本：{item.version or '未知'}"] + if item.frontend_version: + lines.append(f"配套前端：{item.frontend_version}") + if item.release_name and item.release_name != item.version: + lines.append(f"发布名称：{item.release_name}") + if item.published_at: + lines.append(f"发布时间：{item.published_at}") + notes = self._release_notes(item.release_notes) + if notes: + lines.extend(("", "更新说明：", notes)) + actions = (("确认升级", "download"), ("稍后", "close")) + fallback = "回复“确认升级”开始下载，回复“稍后”关闭本次交互。" + elif item.state == "downloading": + title = "正在下载 MoviePilot 更新" + lines = [ + f"目标版本：{item.version or '未知'}", + self._progress_line(item), + "下载完成后会继续提示确认重启。", + ] + actions = () + fallback = "" + elif item.state == "ready": + title = "MoviePilot 更新包已准备完成" + lines = [ + f"目标版本：{item.version or '未知'}", + "更新包已下载并校验完成。确认后系统将重启并安装更新。", + ] + if item.frontend_version: + lines.insert(1, f"配套前端：{item.frontend_version}") + actions = (("确认重启", "install"), ("稍后重启", "close")) + fallback = "回复“确认重启”开始安装，回复“稍后”关闭本次交互。" + elif item.state == "installing": + title = "正在重启并安装 MoviePilot 更新" + lines = ["重启请求已提交，请等待服务恢复。"] + actions = () + fallback = "" + elif item.state == "failed": + title = "MoviePilot 更新下载失败" + lines = [item.error or "更新包下载失败，请重试。"] + actions = (("重试", "download"), ("关闭", "close")) + fallback = "回复“重试”重新下载，回复“关闭”结束。" + elif item.error: + title = "检查 MoviePilot 更新失败" + lines = [item.error] + actions = (("重新检查", "refresh"), ("关闭", "close")) + fallback = "回复“刷新”重试，回复“关闭”结束。" + else: + title = "MoviePilot 已是最新版本" + lines = [f"当前版本：{item.current_version or '未知'}"] + actions = () + fallback = "" + + if operation_error: + title = "MoviePilot 升级操作失败" + lines = [operation_error, "", *lines] + buttons = self._buttons(request=request, channel=channel, actions=actions) + if fallback and not buttons: + lines.extend(("", fallback)) + return SystemUpdateInteractionView( + title=title, + text="\n".join(lines).strip(), + buttons=buttons, + ) + + @staticmethod + def _application_item(status: SystemUpdateStatus) -> SystemUpdateItemStatus: + """读取主程序明细，并兼容旧版只有聚合字段的状态。""" + item = next((value for value in status.updates if value.type == "application"), None) + if item is not None: + return item + return SystemUpdateItemStatus( + type="application", + state=status.state, + current_version=status.current_version, + version=status.version, + frontend_version=status.frontend_version, + release_name=status.release_name, + release_notes=status.release_notes, + published_at=status.published_at, + checked_at=status.checked_at, + downloaded_bytes=status.downloaded_bytes, + total_bytes=status.total_bytes, + progress=status.progress, + error=status.error, + can_update=status.can_update, + can_install=status.can_install, + ) + + @staticmethod + def _awaiting_input(item: SystemUpdateItemStatus) -> Optional[str]: + """把更新状态映射为文本渠道下一步输入阶段。""" + if item.state in {"available", "failed"}: + return "download" + if item.state == "ready": + return "install" + if item.error: + return "refresh" + return None + + @staticmethod + def _item_fingerprint(item: SystemUpdateItemStatus) -> tuple[Any, ...]: + """生成需要刷新消息的状态指纹。""" + return ( + item.state, + item.version, + item.frontend_version, + item.downloaded_bytes, + item.total_bytes, + item.progress, + item.error, + item.can_update, + item.can_install, + ) + + @staticmethod + def _release_notes(notes: Optional[str], limit: int = 1200) -> str: + """限制发布说明长度，避免超过通知渠道单条消息上限。""" + normalized = str(notes or "").strip() + if len(normalized) <= limit: + return normalized + return f"{normalized[:limit].rstrip()}..." + + @staticmethod + def _format_bytes(value: int) -> str: + """使用与 Web 端相同的 MB 展示下载量。""" + megabytes = max(0, int(value or 0)) / 1024 / 1024 + return f"{megabytes:.0f} MB" if megabytes >= 100 else f"{megabytes:.1f} MB" + + @classmethod + def _progress_line(cls, item: SystemUpdateItemStatus) -> str: + """构造宽度稳定的文本进度条和下载量。""" + progress = min(100, max(0, int(item.progress or 0))) + filled = min(10, progress // 10) + progress_bar = f"[{'=' * filled}{'.' * (10 - filled)}] {progress}%" + downloaded = cls._format_bytes(item.downloaded_bytes) + if item.total_bytes > 0: + return f"{progress_bar}\n{downloaded} / {cls._format_bytes(item.total_bytes)}" + return f"{progress_bar}\n已下载 {downloaded}" + + @staticmethod + def _buttons( + *, + request: PendingSlashInteraction, + channel: NotificationChannel, + actions: tuple[tuple[str, str], ...], + ) -> Optional[list[list[dict[str, str]]]]: + """为支持按钮回调的渠道构造单行操作按钮。""" + if not actions or not supports_interaction_buttons(channel): + return None + return [ + [ + { + "text": text, + "callback_data": f"update:{request.request_id}:{action}", + } + for text, action in actions + ] + ] + + def _edit_view( + self, + *, + view: SystemUpdateInteractionView, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + original_message_id: Union[str, int], + original_chat_id: str, + ) -> bool: + """直接编辑进度锚点，避免轮询阶段重复发送消息。""" + if not source: + return False + metadata = {"userid": userid} if channel == NotificationChannel.WebAgent else None + return bool( + self._messenger.edit_message( + channel=channel, + source=source, + message_id=original_message_id, + chat_id=original_chat_id, + title=view.title, + text=view.text, + buttons=view.buttons, + metadata=metadata, + ) + ) + + def _post_view( + self, + *, + view: SystemUpdateInteractionView, + channel: NotificationChannel, + source: Optional[str], + userid: Union[str, int], + username: Optional[str], + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> None: + """在无法编辑时发送一次进度或终态消息。""" + self._messenger.post_message( + Message( + channel=channel, + source=source, + userid=userid, + username=username, + title=view.title, + text=view.text, + buttons=view.buttons, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + save_history=False, + ) + ) diff --git a/app/chain/base.py b/app/chain/base.py index 84c26b49e..84d004434 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -74,6 +74,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, met self.classification_service = context.classification_service self.runtime_config = context.configuration self.stop_state = context.stop_state + self.system_service = context.system_service self.durable_event_writer = context.durable_event_writer self._module_dispatcher = context.module_dispatcher_factory( module_catalog=self.modulemanager, diff --git a/app/chain/message.py b/app/chain/message.py index bd3a7931d..7d7489d03 100644 --- a/app/chain/message.py +++ b/app/chain/message.py @@ -9,7 +9,7 @@ from concurrent.futures import CancelledError as FutureCancelledError from dataclasses import dataclass from datetime import datetime from pathlib import Path -from typing import Any, Dict, List, Optional, Protocol, Tuple, Union +from typing import Any, Callable, Dict, List, Optional, Protocol, Tuple, Union from urllib.parse import unquote, urlparse from app.application.agent import ( @@ -27,10 +27,12 @@ from app.application.messaging.session import MessageSessionService from app.application.messaging.site import site_interaction_manager from app.application.messaging.skill import SkillInteractionHandler, skill_interaction_manager from app.application.messaging.subscribe import subscribe_interaction_manager +from app.application.messaging.update import update_interaction_manager from app.chain.base import ChainBase from app.chain.interaction import MediaInteractionChain as _MediaInteractionChain from app.chain.site import SiteChain from app.chain.subscribe.facade import SubscribeChain +from app.chain.system import SystemChain from app.chain.transfer.facade import TransferChain from app.runtime.log import logger from app.runtime.loop import main_loop_registry @@ -705,7 +707,9 @@ class MessageChain(ChainBase): def _interaction_router(self) -> InteractionRouter: """构造交互路由器，文本会话按创建时间选择，回调路由注册顺序即优先级。""" - def session_text(handle): + def session_text( + handle: Callable[..., Any], + ) -> Callable[[InteractionContext, str], bool]: """包装传统交互入口为会话路由的文本处理函数，保持懒构造。""" def _handle(context: InteractionContext, text: str) -> bool: return bool(handle( @@ -717,7 +721,9 @@ class MessageChain(ChainBase): )) return _handle - def callback_dispatch(handle): + def callback_dispatch( + handle: Callable[..., Any], + ) -> Callable[[str, InteractionContext], InteractionDispatch]: """包装传统回调入口为回调路由的派发函数，保持懒构造。""" def _dispatch(callback_data: str, context: InteractionContext) -> InteractionDispatch: return InteractionDispatch(handled=bool(handle( @@ -754,6 +760,11 @@ class MessageChain(ChainBase): get_pending=media_interaction_manager.get_by_user, handle_text=session_text(lambda **kw: _MediaInteractionChain().handle_text_interaction(**kw)), ), + SessionRoute( + name="update", + get_pending=update_interaction_manager.get_by_user, + handle_text=session_text(lambda **kw: SystemChain().handle_update_text_interaction(**kw)), + ), ] def _dispatch_agent_choice(callback_data: str, context: InteractionContext) -> InteractionDispatch: @@ -820,6 +831,11 @@ class MessageChain(ChainBase): matches=lambda data: _MediaInteractionChain.parse_callback(data) is not None, dispatch=callback_dispatch(lambda **kw: _MediaInteractionChain().handle_callback_interaction(**kw)), ), + CallbackRoute( + name="update", + matches=lambda data: data.startswith("update:"), + dispatch=callback_dispatch(lambda **kw: SystemChain().handle_update_callback_interaction(**kw)), + ), CallbackRoute( name="agent_choice", matches=lambda data: parse_agent_choice_callback(data) is not None, diff --git a/app/chain/system.py b/app/chain/system.py index 703e5d725..022638345 100644 --- a/app/chain/system.py +++ b/app/chain/system.py @@ -4,15 +4,18 @@ import re import shutil import threading import uuid -from collections.abc import Mapping +from collections.abc import Coroutine, Mapping from pathlib import Path from typing import Any, Optional, Protocol, Union from app.application.configuration import get_chain_runtime_config_snapshot +from app.application.messaging.update import SystemUpdateInteractionHandler from app.chain.base import ChainBase from app.runtime import version as runtime_version from app.runtime.log import logger +from app.runtime.loop import main_loop_registry from app.runtime.state import SystemHelper +from app.runtime.tasks import get_task_registry from app.schemas.message import Message from app.schemas.notification import NotificationChannel @@ -105,8 +108,99 @@ class SystemChain(ChainBase): """ _restart_file = "__system_restart__" + _update_restart_file = "__system_update_restart__" _plugin_restore_pending_file = "__plugin_restore_pending__" + def _update_interaction_handler(self) -> SystemUpdateInteractionHandler: + """构造复用当前消息网关和系统应用服务的更新交互控制器。""" + if self.system_service is None: + raise RuntimeError("系统更新服务尚未由启动组合根装配") + return SystemUpdateInteractionHandler( + messenger=self, + actions=self.system_service, + submit_monitor=self._submit_update_monitor, + mark_restart=self._mark_update_restart, + clear_restart_marker=self._clear_update_restart_marker, + ) + + @staticmethod + def _submit_update_monitor(monitor: Coroutine[Any, Any, None]) -> None: + """把进度监视协程跨线程登记到宿主事件循环。""" + get_task_registry().submit_threadsafe( + monitor, + loop=main_loop_registry.require(), + owner="chain.system.update_progress", + ) + + def _mark_update_restart( + self, + channel: NotificationChannel, + userid: Union[int, str], + source: Optional[str], + ) -> None: + """记录升级重启的回复目标，供服务恢复后发送完成通知。""" + self.save_cache( + {"channel": channel.value, "userid": userid, "source": source}, + self._update_restart_file, + ) + + def _clear_update_restart_marker(self) -> None: + """升级安装未能进入重启阶段时删除完成通知标记。""" + self.remove_cache(self._update_restart_file) + + def remote_update( + self, + arg_str: str = "", + channel: Optional[NotificationChannel] = None, + userid: Optional[Union[int, str]] = None, + source: Optional[str] = None, + ) -> None: + """检查正式版本并启动通知渠道中的升级确认交互。""" + self._update_interaction_handler().remote_update( + arg_str=arg_str, + channel=channel, + userid=userid, + source=source, + ) + + def handle_update_callback_interaction( + self, + callback_data: str, + channel: NotificationChannel, + source: Optional[str], + userid: Union[int, str], + username: Optional[str], + original_message_id: Optional[Union[str, int]] = None, + original_chat_id: Optional[str] = None, + ) -> bool: + """处理更新交互按钮并把原消息定位参数交给进度编辑器。""" + return self._update_interaction_handler().handle_callback_interaction( + callback_data=callback_data, + channel=channel, + source=source, + userid=userid, + username=username, + original_message_id=original_message_id, + original_chat_id=original_chat_id, + ) + + def handle_update_text_interaction( + self, + channel: NotificationChannel, + source: Optional[str], + userid: Union[int, str], + username: Optional[str], + text: str, + ) -> bool: + """处理不支持按钮渠道中的升级或重启确认文本。""" + return self._update_interaction_handler().handle_text_interaction( + channel=channel, + source=source, + userid=userid, + username=username, + text=text, + ) + def remote_clear_cache(self, channel: NotificationChannel, userid: Union[int, str], source: Optional[str] = None): """ 清理系统缓存 @@ -445,6 +539,35 @@ class SystemChain(ChainBase): save_history=False)) self.remove_cache(self._restart_file) + update_restart_channel = self.load_cache(self._update_restart_file) + if update_restart_channel: + if not isinstance(update_restart_channel, dict): + update_restart_channel = json.loads(update_restart_channel) + channel = next( + ( + candidate + for candidate in NotificationChannel.__members__.values() + if candidate.value == update_restart_channel.get("channel") + ), + None, + ) + userid = update_restart_channel.get("userid") + source = update_restart_channel.get("source") + self.post_message( + Message( + channel=channel, + source=source, + title=( + "MoviePilot 更新安装完成！\n" + f"当前后端版本：{runtime_version.get_app_version()}\n" + f"当前前端版本：{runtime_version.get_frontend_version()}" + ), + userid=userid, + save_history=False, + ) + ) + self.remove_cache(self._update_restart_file) + @staticmethod def __get_server_release_version(): """ diff --git a/app/command.py b/app/command.py index 7b65a7487..ebd9db14e 100644 --- a/app/command.py +++ b/app/command.py @@ -110,6 +110,12 @@ class Command(metaclass=Singleton): "category": "管理", "data": {}, }, + "/update": { + "func": SystemChain().remote_update, + "description": "检查更新", + "category": "管理", + "data": {}, + }, "/version": { "func": SystemChain().version, "description": "当前版本", diff --git a/app/modules/slack/slack.py b/app/modules/slack/slack.py index 7fe51f351..b3eab98d1 100644 --- a/app/modules/slack/slack.py +++ b/app/modules/slack/slack.py @@ -1,21 +1,20 @@ import json import re -from threading import Lock from pathlib import Path +from threading import Lock from typing import Any, Dict, List, Optional, Tuple from slack_bolt import App from slack_bolt.adapter.socket_mode import SocketModeHandler from slack_sdk import WebClient -from app.runtime.settings import get_runtime_setting - -from app.application.messaging.ingress import forward_message_to_host -from app.domain.context import MediaInfo, Context -from app.domain.metainfo import MetaInfo -from app.runtime.log import logger from app.adapters.network.http import RequestUtils +from app.application.messaging.ingress import forward_message_to_host +from app.domain.context import Context, MediaInfo +from app.domain.metainfo import MetaInfo from app.foundation import size as size_tools +from app.runtime.log import logger +from app.runtime.settings import get_runtime_setting lock = Lock() @@ -303,75 +302,79 @@ class Slack: # 消息广播 channel = self.__find_public_channel() # 消息文本 - message_text = "" + message_text = f"{title}\n{text or ''}" # 结构体 blocks = [] - if not image: - message_text = f"{title}\n{text or ''}" - else: + if image: # 消息图片 - if image: - # 拼装消息内容 - blocks.append({"type": "section", "text": { - "type": "mrkdwn", - "text": f"*{title}*\n{text or ''}" - }, 'accessory': { + blocks.append({"type": "section", "text": { + "type": "mrkdwn", + "text": f"*{title}*\n{text or ''}" + }, 'accessory': { "type": "image", "image_url": f"{image}", "alt_text": f"{title}" - }}) - # 自定义按钮 - if buttons: - for button_row in buttons: - elements = [] - for button in button_row: - if "url" in button: - # URL按钮 - elements.append({ - "type": "button", - "text": { - "type": "plain_text", - "text": button["text"], - "emoji": True - }, - "url": button["url"], - "action_id": f"actionId-url-{button.get('text', 'url')}-{len(elements)}" - }) - else: - # 回调按钮 - elements.append({ - "type": "button", - "text": { - "type": "plain_text", - "text": button["text"], - "emoji": True - }, - "value": button["callback_data"], - "action_id": f"actionId-{button['callback_data']}" - }) - if elements: - blocks.append({ - "type": "actions", - "elements": elements - }) - elif link: - # 默认链接按钮 - blocks.append({ - "type": "actions", - "elements": [ - { + }}) + elif buttons or link: + blocks.append({ + "type": "section", + "text": { + "type": "mrkdwn", + "text": f"*{title}*\n{text or ''}", + }, + }) + # 自定义按钮 + if buttons: + for button_row in buttons: + elements = [] + for button in button_row: + if "url" in button: + # URL按钮 + elements.append({ "type": "button", "text": { "type": "plain_text", - "text": "查看详情", + "text": button["text"], "emoji": True }, - "value": "click_me_url", - "url": f"{link}", - "action_id": "actionId-url" - } - ] - }) + "url": button["url"], + "action_id": f"actionId-url-{button.get('text', 'url')}-{len(elements)}" + }) + else: + # 回调按钮 + elements.append({ + "type": "button", + "text": { + "type": "plain_text", + "text": button["text"], + "emoji": True + }, + "value": button["callback_data"], + "action_id": f"actionId-{button['callback_data']}" + }) + if elements: + blocks.append({ + "type": "actions", + "elements": elements + }) + elif link: + # 默认链接按钮 + blocks.append({ + "type": "actions", + "elements": [ + { + "type": "button", + "text": { + "type": "plain_text", + "text": "查看详情", + "emoji": True + }, + "value": "click_me_url", + "url": f"{link}", + "action_id": "actionId-url" + } + ] + }) # 判断是编辑消息还是发送新消息 if original_message_id and original_chat_id: diff --git a/app/startup/composition/chain.py b/app/startup/composition/chain.py index cb0ad59f4..0f1705a77 100644 --- a/app/startup/composition/chain.py +++ b/app/startup/composition/chain.py @@ -23,6 +23,7 @@ from app.application.image import ( configure_wallpaper_providers, reset_wallpaper_providers, ) +from app.application.system import SystemService from app.chain._recognition import ( RecognitionSharePort, configure_recognition_share_port, @@ -85,6 +86,7 @@ def build_chain_runtime_context( system_config: SystemConfigOper, configuration: Callable[[], ChainRuntimeConfig], classification_service: ClassificationExecutionService, + system_service: SystemService, ) -> ChainRuntimeContext: """创建 Chain 无参兼容入口共享的运行时对象与数据端口。""" return ChainRuntimeContext( @@ -125,6 +127,7 @@ def build_chain_runtime_context( configuration=configuration(), durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory), stop_state=runtime_stop_state, + system_service=system_service, ) @@ -153,6 +156,7 @@ def configure_chain_runtime_context( system_config: SystemConfigOper, configuration: Callable[[], ChainRuntimeConfig], classification_service: ClassificationExecutionService, + system_service: SystemService, ) -> None: """登记按需构造的 Chain 上下文，保持无参 Chain 的插件兼容合同。""" configure_chain_runtime_context_provider( @@ -161,6 +165,7 @@ def configure_chain_runtime_context( system_config=system_config, configuration=configuration, classification_service=classification_service, + system_service=system_service, ) ) diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index c72f554c8..1fb404b5e 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -711,6 +711,7 @@ async def _initialize_modules() -> HostRuntime: system_config=system_config, configuration=configuration.runtime.chain, classification_service=classification.execution, + system_service=host_runtime.system, ) # 认证访问层不反向依赖数据库实现，由启动组合层注入载荷提供器。 configure_security_access() diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 9f6278f5d..e68d98e61 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -102,7 +102,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,528 / 513 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | +| 全量 mypy 历史债务 | 9,508 / 513 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | | Ruff 历史诊断 | 547 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | diff --git a/mypy.ini b/mypy.ini index 751b76ab9..3466e52c7 100644 --- a/mypy.ini +++ b/mypy.ini @@ -30,6 +30,7 @@ files = app/application/chain/context.py, app/application/chain/events.py, app/application/messaging/ingress.py, + app/application/messaging/update.py, app/application/subscription/delete.py, app/application/subscription/identity.py, app/application/subscription/mutation.py, diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index bb88ebc2e..7f2c7ae0b 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -679,7 +679,7 @@ "type-arg": 8 }, "app/application/messaging/interaction.py": { - "no-untyped-def": 4, + "no-untyped-def": 1, "type-arg": 2 }, "app/application/messaging/media.py": { @@ -707,7 +707,6 @@ "app/application/messaging/site.py": { "arg-type": 1, "assignment": 2, - "no-untyped-call": 1, "no-untyped-def": 2 }, "app/application/messaging/skill.py": { @@ -718,7 +717,6 @@ "app/application/messaging/subscribe.py": { "arg-type": 1, "assignment": 2, - "no-untyped-call": 1, "no-untyped-def": 2 }, "app/application/module.py": { @@ -890,11 +888,10 @@ "type-arg": 1 }, "app/chain/message.py": { - "arg-type": 21, + "arg-type": 18, "assignment": 2, "no-any-return": 11, - "no-untyped-call": 8, - "no-untyped-def": 6, + "no-untyped-def": 4, "type-arg": 2, "union-attr": 1, "var-annotated": 1 @@ -922,7 +919,7 @@ "var-annotated": 1 }, "app/chain/site.py": { - "arg-type": 2, + "arg-type": 1, "assignment": 6, "attr-defined": 1, "misc": 3, @@ -1111,7 +1108,7 @@ "union-attr": 2 }, "app/command.py": { - "arg-type": 3, + "arg-type": 2, "assignment": 5, "misc": 4, "no-untyped-call": 4, diff --git a/tests/test_chain_runtime_context.py b/tests/test_chain_runtime_context.py index fe265e256..b56b023e4 100644 --- a/tests/test_chain_runtime_context.py +++ b/tests/test_chain_runtime_context.py @@ -46,6 +46,7 @@ def _context() -> ChainRuntimeContext: user_repository=Mock(), configuration=ChainRuntimeConfig(media_extensions=(".mkv",)), durable_event_writer=Mock(), + system_service=Mock(), ) @@ -60,6 +61,7 @@ def test_chain_accepts_explicit_runtime_context() -> None: assert chain.eventmanager is context.event_manager assert chain.messagehelper is context.message_helper assert chain.durable_event_writer is context.durable_event_writer + assert chain.system_service is context.system_service context.message_queue.bind.assert_called_once_with(chain.run_module) @@ -141,6 +143,7 @@ def test_chain_composition_registers_lazy_compatibility_provider(monkeypatch) -> "system_config": Mock(), "configuration": Mock(), "classification_service": Mock(), + "system_service": Mock(), } monkeypatch.setattr(chain_composition, "build_chain_runtime_context", builder) monkeypatch.setattr( diff --git a/tests/test_slack_command_registration.py b/tests/test_slack_command_registration.py index 09c9226a1..d9bc50130 100644 --- a/tests/test_slack_command_registration.py +++ b/tests/test_slack_command_registration.py @@ -3,7 +3,7 @@ from unittest.mock import Mock, patch from app.modules.slack import SlackModule from app.modules.slack.slack import Slack -from app.schemas import CommandRegisterEventData +from app.schemas.event import CommandRegisterEventData def test_slack_module_register_commands_filters_event_subset(): @@ -33,9 +33,7 @@ def test_slack_module_register_commands_filters_event_subset(): ): module.register_commands(original_commands) - client.register_commands.assert_called_once_with( - {"/sites": {"description": "管理站点"}} - ) + client.register_commands.assert_called_once_with({"/sites": {"description": "管理站点"}}) client.delete_commands.assert_not_called() @@ -103,3 +101,41 @@ def test_slack_normalizes_slash_command_names(): assert Slack._normalize_slack_command("CLEAR_CACHE") == "/clear_cache" assert Slack._normalize_slack_command("/中文") == "" assert Slack._normalize_slack_command("/" + "a" * 32) == "" + + +def test_slack_plain_text_interaction_buttons_render_and_clear_on_edit(): + """纯文本交互必须显示按钮，后续无按钮编辑应清除旧操作区。""" + client = Slack.__new__(Slack) + client._client = Mock() + client._client.chat_postMessage.return_value = {"ok": True, "ts": "1", "channel": "C1"} + client._client.chat_update.return_value = {"ok": True, "ts": "1", "channel": "C1"} + buttons = [[{"text": "确认升级", "callback_data": "update:req:download"}]] + + assert ( + client.send_msg( + title="发现更新", + text="v3.0.0 -> v3.1.0", + userid="C1", + buttons=buttons, + )[0] + is True + ) + + posted = client._client.chat_postMessage.call_args.kwargs + assert posted["text"] == "发现更新\nv3.0.0 -> v3.1.0" + assert [block["type"] for block in posted["blocks"]] == ["section", "actions"] + assert posted["blocks"][1]["elements"][0]["value"] == "update:req:download" + + assert ( + client.send_msg( + title="正在下载", + text="50%", + original_message_id="1", + original_chat_id="C1", + )[0] + is True + ) + + updated = client._client.chat_update.call_args.kwargs + assert updated["text"] == "正在下载\n50%" + assert updated["blocks"] == [] diff --git a/tests/test_system_update_interaction.py b/tests/test_system_update_interaction.py new file mode 100644 index 000000000..c8778ba21 --- /dev/null +++ b/tests/test_system_update_interaction.py @@ -0,0 +1,429 @@ +"""通知渠道主程序升级交互的状态机、进度编辑和路由测试。""" + +from __future__ import annotations + +import asyncio +from collections.abc import Coroutine +from typing import Any +from unittest.mock import Mock, patch + +import pytest + +from app.application.messaging import update as update_module +from app.application.messaging.interaction import InteractionContext +from app.application.messaging.router import has_pending_interaction +from app.application.messaging.update import ( + SystemUpdateInteractionHandler, + update_interaction_manager, +) +from app.application.system import SystemOperationResult +from app.chain.message import MessageChain +from app.chain.system import SystemChain +from app.schemas.message import Message +from app.schemas.system import SystemUpdateItemStatus, SystemUpdateStatus +from app.schemas.types import NotificationChannel + + +class _Messenger: + """记录交互发送与编辑调用的内存消息网关。""" + + def __init__(self, *, edit_success: bool = True) -> None: + """初始化消息记录并配置编辑调用结果。""" + self.messages: list[Message] = [] + self.edits: list[dict[str, Any]] = [] + self.edit_success = edit_success + + def post_message(self, message: Message) -> None: + """记录一条新发送的消息。""" + self.messages.append(message) + + def edit_message(self, **kwargs: Any) -> bool: + """记录一次原消息编辑并返回可控结果。""" + self.edits.append(kwargs) + return self.edit_success + + +class _Actions: + """提供可排队状态和结果的系统更新应用用例替身。""" + + def __init__(self, check_status: SystemUpdateStatus) -> None: + """使用初始检查状态构造更新用例替身。""" + self.check_status = check_status + self.current_status = check_status + self.monitor_statuses: list[SystemUpdateStatus] = [] + self.download_result = SystemOperationResult(True, data=check_status) + self.install_result = SystemOperationResult(True, "restarting") + self.download_calls: list[str] = [] + self.install_calls: list[str] = [] + + def check_update(self) -> SystemUpdateStatus: + """返回配置的检查结果。""" + self.current_status = self.check_status + return self.check_status + + def update_status(self) -> SystemUpdateStatus: + """按顺序返回监视状态，耗尽后保留最后状态。""" + if self.monitor_statuses: + self.current_status = self.monitor_statuses.pop(0) + return self.current_status + + def download_update(self, target: str = "application") -> SystemOperationResult: + """记录下载目标并返回配置结果。""" + self.download_calls.append(target) + if isinstance(self.download_result.data, SystemUpdateStatus): + self.current_status = self.download_result.data + return self.download_result + + def install_update(self, target: str = "application") -> SystemOperationResult: + """记录安装目标并返回配置结果。""" + self.install_calls.append(target) + return self.install_result + + +def _status( + state: str, + *, + progress: int = 0, + downloaded_bytes: int = 0, + total_bytes: int = 0, + error: str | None = None, +) -> SystemUpdateStatus: + """构造只包含主程序明细的聚合更新快照。""" + version = "v3.1.0" if state != "idle" or error else None + item = SystemUpdateItemStatus( + type="application", + state=state, + current_version="v3.0.0", + version=version, + frontend_version="v3.1.0" if version else None, + release_name="MoviePilot v3.1.0" if version else None, + release_notes="修复升级流程并更新前端资源" if version else None, + downloaded_bytes=downloaded_bytes, + total_bytes=total_bytes, + progress=progress, + error=error, + can_update=state in {"available", "failed"}, + can_install=state == "ready", + ) + return SystemUpdateStatus( + state=state, + current_version="v3.0.0", + version=version, + frontend_version=item.frontend_version, + downloaded_bytes=downloaded_bytes, + total_bytes=total_bytes, + progress=progress, + error=error, + can_update=item.can_update, + can_install=item.can_install, + updates=[item], + ) + + +@pytest.fixture(autouse=True) +def _reset_update_interactions() -> None: + """隔离全局更新会话和活动监视请求。""" + update_interaction_manager.clear() + with update_module._monitor_lock: + update_module._monitored_requests.clear() + yield + update_interaction_manager.clear() + with update_module._monitor_lock: + update_module._monitored_requests.clear() + + +def _handler( + messenger: _Messenger, + actions: _Actions, + submitted: list[Coroutine[Any, Any, None]], + *, + mark_restart: Mock | None = None, + clear_restart_marker: Mock | None = None, +) -> SystemUpdateInteractionHandler: + """构造零等待且可观察后台协程的更新交互控制器。""" + return SystemUpdateInteractionHandler( + messenger=messenger, + actions=actions, + submit_monitor=submitted.append, + mark_restart=mark_restart or Mock(), + clear_restart_marker=clear_restart_marker or Mock(), + poll_interval_seconds=0, + ) + + +def test_update_command_prompts_for_download_when_release_is_available() -> None: + """检测到新版本后应显示版本信息和确认升级按钮。""" + messenger = _Messenger() + actions = _Actions(_status("available")) + submitted: list[Coroutine[Any, Any, None]] = [] + handler = _handler(messenger, actions, submitted) + + handler.remote_update( + channel=NotificationChannel.Telegram, + userid="10001", + source="telegram-main", + ) + + request = update_interaction_manager.get_by_user("10001") + assert request is not None + assert request.awaiting_input == "download" + assert messenger.messages[-1].title == "发现 MoviePilot 主程序更新" + assert "当前版本：v3.0.0" in messenger.messages[-1].text + assert messenger.messages[-1].buttons == [ + [ + { + "text": "确认升级", + "callback_data": f"update:{request.request_id}:download", + }, + {"text": "稍后", "callback_data": f"update:{request.request_id}:close"}, + ] + ] + assert submitted == [] + + +def test_download_callback_edits_same_message_until_restart_confirmation() -> None: + """下载进度应持续编辑回调原消息，完成后在同一消息显示重启按钮。""" + messenger = _Messenger() + actions = _Actions(_status("available")) + actions.download_result = SystemOperationResult( + True, + data=_status( + "downloading", + progress=10, + downloaded_bytes=10 * 1024 * 1024, + total_bytes=100 * 1024 * 1024, + ), + ) + actions.monitor_statuses = [ + _status( + "downloading", + progress=45, + downloaded_bytes=45 * 1024 * 1024, + total_bytes=100 * 1024 * 1024, + ), + _status( + "ready", + progress=100, + downloaded_bytes=100 * 1024 * 1024, + total_bytes=100 * 1024 * 1024, + ), + ] + submitted: list[Coroutine[Any, Any, None]] = [] + handler = _handler(messenger, actions, submitted) + request = update_interaction_manager.create_or_replace( + user_id="10001", + command="/update", + channel=NotificationChannel.Telegram, + source="telegram-main", + username="tester", + ) + + handled = handler.handle_callback_interaction( + callback_data=f"update:{request.request_id}:download", + channel=NotificationChannel.Telegram, + source="telegram-main", + userid="10001", + username="tester", + original_message_id="message-1", + original_chat_id="chat-1", + ) + + assert handled is True + assert actions.download_calls == ["application"] + assert len(submitted) == 1 + asyncio.run(submitted.pop()) + + assert [edit["message_id"] for edit in messenger.edits] == [ + "message-1", + "message-1", + "message-1", + ] + assert "[=.........] 10%" in messenger.edits[0]["text"] + assert "[====......] 45%" in messenger.edits[1]["text"] + assert messenger.edits[-1]["title"] == "MoviePilot 更新包已准备完成" + assert messenger.edits[-1]["buttons"][0][0]["text"] == "确认重启" + pending = update_interaction_manager.get_by_user("10001") + assert pending is not None + assert pending.awaiting_input == "install" + + +def test_text_confirmation_uses_milestone_messages_without_editing_support() -> None: + """无编辑能力的渠道仍应通过文本确认并按进度里程碑继续流程。""" + messenger = _Messenger() + actions = _Actions(_status("available")) + actions.download_result = SystemOperationResult( + True, + data=_status("downloading", progress=0, total_bytes=100 * 1024 * 1024), + ) + actions.monitor_statuses = [ + _status( + "downloading", + progress=51, + downloaded_bytes=51 * 1024 * 1024, + total_bytes=100 * 1024 * 1024, + ), + _status( + "ready", + progress=100, + downloaded_bytes=100 * 1024 * 1024, + total_bytes=100 * 1024 * 1024, + ), + ] + submitted: list[Coroutine[Any, Any, None]] = [] + handler = _handler(messenger, actions, submitted) + handler.remote_update( + channel=NotificationChannel.Wechat, + userid="wx-user", + source="wechat-main", + ) + + assert "回复“确认升级”" in messenger.messages[-1].text + assert ( + handler.handle_text_interaction( + channel=NotificationChannel.Wechat, + source="wechat-main", + userid="wx-user", + username="tester", + text="确认升级", + ) + is True + ) + asyncio.run(submitted.pop()) + + assert messenger.edits == [] + assert any("51%" in str(message.text) for message in messenger.messages) + assert messenger.messages[-1].title == "MoviePilot 更新包已准备完成" + assert "回复“确认重启”" in messenger.messages[-1].text + + +@pytest.mark.parametrize("install_success", [True, False]) +def test_restart_confirmation_marks_receipt_and_recovers_failed_install( + install_success: bool, +) -> None: + """确认重启应先更新消息，成功结束会话，失败则清理回执并恢复按钮。""" + messenger = _Messenger() + ready = _status( + "ready", + progress=100, + downloaded_bytes=100 * 1024 * 1024, + total_bytes=100 * 1024 * 1024, + ) + actions = _Actions(ready) + actions.current_status = ready + actions.install_result = SystemOperationResult( + install_success, + "restarting" if install_success else "restart failed", + ) + mark_restart = Mock() + clear_restart_marker = Mock() + handler = _handler( + messenger, + actions, + [], + mark_restart=mark_restart, + clear_restart_marker=clear_restart_marker, + ) + request = update_interaction_manager.create_or_replace( + user_id="10001", + command="/update", + channel=NotificationChannel.Telegram, + source="telegram-main", + username="tester", + ) + + assert ( + handler.handle_callback_interaction( + callback_data=f"update:{request.request_id}:install", + channel=NotificationChannel.Telegram, + source="telegram-main", + userid="10001", + username="tester", + original_message_id="message-1", + original_chat_id="chat-1", + ) + is True + ) + + assert messenger.edits[0]["title"] == "正在重启并安装 MoviePilot 更新" + mark_restart.assert_called_once_with( + NotificationChannel.Telegram, + "10001", + "telegram-main", + ) + assert actions.install_calls == ["application"] + if install_success: + clear_restart_marker.assert_not_called() + assert update_interaction_manager.get_by_user("10001") is None + else: + clear_restart_marker.assert_called_once_with() + assert messenger.edits[-1]["title"] == "MoviePilot 升级操作失败" + assert messenger.edits[-1]["buttons"][0][0]["text"] == "确认重启" + assert update_interaction_manager.get_by_user("10001") is not None + + +def test_update_session_and_callback_are_registered_in_message_router(monkeypatch) -> None: + """统一消息路由应识别更新会话文本和 update 回调前缀。""" + request = update_interaction_manager.create_or_replace( + user_id="10001", + command="/update", + channel=NotificationChannel.Telegram, + source="telegram-main", + username="tester", + ) + assert has_pending_interaction("10001") is True + callback = Mock(return_value=True) + monkeypatch.setattr( + SystemChain, + "handle_update_callback_interaction", + callback, + ) + context = InteractionContext( + channel=NotificationChannel.Telegram, + source="telegram-main", + user_id="10001", + username="tester", + original_message_id="message-1", + original_chat_id="chat-1", + ) + + result = ( + MessageChain() + ._interaction_router() + .dispatch_callback( + context, + f"update:{request.request_id}:refresh", + ) + ) + + assert result.handled is True + callback.assert_called_once() + + +def test_restart_finish_reports_actual_versions_for_update_receipt() -> None: + """升级重启完成通知应报告当前运行版本而非旧版远端查询。""" + chain = SystemChain() + with ( + patch.object( + chain, + "load_cache", + side_effect=[ + None, + { + "channel": NotificationChannel.Telegram.value, + "userid": "10001", + "source": "telegram-main", + }, + ], + ), + patch.object(chain, "post_message") as post_message, + patch.object(chain, "remove_cache") as remove_cache, + patch("app.chain.system.runtime_version.get_app_version", return_value="v3.1.0"), + patch("app.chain.system.runtime_version.get_frontend_version", return_value="v3.1.0"), + ): + chain.restart_finish() + + message = post_message.call_args.args[0] + assert message.source == "telegram-main" + assert "当前后端版本：v3.1.0" in message.title + assert "当前前端版本：v3.1.0" in message.title + remove_cache.assert_called_once_with(chain._update_restart_file)