Zhiwen Assistant MCP

""" 用户交互工具模块 提供交互式配置、错误处理、用户提示等体验优化功能 """ import sys import traceback from typing import Optional, Dict, Any, List, Union, Callable from dataclasses import dataclass from enum import Enum import json import re class InteractionLevel(Enum): """交互级别""" SILENT = "silent" # 静默模式，不输出任何信息 BASIC = "basic" # 基本信息，只显示关键结果 NORMAL = "normal" # 正常模式，显示进度和基本信息 VERBOSE = "verbose" # 详细模式，显示所有信息 DEBUG = "debug" # 调试模式，显示详细信息 class MessageType(Enum): """消息类型""" INFO = "info" SUCCESS = "success" WARNING = "warning" ERROR = "error" DEBUG = "debug" @dataclass class UserMessage: """用户消息""" type: MessageType title: str content: str details: Optional[Dict[str, Any]] = None timestamp: Optional[float] = None def __str__(self) -> str: """格式化消息输出""" icons = { MessageType.INFO: "ℹ️", MessageType.SUCCESS: "✅", MessageType.WARNING: "⚠️", MessageType.ERROR: "❌", MessageType.DEBUG: "🐛" } icon = icons.get(self.type, "📝") result = f"{icon} {self.title}" if self.content: result += f": {self.content}" return result class InteractionManager: """交互管理器""" def __init__(self, level: InteractionLevel = InteractionLevel.NORMAL): self.level = level self.message_history: List[UserMessage] = [] self.error_handlers: Dict[type, Callable] = {} self.confirmation_handlers: Dict[str, Callable] = {} def set_level(self, level: InteractionLevel) -> None: """设置交互级别""" self.level = level def should_show(self, message_type: MessageType) -> bool: """判断是否应该显示某类型的消息""" level_map = { InteractionLevel.SILENT: [], InteractionLevel.BASIC: [MessageType.ERROR, MessageType.SUCCESS], InteractionLevel.NORMAL: [MessageType.ERROR, MessageType.WARNING, MessageType.SUCCESS, MessageType.INFO], InteractionLevel.VERBOSE: [MessageType.ERROR, MessageType.WARNING, MessageType.SUCCESS, MessageType.INFO, MessageType.DEBUG], InteractionLevel.DEBUG: list(MessageType) } return message_type in level_map.get(self.level, []) def message(self, message_type: MessageType, title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送消息""" user_message = UserMessage( type=message_type, title=title, content=content, details=details ) self.message_history.append(user_message) if self.should_show(message_type): print(user_message) # 如果是详细信息且交互级别为DEBUG，打印详细信息 if details and self.level == InteractionLevel.DEBUG: print("详细信息:") for key, value in details.items(): print(f" {key}: {value}") return user_message def info(self, title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送信息消息""" return self.message(MessageType.INFO, title, content, details) def success(self, title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送成功消息""" return self.message(MessageType.SUCCESS, title, content, details) def warning(self, title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送警告消息""" return self.message(MessageType.WARNING, title, content, details) def error(self, title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送错误消息""" return self.message(MessageType.ERROR, title, content, details) def debug(self, title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送调试消息""" return self.message(MessageType.DEBUG, title, content, details) def confirm(self, title: str, content: str = "", default: bool = False) -> bool: """确认对话框""" message = f"❓ {title}" if content: message += f": {content}" suffix = " [Y/n]" if default else " [y/N]" message += suffix while True: try: response = input(f"{message}: ").strip().lower() if not response: return default if response in ['y', 'yes', '是', '确定']: return True elif response in ['n', 'no', '否', '取消']: return False else: print("请输入 y/yes/是/确定 或 n/no/否/取消") except (EOFError, KeyboardInterrupt): print("\n操作已取消") return False def select(self, title: str, options: List[Union[str, Dict[str, Any]]], default_index: int = 0) -> Any: """选择对话框""" message = f"🔘 {title}" # 格式化选项 formatted_options = [] for i, option in enumerate(options): if isinstance(option, str): formatted_options.append({"value": option, "label": option, "index": i}) elif isinstance(option, dict) and "label" in option: formatted_options.append({"value": option.get("value", option["label"]), "label": option["label"], "index": i}) else: formatted_options.append({"value": str(option), "label": str(option), "index": i}) # 显示选项 print(f"{message}:") for option in formatted_options: marker = "→" if option["index"] == default_index else " " print(f" {marker} {option['index'] + 1}. {option['label']}") while True: try: response = input(f"请选择 (1-{len(formatted_options)}) [默认: {default_index + 1}]: ").strip() if not response: selected_index = default_index else: try: selected_index = int(response) - 1 except ValueError: print("请输入有效的数字") continue if 0 <= selected_index < len(formatted_options): return formatted_options[selected_index]["value"] else: print(f"请输入 1-{len(formatted_options)} 之间的数字") except (EOFError, KeyboardInterrupt): print("\n操作已取消，使用默认选项") return formatted_options[default_index]["value"] def input_text(self, title: str, default: str = "", validator: Callable[[str], bool] = None, error_message: str = "输入无效") -> str: """文本输入对话框""" message = f"✏️ {title}" if default: message += f" [默认: {default}]" while True: try: response = input(f"{message}: ").strip() if not response and default: return default if validator is None or validator(response): return response else: print(error_message) except (EOFError, KeyboardInterrupt): print("\n操作已取消") return default def register_error_handler(self, exception_type: type, handler: Callable) -> None: """注册错误处理器""" self.error_handlers[exception_type] = handler def handle_exception(self, exception: Exception, context: str = "") -> Optional[Any]: """处理异常""" exception_type = type(exception) # 调用注册的错误处理器 if exception_type in self.error_handlers: try: return self.error_handlers[exception_type](exception) except (RuntimeError, ValueError) as e: self.error("错误处理器失败", f"处理 {exception_type.__name__} 时出错: {e}") # 默认错误处理 error_details = { "exception_type": exception_type.__name__, "context": context, "traceback": traceback.format_exc() if self.level == InteractionLevel.DEBUG else None } self.error("发生错误", str(exception), error_details) return None def get_history(self, message_type: Optional[MessageType] = None, limit: int = 100) -> List[UserMessage]: """获取消息历史""" history = self.message_history if message_type: history = [msg for msg in history if msg.type == message_type] return history[-limit:] def clear_history(self) -> None: """清空消息历史""" self.message_history.clear() def export_history(self, filepath: str) -> None: """导出消息历史到JSON文件""" history_data = [] for msg in self.message_history: history_data.append({ "type": msg.type.value, "title": msg.title, "content": msg.content, "details": msg.details, "timestamp": msg.timestamp }) with open(filepath, 'w', encoding='utf-8') as f: json.dump(history_data, f, ensure_ascii=False, indent=2) class ConfigurableInteraction(InteractionManager): """可配置的交互管理器""" def __init__(self, config: Dict[str, Any] = None): default_config = { "level": InteractionLevel.NORMAL, "auto_confirm": False, "auto_select_default": False, "timeout": None, "retry_count": 3, "case_sensitive": False } if config: default_config.update(config) super().__init__(default_config["level"]) self.auto_confirm = default_config["auto_confirm"] self.auto_select_default = default_config["auto_select_default"] self.timeout = default_config["timeout"] self.retry_count = default_config["retry_count"] self.case_sensitive = default_config["case_sensitive"] def confirm(self, title: str, content: str = "", default: bool = False) -> bool: """确认对话框（支持自动确认）""" if self.auto_confirm: self.info("自动确认", f"已自动确认: {title}") return True return super().confirm(title, content, default) def select(self, title: str, options: List[Union[str, Dict[str, Any]]], default_index: int = 0) -> Any: """选择对话框（支持自动选择默认项）""" if self.auto_select_default: self.info("自动选择", f"已自动选择默认选项: {options[default_index] if default_index < len(options) else options[0]}") return options[default_index] if default_index < len(options) else options[0] return super().select(title, options, default_index) def validate_input(self, value: str, pattern: str) -> bool: """验证输入模式""" flags = re.IGNORECASE if not self.case_sensitive else 0 return bool(re.match(pattern, value, flags)) def input_text(self, title: str, default: str = "", validator: Callable[[str], bool] = None, error_message: str = "输入无效", pattern: str = None) -> str: """文本输入对话框（支持模式验证）""" if pattern: def pattern_validator(value): return self.validate_input(value, pattern) validator = validator or pattern_validator error_message = error_message or f"输入必须匹配模式: {pattern}" return super().input_text(title, default, validator, error_message) # 全局交互管理器实例 global_interaction = InteractionManager() def set_interaction_level(level: InteractionLevel) -> None: """设置全局交互级别""" global_interaction.set_level(level) def info(title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送信息消息的便捷函数""" return global_interaction.info(title, content, details) def success(title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送成功消息的便捷函数""" return global_interaction.success(title, content, details) def warning(title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送警告消息的便捷函数""" return global_interaction.warning(title, content, details) def error(title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送错误消息的便捷函数""" return global_interaction.error(title, content, details) def debug(title: str, content: str = "", details: Dict[str, Any] = None) -> UserMessage: """发送调试消息的便捷函数""" return global_interaction.debug(title, content, details) def confirm(title: str, content: str = "", default: bool = False) -> bool: """确认对话框的便捷函数""" return global_interaction.confirm(title, content, default) def select(title: str, options: List[Union[str, Dict[str, Any]]], default_index: int = 0) -> Any: """选择对话框的便捷函数""" return global_interaction.select(title, options, default_index) def input_text(title: str, default: str = "", validator: Callable[[str], bool] = None, error_message: str = "输入无效") -> str: """文本输入对话框的便捷函数""" return global_interaction.input_text(title, default, validator, error_message) def handle_exception(exception: Exception, context: str = "") -> Optional[Any]: """处理异常的便捷函数""" return global_interaction.handle_exception(exception, context) class InteractiveError(Exception): """交互式错误""" def __init__(self, title: str, content: str = "", details: Dict[str, Any] = None): self.title = title self.content = content self.details = details or {} super().__init__(f"{title}: {content}") class UserCancelledError(InteractiveError): """用户取消操作错误""" def __init__(self, operation: str = "操作"): super().__init__("用户取消", f"用户取消了{operation}") class ValidationError(InteractiveError): """验证错误""" def __init__(self, field: str, value: str, reason: str = ""): title = f"{field}验证失败" content = f"值 '{value}' 无效" if reason: content += f": {reason}" super().__init__(title, content, {"field": field, "value": value, "reason": reason})