#!/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 from textual.suggester import Suggester try: from textual.suggestion import Suggestion except ImportError: # Textual < 2.0 兼容 — 自行构造 Suggestion class Suggestion: __slots__ = ("value",) def __init__(self, value: str = ""): self.value = value from typing import List, Dict import asyncio from datetime import datetime import psutil import os 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) log_level = getattr(logging, self.log_level.upper(), logging.DEBUG) if hasattr(self, log_level) else logging.DEBUG custom_handler.setLevel(log_level) # 从配置读取日志级别 # 强制使用包含彩色级别的格式器 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 SystemMonitor(Static): """系统监控组件 — 从共享 SystemInfoCollector 取数""" def __init__(self, collector=None): super().__init__("") self._collector = collector self._timer = None def set_collector(self, c): self._collector = c def refresh(self, **kwargs): """由外部 asyncio task 驱动刷新（比 set_interval 更抗 Android 挂起）""" self._refresh_stats() try: if not self._collector: self.update("📊 系统监控: 数据源未连接") return all_data = self._collector.get_all() cpu_pct = all_data["cpu"]["percent"] mem = all_data["memory"] net_speed = all_data["net_speed"] def bar(val, w=10): filled = int(min(val, 100) / 100 * w) bar_chars = "█" * filled + "░" * (w - filled) if val < 60: color = "\033[32m" elif val < 85: color = "\033[33m" else: color = "\033[31m" return f"{color}{bar_chars}\033[0m" self.update( f" CPU {bar(cpu_pct)} {cpu_pct:5.1f}% │ " f"MEM {bar(mem['percent'])} {mem['percent']:5.1f}% " f"({mem['used_gb']:.1f}/{mem['total_gb']:.1f}G) │ " f"NET ↓{self._fmt_speed(net_speed['rx_bytes_sec'])} " f"↑{self._fmt_speed(net_speed['tx_bytes_sec'])}" ) except Exception: pass @staticmethod def _fmt_speed(bps): if bps < 1024: return f"{bps:.0f}B/s" elif bps < 1024 * 1024: return f"{bps / 1024:.0f}K/s" else: return f"{bps / 1024 / 1024:.1f}M/s" class PluginStatusPanel(Static): """插件实时状态面板 — 展示已加载插件的状态/命令数""" def __init__(self, plugin_service=None): super().__init__("🔌 插件状态: 加载中...") self._plugin_service = plugin_service self._timer = None def set_plugin_service(self, svc): self._plugin_service = svc def refresh(self, **kwargs): """由外部 asyncio task 驱动刷新""" self._refresh() def _refresh(self): try: if not self._plugin_service: self.update("🔌 插件状态: 未连接") return infos = getattr(self._plugin_service, "plugin_info", {}) if not infos: self.update("🔌 插件状态: 无已加载插件") return lines = [] for name, info in infos.items(): cmds = len(info.commands) if info.commands else 0 status_icon = "✅" if info.loaded else "❌" lines.append(f"{status_icon} {name} v{info.version} | {cmds} commands") self.update("🔌 " + " │ ".join(lines)) except Exception: pass class CommandSuggester(Suggester): """命令补全 — 根据已注册命令提供 Tab 补全建议""" def __init__(self, command_service=None): self._cmd_svc = command_service def set_command_service(self, svc): self._cmd_svc = svc async def get_suggestion(self, value: str) -> Suggestion | None: """根据当前输入返回匹配的命令建议""" if not value or not self._cmd_svc: return None value_lower = value.lower().strip() commands = getattr(self._cmd_svc, "commands", {}) # 优先前缀匹配 candidates = sorted( [n for n in commands if n.startswith(value_lower)], key=len, ) if not candidates: candidates = sorted( [n for n in commands if value_lower in n], key=len, ) if candidates: cmd = candidates[0] return Suggestion(cmd[len(value_lower):]) return None class TUIFramework(App): """TUI框架应用""" def __init__(self, config, log_service, command_service, plugin_service=None): super().__init__() self.config = config self.log_service = log_service self.command_service = command_service self._plugin_service = plugin_service self.log_display = LogDisplay() self.message_display = MessageDisplay() self.system_monitor = SystemMonitor() self.plugin_panel = PluginStatusPanel(plugin_service) self.title_bar = Static("🐱 SenSu", id="title-bar") self.command_input = None self.CSS = self._generate_css() def _generate_css(self): """动态生成 CSS — grid 布局：空行 + 标题 + 状态栏 + 主内容 + 底部输入""" return """ Screen { layout: grid; grid-size: 1 5; grid-rows: 1 2 auto 1fr 3; } #title-bar { padding: 0 1; background: $accent; color: $text; text-style: bold; } #monitor-area { padding: 0 1; background: $panel; color: $text; } #main-area { overflow-y: auto; scrollbar-size: 1 1; border: solid $primary; } #input-area { padding: 0 1; } ScrollableContainer { scrollbar-color: #666 #222; scrollbar-color-hover: #888 #333; } Static { width: 100%; content-align: left top; } """ def compose(self): """组合界面 — 空行 + 标题 + 状态栏 + 主显示区 + 底部输入""" yield Static("") # Row 1: blank yield self.title_bar # Row 2: title yield Container( # Row 3: monitor self.system_monitor, self.plugin_panel, id="monitor-area", ) yield ScrollableContainer( # Row 4: main (1fr) self.log_display, id="main-area", ) self.command_suggester = CommandSuggester(self.command_service) self.command_input = Input( placeholder="🐱 输入指令... Tab 补全", suggester=self.command_suggester, id="command-input", ) yield Container( # Row 5: input (3 rows) self.command_input, id="input-area", ) 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): """显示消息到主显示区（log_display）""" try: prefix = {"info": "ℹ️", "success": "✅", "error": "❌", "warning": "⚠️", "command": "▶️", "debug": "🔍"}.get(msg_type, "📝") for line in message.strip().split("\n"): if line.strip(): self.log_display.add_log_line(f"{prefix} {line.strip()}") except Exception as e: print(f"❌ 显示TUI消息时出错: {str(e)}") def clear_messages(self, clear_persistent: bool = False): """清空主显示区""" try: self.log_display.log_lines.clear() self.log_display.update("") 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)}") class TuiService: """TUI服务""" def __init__(self, config: Dict, log_service, command_service, sys_collector=None, plugin_service=None): self.config = config self.log_service = log_service self.command_service = command_service self.sys_collector = sys_collector self.plugin_service = plugin_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, plugin_service=self.plugin_service, ) # 注入系统数据采集器 self.tui_app.system_monitor.set_collector(self.sys_collector) self.tui_app.plugin_panel.set_plugin_service(self.plugin_service) # 设置动态标题 self._setup_title() # 启动消息处理任务 self._message_processor_task = asyncio.create_task(self._process_message_queue()) # 启动后台刷新任务 (asyncio.sleep 比 set_interval 更抗 Android 挂起) self._refresh_task = asyncio.create_task(self._refresh_monitors()) # 在后台运行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", "message"]: log_state = self.tui_app.log_display.toggle_auto_scroll(enabled) result.append(f"📜 自动滚动: {'✅ 启用' if log_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", "message"]: self.tui_app.log_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)}") async def _refresh_monitors(self): """后台刷新系统监控 + 插件面板 (asyncio.sleep, Android 挂起后可恢复)""" try: while True: if self.tui_app: self.tui_app.system_monitor.refresh() self.tui_app.plugin_panel.refresh() await asyncio.sleep(2) except asyncio.CancelledError: logger.debug("监控刷新任务被取消") 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 = f"🐱 {name} Ver.{version}" if debug_mode: title += " [DEBUG]" self.tui_app.title_bar.update(title) logger.debug(f"设置TUI标题: {title}") except Exception as e: logger.error(f"设置TUI标题时出错: {e}") 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 hasattr(self, '_refresh_task') and self._refresh_task: self._refresh_task.cancel() if self.tui_app: self.tui_app.shutdown() print("TUI服务已关闭") except Exception as e: print(f"关闭TUI服务时出错: {str(e)}")