#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import sys import io import time from textual.app import App from textual.containers import Container, ScrollableContainer from textual.widgets import Static, Input, Header, Footer from textual.reactive import reactive from services.sysmon_widget import SysMonWidget from typing import List, Dict import asyncio from datetime import datetime logger = logging.getLogger(__name__) class SystemExitGraceful(Exception): """优雅的系统退出异常""" pass class LogDisplay(Static): """日志显示组件 - 直接捕获所有日志输出""" def __init__(self): super().__init__("日志显示区域 - 等待日志输入...") self.log_lines: List[str] = [] self.max_lines = 500 self.auto_scroll_enabled = True # 启用自动滚动 # 保存原始的logging处理器和格式器 self.original_handlers = [] self.original_formatters = {} logger.debug("LogDisplay初始化完成") def start_capture(self): """开始捕获所有日志输出""" try: # 获取根日志记录器 root_logger = logging.getLogger() # 保存原始处理器和它们的格式器 self.original_handlers = root_logger.handlers.copy() for handler in self.original_handlers: self.original_formatters[handler] = handler.formatter # 清除所有现有处理器 for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) # 添加我们的自定义处理器 custom_handler = self.TUILogHandler(self) custom_handler.setLevel(logging.DEBUG) # 捕获所有级别的日志 # 强制使用包含彩色级别的格式器 formatter = self.ColoredFormatter( '%(asctime)s %(levelname_color)s %(name)s: %(message)s', datefmt='%H:%M:%S' ) custom_handler.setFormatter(formatter) root_logger.addHandler(custom_handler) # 同时重定向stdout和stderr作为备份 self.original_stdout = sys.stdout self.original_stderr = sys.stderr sys.stdout = self.TUIOutput(self) sys.stderr = self.TUIOutput(self, is_error=True) print("✅ TUI日志捕获已启动 - 捕获所有日志输出") except Exception as e: logger.error(f"启动日志捕获失败: {e}") def stop_capture(self): """停止捕获输出""" try: # 恢复logging处理器 root_logger = logging.getLogger() # 移除我们的处理器 for handler in root_logger.handlers[:]: if hasattr(handler, 'log_display'): root_logger.removeHandler(handler) # 恢复原始处理器和格式器 for handler in self.original_handlers: # 恢复格式器 if handler in self.original_formatters: handler.setFormatter(self.original_formatters[handler]) root_logger.addHandler(handler) # 恢复stdout和stderr sys.stdout = self.original_stdout sys.stderr = self.original_stderr print("🛑 TUI日志捕获已停止") except Exception as e: logger.error(f"停止日志捕获失败: {e}") def add_log_line(self, line: str): """添加日志行到TUI显示""" try: # 添加到缓冲区 self.log_lines.append(line) if len(self.log_lines) > self.max_lines: self.log_lines.pop(0) # 更新显示 display_content = "\n".join(self.log_lines) self.update(display_content) # 自动滚动到底部 if self.auto_scroll_enabled: self.scroll_to_bottom() except Exception as e: # 如果TUI更新失败，回退到原始输出 if hasattr(self, 'original_stdout'): self.original_stdout.write(f"TUI日志显示错误: {e}\n") def scroll_to_bottom(self): """滚动到底部""" try: # 获取父容器（ScrollableContainer） parent = self.parent if parent and hasattr(parent, 'scroll_end'): parent.scroll_end() except Exception as e: # 忽略滚动错误，不影响主要功能 pass def toggle_auto_scroll(self, enabled: bool = None): """切换自动滚动状态""" if enabled is None: self.auto_scroll_enabled = not self.auto_scroll_enabled else: self.auto_scroll_enabled = enabled logger.debug(f"日志自动滚动: {'启用' if self.auto_scroll_enabled else '禁用'}") return self.auto_scroll_enabled class ColoredFormatter(logging.Formatter): """带颜色的日志格式器 - 增强版""" # ANSI颜色代码 COLORS = { 'DEBUG': '\033[36m', # 青色 - DEBUG信息 'INFO': '\033[32m', # 绿色 - 正常信息 'WARNING': '\033[33m', # 黄色 - 警告信息 'ERROR': '\033[31m', # 红色 - 错误信息 'CRITICAL': '\033[35m', # 紫色 - 严重错误 'RESET': '\033[0m' # 重置颜色 } # 级别显示宽度 LEVEL_WIDTH = 8 def format(self, record): """格式化日志记录，为级别添加颜色""" try: # 为级别添加颜色和固定宽度 levelname = record.levelname if levelname in self.COLORS: # 添加颜色并保持固定宽度 colored_level = f"{self.COLORS[levelname]}[{levelname:<{self.LEVEL_WIDTH}}]{self.COLORS['RESET']}" record.levelname_color = colored_level else: record.levelname_color = f"[{levelname:<{self.LEVEL_WIDTH}}]" # 调用父类格式化方法 formatted_message = super().format(record) return formatted_message except Exception: # 如果格式化失败，返回简单格式 return f"{record.asctime} [{record.levelname}] {record.name}: {record.getMessage()}" class TUILogHandler(logging.Handler): """自定义logging处理器，同时输出到终端和TUI""" def __init__(self, log_display): super().__init__() self.log_display = log_display def emit(self, record): """处理日志记录""" try: # 格式化日志记录（使用我们的格式器） formatted_message = self.format(record) # 输出到原始终端（通过原始处理器，但使用我们的格式器） for original_handler in self.log_display.original_handlers: if original_handler.level <= record.levelno: # 临时使用我们的格式器来确保级别显示一致 original_handler.setFormatter(self.formatter) original_handler.emit(record) # 恢复原始格式器 original_formatter = self.log_display.original_formatters.get(original_handler) if original_formatter: original_handler.setFormatter(original_formatter) # 添加到TUI显示 self.log_display.add_log_line(formatted_message) except Exception as e: # 如果处理失败，使用简单格式 try: simple_message = f"{datetime.now().strftime('%H:%M:%S')} [{record.levelname:8}] {record.name}: {record.getMessage()}" self.log_display.add_log_line(simple_message) except: pass class TUIOutput(io.TextIOBase): """自定义输出流，捕获print等输出""" def __init__(self, log_display, is_error=False): self.log_display = log_display self.is_error = is_error self.original_stream = sys.stderr if is_error else sys.stdout # 颜色定义 self.COLORS = { 'INFO': '\033[32m', # 绿色 'ERROR': '\033[31m', # 红色 'RESET': '\033[0m' # 重置颜色 } def write(self, text): """写入文本""" try: # 写入到原始终端 self.original_stream.write(text) self.original_stream.flush() # 如果文本不是空的，添加到TUI if text.strip(): # 添加简单的时间戳和级别 timestamp = datetime.now().strftime('%H:%M:%S') level = "ERROR" if self.is_error else "INFO" # 添加颜色 if level in self.COLORS: colored_level = f"{self.COLORS[level]}[{level}]{self.COLORS['RESET']}" else: colored_level = f"[{level}]" # 分割多行文本 lines = text.split('\n') for line in lines: if line.strip(): # 忽略空行 log_line = f"{timestamp} {colored_level} {line.strip()}" self.log_display.add_log_line(log_line) return len(text) except Exception: # 如果TUI处理失败，只输出到终端 self.original_stream.write(text) self.original_stream.flush() return len(text) def flush(self): """刷新缓冲区""" self.original_stream.flush() def close(self): """关闭流""" pass class MessageDisplay(Static): """消息显示组件 - 增强版""" def __init__(self): super().__init__("消息区域") self.current_messages: List[Dict] = [] self.max_messages = 200 # 更大的消息容量 self.auto_scroll_enabled = True # 启用自动滚动 # ANSI颜色代码 self.COLORS = { 'INFO': '\033[37m', # 黑底白字 'DEBUG': '\033[36m', # 青色 - DEBUG信息 'SUCCESS': '\033[32m', # 绿色 - 正常信息 'WARNING': '\033[33m', # 黄色 - 警告信息 'ERROR': '\033[31m', # 红色 - 错误信息 'COMMAND': '\033[40;37m', # 灰底白字 - 命令信息 'RESET': '\033[0m' # 重置颜色 } self.message_types = { 'info': {'icon': f"{self.COLORS['INFO']}[INFO ]{self.COLORS['RESET']}", 'color': 'white'}, 'success': {'icon': f"{self.COLORS['SUCCESS']}[SUCCESS]{self.COLORS['RESET']}", 'color': 'green'}, 'error': {'icon': f"{self.COLORS['ERROR']}[ERROR ]{self.COLORS['RESET']}", 'color': 'red'}, 'warning': {'icon': f"{self.COLORS['WARNING']}[WARNING]{self.COLORS['RESET']}", 'color': 'yellow'}, 'debug': {'icon': f"{self.COLORS['DEBUG']}[DEBUG ]{self.COLORS['RESET']}", 'color': 'cyan'}, 'command': {'icon': f"{self.COLORS['COMMAND']}[COMMAND]{self.COLORS['RESET']}", 'color': 'meow'} } logger.debug("MessageDisplay初始化完成") def add_message(self, message: str, msg_type: str = "info", persistent: bool = False): """添加消息 - 支持多行消息""" try: # 分割多行消息为单独的消息 lines = message.strip().split('\n') for line in lines: if line.strip(): # 忽略空行 message_data = { "text": line.strip(), "type": msg_type, "persistent": persistent, "timestamp": asyncio.get_event_loop().time(), "display_time": datetime.now().strftime('%H:%M:%S') } self.current_messages.append(message_data) # 智能消息管理 self._manage_messages() self._update_display() except Exception as e: logger.error(f"添加消息时出错: {str(e)}") def reset_display(self): """重置显示状态""" try: # 清空所有消息 self.current_messages.clear() # 更新显示 self.update("消息区域已重置") # 强制刷新 self.refresh() except Exception as e: logger.error(f"重置消息显示时出错: {str(e)}") def _manage_messages(self): """智能管理消息数量""" try: # 计算非持久化消息的数量 non_persistent_messages = [msg for msg in self.current_messages if not msg['persistent']] if len(non_persistent_messages) > self.max_messages: # 移除最旧的非持久化消息 for i, msg in enumerate(self.current_messages): if not msg['persistent']: self.current_messages.pop(i) break except Exception as e: logger.error(f"管理消息时出错: {str(e)}") def _update_display(self): """更新显示 - 带时间戳的格式化消息""" try: if not self.current_messages: display_text = "📭 暂无消息" else: display_text = [] for msg in self.current_messages: # 获取消息类型配置 msg_config = self.message_types.get(msg['type'], self.message_types['info']) icon = msg_config['icon'] # 构建显示行 persistent_mark = "🔒 " if msg['persistent'] else "" time_stamp = f"[{msg['display_time']}] " if len(self.current_messages) > 1 else "" display_line = f"{time_stamp}{persistent_mark}{icon} {msg['text']}" display_text.append(display_line) display_text = "\n".join(display_text) self.update(display_text) # 自动滚动到底部 if self.auto_scroll_enabled: self.scroll_to_bottom() except Exception as e: logger.error(f"更新消息显示时出错: {str(e)}") def scroll_to_bottom(self): """滚动到底部""" try: # 获取父容器（ScrollableContainer） parent = self.parent if parent and hasattr(parent, 'scroll_end'): parent.scroll_end() except Exception as e: # 忽略滚动错误，不影响主要功能 pass def toggle_auto_scroll(self, enabled: bool = None): """切换自动滚动状态""" if enabled is None: self.auto_scroll_enabled = not self.auto_scroll_enabled else: self.auto_scroll_enabled = enabled logger.debug(f"消息自动滚动: {'启用' if self.auto_scroll_enabled else '禁用'}") return self.auto_scroll_enabled class TUIFramework(App): """TUI框架应用""" def __init__(self, config, log_service, command_service): super().__init__() self.config = config self.log_service = log_service self.command_service = command_service self.log_display = LogDisplay() self.message_display = MessageDisplay() self.command_input = None self.CSS = self._generate_css() def _generate_css(self): """根据配置动态生成CSS - 增强版""" try: tui_config = self.config.get('tui', {}) layout_config = tui_config.get('layout', {}) styles_config = tui_config.get('styles', {}) # 获取布局配置，使用默认值 grid_rows = layout_config.get('grid_rows', '7fr 2fr 1fr') # 获取样式配置，使用默认值 log_area_style = styles_config.get('log_area', 'border: solid green; overflow-y: auto;') message_area_style = styles_config.get('message_area', 'border: solid yellow; overflow-y: auto;') input_area_style = styles_config.get('input_area', 'border: solid red;') css = f""" Screen {{ layout: grid; grid-size: 1 3; grid-rows: {grid_rows}; }} #log-area {{ {log_area_style} overflow-y: auto; scrollbar-size: 1 1; }} #message-area {{ {message_area_style} overflow-y: auto; scrollbar-size: 1 1; }} #input-area {{ {input_area_style} }} /* 自定义滚动条样式 */ ScrollableContainer {{ scrollbar-color: #666 #222; scrollbar-color-hover: #888 #333; overflow-y: auto; }} /* 确保内容正确换行 */ Static {{ width: 100%; content-align: left middle; overflow-y: auto; }} """ logger.debug(f"生成的TUI CSS:\n{css}") return css except Exception as e: logger.error(f"生成TUI CSS时出错: {str(e)}", exc_info=True) # 返回默认CSS作为回退 return """ Screen { layout: grid; grid-size: 1 3; grid-rows: 7fr 2fr 1fr; } #log-area { border: solid green; overflow-y: auto; scrollbar-size: 1 1; } #message-area { border: solid yellow; overflow-y: auto; scrollbar-size: 1 1; } #input-area { border: solid red; } /* 自定义滚动条样式 */ ScrollableContainer { scrollbar-color: #666 #222; scrollbar-color-hover: #888 #333; overflow-y: auto; } /* 确保内容正确换行 */ Static { width: 100%; content-align: left middle; overflow-y: auto; } """ def compose(self): """组合界面""" yield Header() yield ScrollableContainer( self.log_display, id="log-area" ) yield ScrollableContainer( self.message_display, id="message-area" ) self.command_input = Input(placeholder="输入指令...", id="command-input") yield Container( self.command_input, id="input-area" ) yield Footer() async def on_mount(self): """挂载完成事件""" try: # 开始捕获所有输出 self.log_display.start_capture() # 设置输入框焦点 if self.command_input: self.command_input.focus() # 显示欢迎消息 self.show_message("🐱 SenSu TUI 已就绪！输入 'help' 查看命令\n", "info") print("✅ TUI已启动，开始捕获所有输出") except Exception as e: print(f"❌ TUI挂载时出错: {str(e)}") async def on_input_submitted(self, event): """输入提交事件""" try: if hasattr(event, 'input') and event.input.id == "command-input": command = event.value event.input.value = "" # 清空输入框 if command.strip(): print(f"执行命令: {command}") # 在消息区域显示正在处理 self.show_message(f"执行命令: {command}", "command") # 发送到指令服务处理 result = await self.command_service.process_command(command, "tui") # 显示命令结果 if result: self.show_message(f"结果: {result}", "success") else: self.show_message("命令执行完成", "success") except Exception as e: print(f"❌ 指令处理错误: {str(e)}") self.show_message(f"指令处理错误: {str(e)}", "error") def show_message(self, message: str, msg_type: str = "info", persistent: bool = False): """显示消息""" try: self.message_display.add_message(message, msg_type, persistent) except Exception as e: print(f"❌ 显示TUI消息时出错: {str(e)}") def clear_messages(self, clear_persistent: bool = False): """清空消息区域""" try: self.message_display.clear_messages(clear_persistent) except Exception as e: print(f"❌ 清空消息时出错: {str(e)}") async def action_quit(self): """重写退出动作 - 最佳方案：优雅关闭""" try: logger.info("🐱 TUI接收到退出信号，开始关闭流程") # 显示关闭消息 self.show_message("🐱 正在关闭框架...", "info", persistent=True) # 停止捕获输出 self.log_display.stop_capture() # 使用异步任务来优雅关闭，避免阻塞 asyncio.create_task(self._async_graceful_shutdown()) except Exception as e: logger.error(f"TUI退出处理时出错: {str(e)}") # 紧急退出 import os os._exit(0) async def _async_graceful_shutdown(self): """异步优雅关闭""" try: # 给一点时间显示消息 self.show_message("🐱 3...", "info", persistent=True) await asyncio.sleep(1) self.show_message("🐱 2..", "info", persistent=True) await asyncio.sleep(1) self.show_message("🐱 1.", "info", persistent=True) await asyncio.sleep(1) logger.info("🐱 执行异步关闭") logger.debug("使用事件循环停止") # 获取当前事件循环 loop = asyncio.get_event_loop() # 停止所有运行中的任务（除了当前任务） tasks = [t for t in asyncio.all_tasks(loop) if t is not asyncio.current_task()] if tasks: logger.debug(f"取消 {len(tasks)} 个运行中的任务") for task in tasks: task.cancel() # 等待任务取消完成 await asyncio.gather(*tasks, return_exceptions=True) # 停止事件循环 loop.stop() logger.info("🐱 事件循环已停止，框架关闭完成") except Exception as e: logger.error(f"异步关闭失败: {str(e)}") # 最后的手段 import os os._exit(0) def shutdown(self): """关闭TUI""" try: # 停止捕获输出 self.log_display.stop_capture() self.action_quit() self.exit() print("🛑 TUI已关闭") except Exception as e: print(f"❌ 关闭TUI时出错: {str(e)}") def _start_plugin_status_refresh(self): async def refresh(): while True: try: ps = self.service_manager.get_service("plugin") if ps and hasattr(ps, "plugin_status"): lines = [] for name, status in ps.plugin_status.items(): icon = {"running":"🟢","error":"🔴","loading":"🟡","unloaded":"⚫"}.get(str(status),"⚪") lines.append(f"{icon} {name}: {status}") widget = self.tui_app.query_one(SysMonWidget) widget.update_plugins("\n".join(lines) if lines else "无已加载插件") except: pass await asyncio.sleep(3) import asyncio asyncio.create_task(refresh()) def _get_suggester(self): from textual.suggester import Suggester class CmdSuggester(Suggester): async def get_suggestion(self, value): try: cs = self.app.query_one("#input-area").app names = ["help","status","history","testlog","netdiag", "permissions","pmpending","pmhelp","echo","plugin_status", "scroll","autoscroll","create-plugin","exit","quit"] if not value: return None for n in names: if n.startswith(value) and n != value: return n except: pass return None return CmdSuggester() class TuiService: """TUI服务""" def __init__(self, config: Dict, log_service, command_service): self.config = config self.log_service = log_service self.command_service = command_service self.tui_app = None self._message_queue = asyncio.Queue() self._message_processor_task = None async def start(self): """启动TUI""" try: if not self.config.get('tui', {}).get('enabled', True): print("TUI已禁用") return print("启动TUI服务") self.tui_app = TUIFramework(self.config, self.log_service, self.command_service) # 设置动态标题 self._setup_title() # 启动消息处理任务 self._message_processor_task = asyncio.create_task(self._process_message_queue()) # 在后台运行TUI asyncio.create_task(self._run_tui()) except Exception as e: print(f"启动TUI服务时出错: {str(e)}") raise def toggle_auto_scroll(self, target: str = "all", enabled: bool = None): """切换自动滚动状态""" try: if not self.tui_app: return "❌ TUI未启动" result = [] if target in ["all", "log"]: log_state = self.tui_app.log_display.toggle_auto_scroll(enabled) result.append(f"📜 日志自动滚动: {'✅ 启用' if log_state else '❌ 禁用'}") if target in ["all", "message"]: msg_state = self.tui_app.message_display.toggle_auto_scroll(enabled) result.append(f"💬 消息自动滚动: {'✅ 启用' if msg_state else '❌ 禁用'}") return "\n".join(result) except Exception as e: return f"❌ 切换自动滚动失败: {str(e)}" def scroll_to_bottom(self, target: str = "all"): """手动滚动到底部""" try: if not self.tui_app: return "❌ TUI未启动" result = [] if target in ["all", "log"]: self.tui_app.log_display.scroll_to_bottom() result.append("📜 日志区域已滚动到底部") if target in ["all", "message"]: self.tui_app.message_display.scroll_to_bottom() result.append("💬 消息区域已滚动到底部") return "\n".join(result) except Exception as e: return f"❌ 滚动到底部失败: {str(e)}" async def _process_message_queue(self): """处理消息队列，避免消息过多导致界面卡顿""" try: while True: # 从队列中获取消息 message_data = await self._message_queue.get() if message_data is None: # 停止信号 break message, msg_type, persistent = message_data # 显示消息 if self.tui_app: self.tui_app.show_message(message, msg_type, persistent) # 小延迟避免消息过快 await asyncio.sleep(0.05) except asyncio.CancelledError: logger.debug("消息处理任务被取消") except Exception as e: logger.error(f"消息处理任务出错: {str(e)}") def _setup_title(self): """设置TUI标题""" try: framework_config = self.config.get('framework', {}) name = framework_config.get('name', 'SenSu') version = framework_config.get('version', 'Unknown') debug_mode = framework_config.get('debug', False) # 构建标题 title_parts = [f"🐱 {name} Ver.{version}"] if debug_mode: title_parts.append("[DEBUG]") self.tui_app.title = " ".join(title_parts) self.tui_app.sub_title = "Based DreamSu Framework" logger.debug(f"设置TUI标题: {self.tui_app.title}") logger.debug(f"设置TUI副标题: {self.tui_app.sub_title}") except Exception as e: logger.error(f"设置TUI标题时出错: {e}") self.tui_app.title = "🐱 SenSu - Based DreamSu Framework" # 默认标题 async def _run_tui(self): """运行TUI""" try: await self.tui_app.run_async() except Exception as e: print(f"运行TUI时出错: {str(e)}") def show_message(self, message: str, msg_type: str = "info", persistent: bool = False): """显示消息""" try: if self.tui_app: # 将消息放入队列，由后台任务处理 self._message_queue.put_nowait((message, msg_type, persistent)) except Exception as e: print(f"通过TUI服务显示消息时出错: {str(e)}") def shutdown(self): """关闭TUI服务""" try: if self.tui_app: self.tui_app.shutdown() print("TUI服务已关闭") except Exception as e: print(f"关闭TUI服务时出错: {str(e)}")