diff --git a/core/migrate.py b/core/migrate.py deleted file mode 100644 index 74d0727..0000000 --- a/core/migrate.py +++ /dev/null @@ -1,135 +0,0 @@ -""" -自动迁移模块 - 从旧版单用户架构迁移到多用户隔离架构 -""" - -import os -import shutil -import json -import hashlib -from pathlib import Path -from typing import Dict, Optional -from datetime import datetime - - -def _trulymem_dir() -> Path: - return Path.home() / ".trulymem" - -def _old_config_path() -> Path: - return _trulymem_dir() / "config.json" - -def _old_db_path() -> Path: - return _trulymem_dir() / "graph_memory.db" - -def _new_global_db_path() -> Path: - return _trulymem_dir() / "trulymem.db" - -def _migrated_flag() -> Path: - return _trulymem_dir() / ".migrated" - - -def need_migration() -> bool: - """检测是否需要迁移""" - # 如果已经迁移过,不需要再迁移 - if is_migrated(): - return False - - old_config_exists = _old_config_path().exists() - old_db_exists = _old_db_path().exists() - new_db_exists = _new_global_db_path().exists() - if (old_config_exists or old_db_exists) and not new_db_exists: - return True - - return False - - -def is_migrated() -> bool: - """检查是否已完成迁移""" - return _migrated_flag().exists() - - -def _mark_migrated(): - """标记迁移完成""" - _trulymem_dir().mkdir(parents=True, exist_ok=True) - with open(_migrated_flag(), 'w') as f: - f.write(datetime.now().isoformat()) - - -def run_migration(username: str, password: str) -> Dict: - """ - 执行迁移 - - Args: - username: 新用户名 - password: 新用户密码 - - Returns: - 迁移结果字典 - """ - try: - # 1. 创建用户目录 - user_dir = _trulymem_dir() / username - user_dir.mkdir(parents=True, exist_ok=True) - new_config_path = user_dir / "config.json" - if _old_config_path().exists(): - shutil.copy2(_old_config_path(), new_config_path) - new_db_path = user_dir / f"{username}_graph.db" - if _old_db_path().exists(): - shutil.copy2(_old_db_path(), new_db_path) - - # 4. 创建全局数据库并写入 web_users 表 - from .embedded_db import EmbeddedGraphDB - - global_db = EmbeddedGraphDB(db_path=str(_new_global_db_path())) - - # 设置用户(会自动创建记录) - result = global_db.set_web_user(username, password) - if not result.get("success"): - return {"success": False, "error": f"创建用户失败: {result.get('error')}"} - - # 如果用户目录已存在,更新路径(确保正确) - cursor = global_db.conn.cursor() - config_path = str(new_config_path) - db_path = str(new_db_path) - cursor.execute(""" - UPDATE web_users - SET config_path = ?, db_path = ? - WHERE username = ? - """, (config_path, db_path, username)) - global_db.conn.commit() - - # 5. 标记迁移完成 - _mark_migrated() - - global_db.close() - - return { - "success": True, - "username": username, - "config_path": config_path, - "db_path": db_path, - "message": "迁移完成" - } - - except Exception as e: - return {"success": False, "error": str(e)} - - -def rollback_migration(): - """回滚迁移(用于失败恢复)""" - try: - # 删除全局数据库 - if _new_global_db_path().exists(): - _new_global_db_path().unlink() - if _migrated_flag().exists(): - _migrated_flag().unlink() - - return {"success": True, "message": "回滚完成"} - except Exception as e: - return {"success": False, "error": str(e)} - - -if __name__ == '__main__': - # 测试 - print("Migration module test") - print(f"Need migration: {need_migration()}") - print(f"Is migrated: {is_migrated()}") diff --git a/trulymem_entry.py b/trulymem_entry.py deleted file mode 100644 index 3eb977d..0000000 --- a/trulymem_entry.py +++ /dev/null @@ -1,87 +0,0 @@ -#!/usr/bin/env python3 -import argparse -import sys -import os -import signal -import threading -from pathlib import Path - -# 用户配置文件始终放在用户目录 -CONFIG_PATH = Path.home() / ".trulymem" / "config.json" -DB_PATH = Path.home() / ".trulymem" / "graph_memory.db" - -# 源码运行时使用项目目录,打包后使用用户目录 -if getattr(sys, 'frozen', False): - # 打包版本:创建用户目录 - CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) -else: - # 源码版本:检查项目目录是否有配置(向后兼容) - project_dir = Path(__file__).parent - project_config = project_dir / "config.json" - project_db = project_dir / "graph_memory.db" - - if project_config.exists(): - CONFIG_PATH = project_config - if project_db.exists(): - DB_PATH = project_db - -sys.path.insert(0, str(Path(__file__).parent)) -os.chdir(Path(__file__).parent) - -from core import BackendServer -from ui import GraphMemoryApp - - -def main(): - parser = argparse.ArgumentParser(description='TrulyMEM - True Human Memory') - parser.add_argument('--web', action='store_true', help='以 Web 服务模式启动(headless,不启动 TUI)') - parser.add_argument('--port', type=int, default=4096, help='Web 服务端口(默认 4096)') - args = parser.parse_args() - - if args.web: - # Web headless 模式 - from web_api import run_web_server, stop_web_server - - backend_server = BackendServer( - db_path=str(DB_PATH), - use_embedded_db=True, - config_file=str(CONFIG_PATH) - ) - backend_server.start() - - print(f"TrulyMEM Web 服务启动在 http://0.0.0.0:{args.port}") - run_web_server(port=args.port, host='0.0.0.0') - - # 阻塞主线程直到收到信号 - shutdown_event = threading.Event() - def _handle_signal(signum, frame): - print("\n收到停止信号,正在关闭...") - shutdown_event.set() - signal.signal(signal.SIGINT, _handle_signal) - signal.signal(signal.SIGTERM, _handle_signal) - - shutdown_event.wait() - stop_web_server() - backend_server.shutdown() - print("TrulyMEM Web 服务已停止") - else: - # TUI 模式(默认) - backend_server = BackendServer( - db_path=str(DB_PATH), - use_embedded_db=True, - config_file=str(CONFIG_PATH) - ) - backend_server.start() - - app = GraphMemoryApp(backend_server=backend_server, config_file=str(CONFIG_PATH)) - - try: - app.run() - except KeyboardInterrupt: - print("\n退出") - finally: - backend_server.shutdown() - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/ui/__init__.py b/ui/__init__.py deleted file mode 100644 index 5527bfa..0000000 --- a/ui/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -from .app import GraphMemoryApp -from .models.config import AppConfig - -__all__ = ["GraphMemoryApp", "AppConfig"] \ No newline at end of file diff --git a/ui/app.py b/ui/app.py deleted file mode 100644 index 84a83ea..0000000 --- a/ui/app.py +++ /dev/null @@ -1,367 +0,0 @@ -import asyncio -import sys -import threading -import signal -from pathlib import Path -from textual.app import App, ComposeResult -from textual.binding import Binding - -from core import BackendServer -from core.client import BackendClient -from .models.message import Message - - -class GraphMemoryApp(App): - CSS_PATH = [ - Path(__file__).parent / "styles" / "app.css", - Path(__file__).parent / "styles" / "messages.css", - Path(__file__).parent / "styles" / "components.css", - ] - - BINDINGS = [ - Binding("f1", "show_help", "帮助"), - Binding("f2", "toggle_sidebar", "侧边栏"), - Binding("f3", "toggle_tool_details", "工具详情"), - Binding("f5", "clear_history", "清屏"), - Binding("f6", "quit", "退出"), - ] - - def __init__(self, backend_server: BackendServer = None, config_file: str = None, **kwargs): - super().__init__(**kwargs) - self._backend_server = backend_server - self._backend_client = BackendClient(backend_server) if backend_server else None - self._api_configured = False - self._web_running = False - self.login_user = None # 当前登录用户 - self.login_user_info = None # 当前登录用户信息 - - def compose(self) -> ComposeResult: - from .widgets.left_panel import LeftPanel - from .widgets.right_panel import RightPanel - from .widgets.status_bar import StatusBar - from .models.config import AppConfig - - initial_config = AppConfig() - - if self._backend_client: - settings_result = self._backend_client.get_settings() - settings_data = settings_result.get("data", {}) - - api_config = settings_data.get("api_config", {}) - initial_config.api_key = api_config.get("api_key", "") - initial_config.base_url = api_config.get("base_url", "https://api.deepseek.com") - initial_config.model = api_config.get("model", "deepseek-chat") - - tool_limits = settings_data.get("tool_limits", {}) - initial_config.persona_update_max = tool_limits.get("persona_update_max", 1) - initial_config.task_update_max = tool_limits.get("task_update_max", 5) - initial_config.memory_query_max = tool_limits.get("memory_query_max", 20) - initial_config.memory_update_max = tool_limits.get("memory_update_max", 10) - - yield LeftPanel() - yield RightPanel(config=initial_config) - yield StatusBar() - - def on_mount(self) -> None: - from .widgets.status_bar import StatusBar - from .widgets.message_history import MessageHistory - from .login_screen import LoginScreen - from core.migrate import need_migration, is_migrated - - # 检查是否需要登录 - migrated = is_migrated() - need_login = migrated or not need_migration() - - if need_login and not self.login_user: - # 显示登录界面 - self.push_screen(LoginScreen()) - return - - # 已登录或无需登录,继续初始化 - self._init_after_login() - - def on_login_success(self, username: str, user_info: dict) -> None: - """登录成功后调用""" - self.login_user = username - self.login_user_info = user_info - # 更新 config section 的 admin 权限 - from .widgets.config_section import ConfigSection - try: - is_admin = user_info.get('role') == 'admin' - config_section = self.query_one(ConfigSection) - config_section.set_admin(is_admin) - except Exception: - pass - # 重新初始化后端 - self._init_after_login() - - def _init_after_login(self) -> None: - """登录后初始化""" - from .widgets.status_bar import StatusBar - from .widgets.message_history import MessageHistory - status_bar = self.query_one(StatusBar) - - if not self._backend_server: - history = self.query_one(MessageHistory) - error = Message(role="assistant", content="后端未初始化") - history.add_message(error) - status_bar.set_api_status(False) - return - - # 如果已登录,重新初始化后端服务器以使用用户的数据库 - if self.login_user: - from core.server import BackendServer - # 创建新的后端服务器(使用用户的数据库) - self._backend_server = BackendServer(username=self.login_user) - self._backend_client = BackendClient(self._backend_server) - self._backend_server.start() - - status = self._backend_client.get_status() - data = status.get("data", {}) - self._api_configured = data.get("config", {}).get("api_key", "") != "" - status_bar.set_api_status(self._api_configured) - - history = self.query_one(MessageHistory) - - if self._api_configured: - chat_history = self._backend_client.get_history() - if chat_history: - for msg in chat_history: - message = Message(role=msg["role"], content=msg["content"]) - history.add_message(message) - - role_label = "管理员" if self.login_user_info and self.login_user_info.get('role') == 'admin' else "用户" - welcome_msg = f"系统就绪\n用户: {self.login_user or '默认'} ({role_label})\n" - welcome_msg += f"API Key: {'已配置' if self._api_configured else '未配置'}\n\n输入消息开始对话" - welcome = Message(role="assistant", content=welcome_msg) - history.add_message(welcome) - - def _start_web_server(self, port: int = 4096) -> None: - """在当前进程通过线程启动 Web 服务(无需子进程)""" - if self._web_running: - self.notify("Web 服务已在运行", title="提示") - return - - try: - from web_api import run_web_server - run_web_server(port=port) - self._web_running = True - if self._backend_client: - self._backend_client.report_web_status(True, port) - self.notify(f"Web 服务已启动 → http://0.0.0.0:{port}", title="Web 服务") - except Exception as e: - self.notify(f"启动 Web 服务失败: {e}", severity="error") - - def _stop_web_server(self) -> None: - """停止 Web 服务线程""" - if self._web_running: - try: - from web_api import stop_web_server - stop_web_server() - except Exception: - pass - self._web_running = False - if self._backend_client: - self._backend_client.report_web_status(False, 0) - self.notify("Web 服务已停止", title="Web 服务") - - def on_unmount(self) -> None: - # 停止 Web 服务 - self._stop_web_server() - if self._backend_client: - self._backend_client.shutdown() - - def action_show_help(self) -> None: - from pathlib import Path - config_path = Path.home() / ".trulymem" / "config.json" - db_path = Path.home() / ".trulymem" / "graph_memory.db" - - help_text = ( - "F1-帮助 F2-侧边栏 F3-工具详情 F5-清屏 F6-退出\n\n" - f"配置文件: {config_path}\n" - f"数据库: {db_path}" - ) - self.notify(help_text, title="快捷键 & 配置路径", timeout=15) - - def action_toggle_sidebar(self) -> None: - from .widgets.right_panel import RightPanel - sidebar = self.query_one(RightPanel) - sidebar.toggle() - - def action_toggle_tool_details(self) -> None: - from .widgets.message_history import MessageHistory - history = self.query_one(MessageHistory) - history.toggle_latest_tool_details() - - def action_clear_history(self) -> None: - from .widgets.message_history import MessageHistory - history = self.query_one(MessageHistory) - history.clear_messages() - - def on_input_box_send_message(self, event) -> None: - if not self._backend_client: - self.notify("后端未初始化", title="错误", severity="error") - return - - if not self._api_configured: - self.notify("请先配置 API Key (按 F2 打开侧边栏)", title="提示", severity="warning") - return - - user_input = event.content - from .widgets.message_history import MessageHistory - from .widgets.status_bar import StatusBar - - history = self.query_one(MessageHistory) - status_bar = self.query_one(StatusBar) - - history.add_message(Message(role="user", content=user_input)) - history.add_message(Message(role="assistant", content="⏳ 正在处理...")) - status_bar.set_processing(True) - - asyncio.create_task(self._process(user_input)) - - def on_input_box_clear_history(self, event) -> None: - """处理清空聊天记录事件""" - if not self._backend_client: - self.notify("后端未初始化", title="错误", severity="error") - return - - self._backend_client.clear_history() - - from .widgets.message_history import MessageHistory - history = self.query_one(MessageHistory) - history.clear_messages() - - self.notify("聊天记录已清空,AI记忆保持不变", title="提示", severity="information") - - async def _process(self, user_input: str) -> None: - from .widgets.message_history import MessageHistory - from .widgets.status_bar import StatusBar - from .widgets.right_panel import RightPanel - from .models.log_entry import LogEntry - from datetime import datetime - - history = self.query_one(MessageHistory) - status_bar = self.query_one(StatusBar) - - result = await asyncio.get_event_loop().run_in_executor( - None, - lambda: self._backend_client.process_message(user_input) - ) - - if result.get("success"): - # 响应结构: {"success": True, "data": {"content": "...", "tool_calls": [...], ...}, "error": None} - data = result.get("data", {}) - content = data.get("content", "(无回复)") - history.update_latest_message(content) - - # 处理工具调用信息,更新操作日志 - tool_calls = data.get("tool_calls", []) - if tool_calls: - try: - right_panel = self.query_one(RightPanel) - operation_log = right_panel.get_operation_log() - - for tool_call in tool_calls: - entry = LogEntry( - timestamp=datetime.now(), - tool_name=tool_call.get("name", "unknown"), - arguments=tool_call.get("arguments", {}), - result=str(tool_call.get("result", "")), - duration=0.0 # 后端没有返回耗时信息 - ) - operation_log.add_log(entry) - except Exception: - pass # 忽略操作日志更新失败 - else: - error = result.get("error", "未知错误") - history.update_latest_message(f"❌ 错误: {error}") - - status_bar.set_processing(False) - - def on_config_section_config_changed(self, event) -> None: - if not self._backend_client: - self.notify("后端未初始化,无法保存配置", title="错误", severity="error") - return - - asyncio.create_task(self._update_settings_async(event.config)) - - async def _update_settings_async(self, config) -> None: - from .widgets.status_bar import StatusBar - from .widgets.config_section import ConfigSection - - status_bar = self.query_one(StatusBar) - - api_config = { - "api_key": config.api_key, - "base_url": config.base_url, - "model": getattr(config, 'model', 'deepseek-chat') - } - - tool_limits = { - "persona_update_max": config.persona_update_max, - "task_update_max": config.task_update_max, - "memory_query_max": config.memory_query_max, - "memory_update_max": config.memory_update_max, - } - - # 保存 Web 用户(如果用户名和密码都不为空) - if config.web_username and config.web_password: - try: - await asyncio.get_event_loop().run_in_executor( - None, self._backend_client.set_web_user, - config.web_username, config.web_password - ) - except Exception as e: - self.notify(f"保存 Web 用户失败: {e}", severity="warning") - - try: - result = await asyncio.get_event_loop().run_in_executor( - None, - lambda: self._backend_client.update_settings( - api_config=api_config, - tool_limits=tool_limits - ) - ) - - if result.get("success"): - self._api_configured = bool(config.api_key) - status_bar.set_api_status(self._api_configured) - - settings_result = await asyncio.get_event_loop().run_in_executor( - None, - lambda: self._backend_client.get_settings() - ) - settings_data = settings_result.get("data", {}) - - try: - config_section = self.query_one(ConfigSection) - api_cfg = settings_data.get("api_config", {}) - tool_lmts = settings_data.get("tool_limits", {}) - config_section.set_config(AppConfig( - api_key=api_cfg.get("api_key", ""), - base_url=api_cfg.get("base_url", "https://api.deepseek.com"), - model=api_cfg.get("model", "deepseek-chat"), - persona_update_max=tool_lmts.get("persona_update_max", 1), - task_update_max=tool_lmts.get("task_update_max", 5), - memory_query_max=tool_lmts.get("memory_query_max", 20), - memory_update_max=tool_lmts.get("memory_update_max", 10), - enable_web=api_cfg.get("enable_web", False), - web_port=api_cfg.get("web_port", 4096), - enable_tui=api_cfg.get("enable_tui", True), - )) - except Exception: - pass - - # 管理 Web 服务 - if config.enable_web: - self._start_web_server(config.web_port) - else: - self._stop_web_server() - - self.notify("✅ 配置已保存并生效", title="配置成功", severity="information") - else: - error = result.get("error", "未知错误") - self.notify(f"❌ 配置失败: {error}", title="配置失败", severity="error") - except Exception as e: - self.notify(f"❌ 配置异常: {str(e)}", title="配置失败", severity="error") \ No newline at end of file diff --git a/ui/handlers/__init__.py b/ui/handlers/__init__.py deleted file mode 100644 index 14b3896..0000000 --- a/ui/handlers/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Event Handlers for Graph Memory TUI""" diff --git a/ui/handlers/focus_handler.py b/ui/handlers/focus_handler.py deleted file mode 100644 index fa3c99d..0000000 --- a/ui/handlers/focus_handler.py +++ /dev/null @@ -1,64 +0,0 @@ -"""焦点管理器""" - -from textual.app import App - - -class FocusHandler: - """焦点管理器""" - - # 焦点循环顺序 - FOCUS_RING = [ - "input-textarea", # 左侧输入框 - "api-key-input", # 右侧配置区 API Key - "model-input", # 右侧配置区 Model - "base-url-input", # 右侧配置区 Base URL - "cypher-textarea", # 右侧 Cypher 查询框 - ] - - # 焦点名称映射 - FOCUS_NAMES = { - "input-textarea": "Input", - "api-key-input": "Config-API", - "model-input": "Config-Model", - "base-url-input": "Config-URL", - "cypher-textarea": "Query", - } - - def __init__(self): - self._current_index = 0 - - def next_focus(self, app: App) -> None: - """切换到下一个焦点""" - self._current_index = (self._current_index + 1) % len(self.FOCUS_RING) - widget_id = self.FOCUS_RING[self._current_index] - self._focus_widget(app, widget_id) - - def prev_focus(self, app: App) -> None: - """切换到上一个焦点""" - self._current_index = (self._current_index - 1) % len(self.FOCUS_RING) - widget_id = self.FOCUS_RING[self._current_index] - self._focus_widget(app, widget_id) - - def focus_input(self, app: App) -> None: - """聚焦到输入框""" - self._current_index = 0 - self._focus_widget(app, self.FOCUS_RING[0]) - - def focus_query(self, app: App) -> None: - """聚焦到查询框""" - self._current_index = len(self.FOCUS_RING) - 1 - self._focus_widget(app, self.FOCUS_RING[-1]) - - def get_current_focus_name(self) -> str: - """获取当前焦点名称""" - widget_id = self.FOCUS_RING[self._current_index] - return self.FOCUS_NAMES.get(widget_id, "Unknown") - - def _focus_widget(self, app: App, widget_id: str) -> None: - """聚焦到指定组件""" - try: - widget = app.query_one(f"#{widget_id}") - widget.focus() - except Exception: - # 如果找不到组件,回退到输入框 - self.focus_input(app) diff --git a/ui/handlers/key_handler.py b/ui/handlers/key_handler.py deleted file mode 100644 index 0ddc06a..0000000 --- a/ui/handlers/key_handler.py +++ /dev/null @@ -1,68 +0,0 @@ -"""快捷键处理器""" - -from textual.app import App -from textual.message import Message -from .focus_handler import FocusHandler - - -class KeyHandler: - """快捷键处理器""" - - class ShowHelp(Message): - """显示帮助事件""" - pass - - class ToggleSidebar(Message): - """切换侧边栏事件""" - pass - - class ToggleToolDetails(Message): - """切换工具详情事件""" - pass - - class FocusQuery(Message): - """聚焦查询框事件""" - pass - - class ClearHistory(Message): - """清屏事件""" - pass - - class QuitApp(Message): - """退出应用事件""" - pass - - def __init__(self, focus_handler: FocusHandler): - self._focus_handler = focus_handler - - def handle_f1(self, app: App) -> None: - """处理 F1 键 - 显示帮助""" - app.post_message(self.ShowHelp()) - - def handle_f2(self, app: App) -> None: - """处理 F2 键 - 切换侧边栏""" - app.post_message(self.ToggleSidebar()) - - def handle_f3(self, app: App) -> None: - """处理 F3 键 - 切换工具详情""" - app.post_message(self.ToggleToolDetails()) - - def handle_f4(self, app: App) -> None: - """处理 F4 键 - 聚焦查询框""" - app.post_message(self.FocusQuery()) - - def handle_f5(self, app: App) -> None: - """处理 F5 键 - 清屏""" - app.post_message(self.ClearHistory()) - - def handle_f6(self, app: App) -> None: - """处理 F6 键 - 退出""" - app.post_message(self.QuitApp()) - - def handle_tab(self, app: App) -> None: - """处理 Tab 键 - 焦点循环""" - self._focus_handler.next_focus(app) - - def handle_shift_tab(self, app: App) -> None: - """处理 Shift+Tab 键 - 反向焦点循环""" - self._focus_handler.prev_focus(app) diff --git a/ui/login_screen.py b/ui/login_screen.py deleted file mode 100644 index 3fbbd18..0000000 --- a/ui/login_screen.py +++ /dev/null @@ -1,142 +0,0 @@ -""" -TUI 登录页面 -""" - -import asyncio -from pathlib import Path -from textual.app import ComposeResult -from textual.containers import Center, Middle, Vertical -from textual.widgets import Input, Button, Static, Label -from textual.screen import Screen -from core.embedded_db import EmbeddedGraphDB -from core.migrate import need_migration, run_migration, is_migrated - - -class LoginScreen(Screen): - """登录界面""" - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self._migrating = False - self._migration_username = "" - self._migration_password = "" - - def compose(self) -> ComposeResult: - # 检测是否需要迁移 - migrating = need_migration() - migrated = is_migrated() - - if migrating and not migrated: - yield from self._compose_migration() - else: - yield from self._compose_login() - - def _compose_login(self) -> ComposeResult: - with Center(): - with Middle(): - with Vertical(id="login_container"): - yield Static("🔐 TrulyMEM 登录", id="login_title") - yield Label("用户名:") - yield Input(placeholder="请输入用户名", id="username_input") - yield Label("密码:") - yield Input(placeholder="请输入密码", password=True, id="password_input") - yield Button("登录", id="login_button", variant="primary") - yield Static("", id="login_message") - - def _compose_migration(self) -> ComposeResult: - with Center(): - with Middle(): - with Vertical(id="migration_container"): - yield Static("🔄 检测到旧版数据,需要迁移", id="migration_title") - yield Static("请设置管理员账号以完成迁移", id="migration_subtitle") - yield Label("用户名:") - yield Input(placeholder="请输入管理员用户名", id="mig_username_input") - yield Label("密码:") - yield Input(placeholder="请输入管理员密码", password=True, id="mig_password_input") - yield Button("开始迁移", id="migrate_button", variant="primary") - yield Static("", id="migration_message") - - def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == "login_button": - self._handle_login() - elif event.button.id == "migrate_button": - self._handle_migration() - - def on_input_submitted(self, event: Input.Submitted) -> None: - if event.input.id == "username_input": - self.query_one("#password_input", Input).focus() - elif event.input.id == "password_input": - self._handle_login() - elif event.input.id == "mig_username_input": - self.query_one("#mig_password_input", Input).focus() - elif event.input.id == "mig_password_input": - self._handle_migration() - - def _handle_login(self) -> None: - username = self.query_one("#username_input", Input).value.strip() - password = self.query_one("#password_input", Input).value - - if not username or not password: - self.query_one("#login_message", Static).update("❌ 用户名和密码不能为空") - return - - # 验证用户 - try: - global_db_path = Path.home() / ".trulymem" / "trulymem.db" - if not global_db_path.exists(): - self.query_one("#login_message", Static).update("❌ 全局数据库不存在,请先完成迁移") - return - - db = EmbeddedGraphDB(db_path=str(global_db_path)) - user_info = db.get_web_user(username) - - if not user_info: - self.query_one("#login_message", Static).update("❌ 用户不存在") - db.close() - return - - # 验证密码 - import hashlib - password_hash = hashlib.sha256(password.encode()).hexdigest() - - if db.verify_web_user(username, password): - db.close() - # 登录成功,通知应用 - self.app.on_login_success(username, user_info) - else: - self.query_one("#login_message", Static).update("❌ 密码错误") - db.close() - - except Exception as e: - self.query_one("#login_message", Static).update(f"❌ 登录失败: {str(e)}") - - def _handle_migration(self) -> None: - username = self.query_one("#mig_username_input", Input).value.strip() - password = self.query_one("#mig_password_input", Input).value - - if not username or not password: - self.query_one("#migration_message", Static).update("❌ 用户名和密码不能为空") - return - - self.query_one("#migration_message", Static).update("⏳ 正在迁移...") - - # 执行迁移(异步) - asyncio.create_task(self._do_migration(username, password)) - - async def _do_migration(self, username: str, password: str) -> None: - try: - loop = asyncio.get_event_loop() - result = await loop.run_in_executor( - None, run_migration, username, password - ) - - if result.get("success"): - self.query_one("#migration_message", Static).update("✅ 迁移成功!请登录") - # 重新加载界面为登录界面 - await asyncio.sleep(1) - self.app.pop_screen() - self.app.push_screen(LoginScreen()) - else: - self.query_one("#migration_message", Static).update(f"❌ 迁移失败: {result.get('error')}") - except Exception as e: - self.query_one("#migration_message", Static).update(f"❌ 迁移异常: {str(e)}") diff --git a/ui/models/__init__.py b/ui/models/__init__.py deleted file mode 100644 index f08233e..0000000 --- a/ui/models/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Data Models for Graph Memory TUI""" diff --git a/ui/models/config.py b/ui/models/config.py deleted file mode 100644 index d7700df..0000000 --- a/ui/models/config.py +++ /dev/null @@ -1,81 +0,0 @@ -"""配置数据模型""" - -import json -import os -from dataclasses import dataclass, asdict -from pathlib import Path -from typing import Optional - - -@dataclass -class AppConfig: - """应用配置""" - api_key: str = "" - model: str = "deepseek-chat" - base_url: str = "https://api.deepseek.com" - persona_update_max: int = 1 - task_update_max: int = 5 - memory_query_max: int = 20 - memory_update_max: int = 10 - web_username: str = "" - web_password: str = "" - enable_web: bool = False - web_port: int = 4096 - enable_tui: bool = True - - @classmethod - def from_env(cls, username: str = "") -> "AppConfig": - """ - 从环境变量加载配置。 - 如果指定了 username,尝试从用户的配置文件加载。 - """ - # 如果指定了用户名,尝试从用户的配置文件加载 - if username: - from pathlib import Path - user_config_path = Path.home() / ".trulymem" / username / "config.json" - if user_config_path.exists(): - return cls.from_file(user_config_path) - - return cls( - api_key=os.getenv("DEEPSEEK_API_KEY", ""), - model=os.getenv("MODEL_NAME", "deepseek-chat"), - base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"), - persona_update_max=int(os.getenv("PERSONA_UPDATE_MAX", 1)), - task_update_max=int(os.getenv("TASK_UPDATE_MAX", 5)), - memory_query_max=int(os.getenv("MEMORY_QUERY_MAX", 20)), - memory_update_max=int(os.getenv("MEMORY_UPDATE_MAX", 10)), - web_username=os.getenv("WEB_USERNAME", ""), - web_password=os.getenv("WEB_PASSWORD", ""), - enable_web=os.getenv("ENABLE_WEB", "false").lower() == "true", - web_port=int(os.getenv("WEB_PORT", 4096)), - enable_tui=os.getenv("ENABLE_TUI", "true").lower() == "true", - ) - - @classmethod - def from_file(cls, path: Path) -> "AppConfig": - if not path.exists(): - return cls() - - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - - return cls( - api_key=data.get("api_key", ""), - model=data.get("model", "deepseek-chat"), - base_url=data.get("base_url", "https://api.deepseek.com"), - persona_update_max=data.get("persona_update_max", 1), - task_update_max=data.get("task_update_max", 5), - memory_query_max=data.get("memory_query_max", 20), - memory_update_max=data.get("memory_update_max", 10), - web_username=data.get("web_username", ""), - web_password=data.get("web_password", ""), - enable_web=data.get("enable_web", False), - web_port=data.get("web_port", 4096), - enable_tui=data.get("enable_tui", True), - ) - - def save(self, path: Path) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - - with open(path, "w", encoding="utf-8") as f: - json.dump(asdict(self), f, indent=2, ensure_ascii=False) diff --git a/ui/models/log_entry.py b/ui/models/log_entry.py deleted file mode 100644 index 51c5eb3..0000000 --- a/ui/models/log_entry.py +++ /dev/null @@ -1,30 +0,0 @@ -"""日志条目数据模型""" - -from dataclasses import dataclass -from datetime import datetime -from typing import Any, Dict - - -@dataclass -class LogEntry: - """日志条目""" - timestamp: datetime - tool_name: str - arguments: Dict[str, Any] - result: str - duration: float - - @property - def args_summary(self) -> str: - """参数摘要(截断到50字符)""" - args_str = str(self.arguments) - if len(args_str) > 50: - return args_str[:50] + "..." - return args_str - - @property - def result_summary(self) -> str: - """结果摘要(截断到100字符)""" - if len(self.result) > 100: - return self.result[:100] + "..." - return self.result diff --git a/ui/models/message.py b/ui/models/message.py deleted file mode 100644 index 98bbdd4..0000000 --- a/ui/models/message.py +++ /dev/null @@ -1,33 +0,0 @@ -"""消息数据模型""" - -from dataclasses import dataclass, field -from datetime import datetime -from typing import Dict, List, Literal, Optional, Any - - -@dataclass -class ToolCall: - """工具调用""" - id: str - name: str - arguments: Dict[str, Any] - - -@dataclass -class ToolResult: - """工具执行结果""" - tool_call_id: str - name: str - arguments: Dict[str, Any] - result: str - success: bool - - -@dataclass -class Message: - """消息""" - role: Literal["user", "assistant", "system"] - content: str - timestamp: datetime = field(default_factory=datetime.now) - tool_calls: Optional[List[ToolCall]] = None - tool_results: Optional[List[ToolResult]] = None diff --git a/ui/services/__init__.py b/ui/services/__init__.py deleted file mode 100644 index 837ef17..0000000 --- a/ui/services/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Business Services for Graph Memory TUI""" diff --git a/ui/services/config_manager.py b/ui/services/config_manager.py deleted file mode 100644 index e2501b5..0000000 --- a/ui/services/config_manager.py +++ /dev/null @@ -1,46 +0,0 @@ -""" -配置管理 - 支持持久化 -""" - -import json -from pathlib import Path -from ..models.config import AppConfig - - -class ConfigManager: - """配置管理器 - 支持持久化""" - - def __init__(self, config_file: str = "config.json"): - self.config_file = Path(config_file) - - def save(self, config: AppConfig) -> None: - """保存配置到文件""" - data = { - "api_key": config.api_key, - "model": config.model, - "base_url": config.base_url - } - - with open(self.config_file, 'w', encoding='utf-8') as f: - json.dump(data, f, indent=2) - - def load(self) -> AppConfig: - """从文件加载配置""" - if not self.config_file.exists(): - return AppConfig() - - try: - with open(self.config_file, 'r', encoding='utf-8') as f: - data = json.load(f) - - return AppConfig( - api_key=data.get("api_key", ""), - model=data.get("model", "deepseek-chat"), - base_url=data.get("base_url", "https://api.deepseek.com") - ) - except Exception: - return AppConfig() - - def exists(self) -> bool: - """检查配置文件是否存在""" - return self.config_file.exists() diff --git a/ui/services/config_service.py b/ui/services/config_service.py deleted file mode 100644 index dc2a9c8..0000000 --- a/ui/services/config_service.py +++ /dev/null @@ -1,51 +0,0 @@ -"""配置服务""" - -from pathlib import Path -from typing import TYPE_CHECKING -from ..models.config import AppConfig - -if TYPE_CHECKING: - from ..core.imports import GraphMemoryClient - - -class ConfigService: - """配置服务""" - - DEFAULT_CONFIG_FILE = Path.home() / ".graph_memory_tui" / "config.json" - - def __init__(self, config_file: Path | None = None): - self._config_file = config_file or self.DEFAULT_CONFIG_FILE - self._config = self._load_config() - - def _load_config(self) -> AppConfig: - """加载配置""" - # 优先从文件加载 - if self._config_file.exists(): - return AppConfig.from_file(self._config_file) - - # 否则从环境变量加载 - return AppConfig.from_env() - - def get_config(self) -> AppConfig: - """获取当前配置""" - return self._config - - def set_config(self, config: AppConfig) -> None: - """设置配置""" - self._config = config - self._save_config() - - def _save_config(self) -> None: - """保存配置""" - self._config.save(self._config_file) - - def apply_to_client(self, client: "GraphMemoryClient") -> None: - """应用配置到 API 客户端""" - # 更新客户端配置 - client.api_key = self._config.api_key - client.base_url = self._config.base_url - client.model = self._config.model - - def get_config_file(self) -> Path: - """获取配置文件路径""" - return self._config_file diff --git a/ui/styles/__init__.py b/ui/styles/__init__.py deleted file mode 100644 index 190eec3..0000000 --- a/ui/styles/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Styles for Graph Memory TUI""" diff --git a/ui/styles/app.css b/ui/styles/app.css deleted file mode 100644 index f94d904..0000000 --- a/ui/styles/app.css +++ /dev/null @@ -1,40 +0,0 @@ -/* Global Styles for Graph Memory TUI */ - -GraphMemoryApp { - background: $surface; - color: $text; -} - -/* 全局Input样式 - 确保可见 */ -Input { - background: $surface-lighten-1; - color: $text; - border: solid $primary; -} - -Input:focus { - border: double $accent; -} - -LeftPanel { - width: 1fr; - dock: left; -} - -RightPanel { - width: 35; - dock: right; - background: $panel; -} - -RightPanel ScrollableContainer { - height: 100%; - overflow-y: scroll; -} - -StatusBar { - dock: bottom; - height: 1; - background: $primary; - color: $text-primary; -} diff --git a/ui/styles/components.css b/ui/styles/components.css deleted file mode 100644 index d8fd197..0000000 --- a/ui/styles/components.css +++ /dev/null @@ -1,139 +0,0 @@ -/* Component Styles for Graph Memory TUI */ - -/* Input Box - 最重要 */ -InputBox { - background: $surface; - padding: 1 2; - height: auto; - border: solid $primary; -} - -InputBox TextArea { - width: 100%; - height: 5; - background: $surface-lighten-1; - color: $text; - border: none; -} - -InputBox .input-buttons { - height: auto; - margin-top: 1; -} - -InputBox Button { - margin: 0; -} - -/* Config Section */ -ConfigSection { - background: $surface; - padding: 1; - margin: 0 0 1 0; - height: auto; -} - -ConfigSection .config-title { - color: $primary; - text-style: bold; - margin: 0 0 1 0; -} - -ConfigSection .config-label { - color: $text; - margin: 0; - padding: 1 0 0 0; -} - -ConfigSection .config-hint { - color: $text-muted; - text-style: italic; - margin: 1 0 0 0; -} - -ConfigSection Input { - width: 1fr; - height: 3; - margin: 0 0 1 0; - padding: 0 1; - background: $surface-lighten-1; - border: solid $primary; - color: $text; -} - -/* Other Components */ -OperationLog { - background: $surface-darken-1; - height: 1fr; - margin: 1; - overflow-y: auto; - padding: 1; -} - -OperationLog .log-entry { - color: $text; - margin: 0 0 1 0; - height: auto; -} - -OperationLog .log-empty { - color: $text-muted; - text-style: italic; -} - -CypherQueryBox { - border: solid green; - margin: 1; - height: auto; -} - -MessageHistory { - height: 1fr; - margin: 1; - overflow-y: auto; -} - -/* Message Widget */ -MessageWidget { - margin: 1 0; - height: auto; -} - -MessageWidget .message-header { - color: $text-muted; - text-style: bold; - margin: 0 0 0 0; -} - -MessageWidget .message-content { - color: $text; - margin: 0 0 0 2; - height: auto; -} - -MessageWidget .tool-indicator { - color: $warning; - text-style: bold; - margin: 1 0 0 2; -} - -MessageWidget .tool-details { - background: $surface-darken-1; - margin: 1 0 0 2; - padding: 1; -} - -MessageWidget .tool-name { - color: $accent; - text-style: bold; -} - -MessageWidget .tool-args { - color: $text-muted; - margin: 0 0 0 2; -} - -MessageWidget .tool-result { - color: $success; - margin: 0 0 0 2; -} diff --git a/ui/styles/messages.css b/ui/styles/messages.css deleted file mode 100644 index 0d7b1d4..0000000 --- a/ui/styles/messages.css +++ /dev/null @@ -1,24 +0,0 @@ -/* Message Styles for Graph Memory TUI */ - -UserMessage { - border: solid orange; - margin: 1 0; - padding: 1; -} - -ModelMessage { - border: solid blue; - margin: 1 0; - padding: 1; -} - -ToolCallIndicator { - color: yellow; - text-style: bold; -} - -ToolCallDetails { - background: $surface-darken-1; - margin: 1 0 0 2; - padding: 1; -} diff --git a/ui/widgets/__init__.py b/ui/widgets/__init__.py deleted file mode 100644 index fd32402..0000000 --- a/ui/widgets/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""UI Widgets for Graph Memory TUI""" diff --git a/ui/widgets/config_section.py b/ui/widgets/config_section.py deleted file mode 100644 index 33d366c..0000000 --- a/ui/widgets/config_section.py +++ /dev/null @@ -1,274 +0,0 @@ -"""配置区组件""" - -from textual.containers import Vertical -from textual.widgets import Static, Input, Button -from textual.app import ComposeResult -from textual.message import Message -from ..models.config import AppConfig - - -class ConfigSection(Vertical): - - class ConfigChanged(Message): - def __init__(self, config: AppConfig, is_tool_limits: bool = False) -> None: - self.config = config - self.is_tool_limits = is_tool_limits - super().__init__() - - def __init__(self, config: AppConfig | None = None, is_admin: bool = True, **kwargs): - super().__init__(**kwargs) - self._config = config or AppConfig() - self._is_admin = is_admin - - def compose(self) -> ComposeResult: - title = Static("━━ 配置 ━━", classes="config-title") - title.can_focus = False - yield title - - label1 = Static("API Key:", classes="config-label") - label1.can_focus = False - yield label1 - - yield Input( - value=self._config.api_key, - placeholder="sk-xxxxxxxxxxxxx", - id="api-key-input", - password=True - ) - - label2 = Static("模型:", classes="config-label") - label2.can_focus = False - yield label2 - - yield Input( - value=self._config.model, - placeholder="deepseek-chat", - id="model-input" - ) - - label3 = Static("Base URL:", classes="config-label") - label3.can_focus = False - yield label3 - - yield Input( - value=self._config.base_url, - placeholder="https://api.deepseek.com", - id="base-url-input" - ) - - sep = Static("", classes="config-sep") - sep.can_focus = False - yield sep - - limits_title = Static("━━ 工具限制 ━━", classes="config-title") - limits_title.can_focus = False - yield limits_title - - l2 = Static("人设图修改:", classes="config-label") - l2.can_focus = False - yield l2 - yield Input(value=str(self._config.persona_update_max), placeholder="1", id="persona-update-max") - - l4 = Static("工作记忆修改:", classes="config-label") - l4.can_focus = False - yield l4 - yield Input(value=str(self._config.task_update_max), placeholder="5", id="task-update-max") - - l5 = Static("一般记忆查询:", classes="config-label") - l5.can_focus = False - yield l5 - yield Input(value=str(self._config.memory_query_max), placeholder="20", id="memory-query-max") - - l6 = Static("一般记忆修改:", classes="config-label") - l6.can_focus = False - yield l6 - yield Input(value=str(self._config.memory_update_max), placeholder="10", id="memory-update-max") - - sep2 = Static("", classes="config-sep") - sep2.can_focus = False - yield sep2 - - from textual.containers import Vertical - - # Web 登录 — 仅 admin 可见 - with Vertical(id="admin-web-login-section"): - web_title = Static("━━ Web 登录 ━━", classes="config-title") - web_title.can_focus = False - yield web_title - - label_web_user = Static("用户名:", classes="config-label") - label_web_user.can_focus = False - yield label_web_user - yield Input( - value=self._config.web_username, - placeholder="admin", - id="web-username-input" - ) - - label_web_pwd = Static("密码:", classes="config-label") - label_web_pwd.can_focus = False - yield label_web_pwd - yield Input( - value=self._config.web_password, - placeholder="修改密码", - id="web-password-input", - password=True - ) - - hint = Static("按Enter保存配置", classes="config-hint") - hint.can_focus = False - yield hint - - sep3 = Static("", classes="config-sep") - sep3.can_focus = False - yield sep3 - - # Web 服务 — 仅 admin 可见 - with Vertical(id="admin-web-service-section"): - ws_title = Static("━━ Web 服务 ━━", classes="config-title") - ws_title.can_focus = False - yield ws_title - - ws_hint = Static("在侧边栏启用后将自动启动 Web 管理界面", classes="config-hint") - ws_hint.can_focus = False - yield ws_hint - - from textual.widgets import Checkbox - yield Checkbox( - "启用 Web 服务", - value=self._config.enable_web, - id="enable-web-checkbox" - ) - - label_web_port = Static("端口:", classes="config-label") - label_web_port.can_focus = False - yield label_web_port - yield Input( - value=str(self._config.web_port), - placeholder="4096", - id="web-port-input", - type="integer" - ) - - # 默认隐藏 admin 区域,等 login 后决定是否显示 - self._apply_admin_visibility() - - def on_mount(self) -> None: - try: - api_key = self.query_one("#api-key-input", Input) - model = self.query_one("#model-input", Input) - base_url = self.query_one("#base-url-input", Input) - - api_key.tab_index = 0 - model.tab_index = 1 - base_url.tab_index = 2 - - if self._config.api_key: - api_key.value = self._config.api_key - if self._config.model: - model.value = self._config.model - if self._config.base_url: - base_url.value = self._config.base_url - - web_user = self.query_one("#web-username-input", Input) - web_pwd = self.query_one("#web-password-input", Input) - web_user.tab_index = 7 - web_pwd.tab_index = 8 - - from textual.widgets import Checkbox - web_port = self.query_one("#web-port-input", Input) - try: - web_checkbox = self.query_one("#enable-web-checkbox", Checkbox) - except: - pass - web_port.tab_index = 9 - except Exception: - pass - - def on_input_submitted(self, event: Input.Submitted) -> None: - try: - api_key_input = self.query_one("#api-key-input", Input) - model_input = self.query_one("#model-input", Input) - base_url_input = self.query_one("#base-url-input", Input) - - persona_update = self.query_one("#persona-update-max", Input) - task_update = self.query_one("#task-update-max", Input) - memory_query = self.query_one("#memory-query-max", Input) - memory_update = self.query_one("#memory-update-max", Input) - - web_username = self.query_one("#web-username-input", Input) - web_password = self.query_one("#web-password-input", Input) - - from textual.widgets import Checkbox - web_checkbox = self.query_one("#enable-web-checkbox", Checkbox) - web_port_input = self.query_one("#web-port-input", Input) - - self._config = AppConfig( - api_key=api_key_input.value, - model=model_input.value, - base_url=base_url_input.value, - persona_update_max=int(persona_update.value or 1), - task_update_max=int(task_update.value or 5), - memory_query_max=int(memory_query.value or 20), - memory_update_max=int(memory_update.value or 10), - web_username=web_username.value, - web_password=web_password.value, - enable_web=web_checkbox.value, - web_port=int(web_port_input.value) if web_port_input.value else 4096, - ) - - # 先发送 API 配置更新(is_tool_limits=False) - self.post_message(self.ConfigChanged(self._config, is_tool_limits=False)) - # 再发送工具限制更新(is_tool_limits=True) - self.post_message(self.ConfigChanged(self._config, is_tool_limits=True)) - except Exception: - pass - - def set_admin(self, is_admin: bool) -> None: - """设置是否 admin 模式,动态显示/隐藏 admin 区域""" - self._is_admin = is_admin - self._apply_admin_visibility() - - def _apply_admin_visibility(self) -> None: - """根据 _is_admin 显示/隐藏 admin 专用区域""" - try: - login_section = self.query_one("#admin-web-login-section") - login_section.styles.display = "block" if self._is_admin else "none" - except Exception: - pass - try: - service_section = self.query_one("#admin-web-service-section") - service_section.styles.display = "block" if self._is_admin else "none" - except Exception: - pass - - def get_config(self) -> AppConfig: - return self._config - - def set_config(self, config: AppConfig) -> None: - self._config = config - try: - api_key_input = self.query_one("#api-key-input", Input) - model_input = self.query_one("#model-input", Input) - base_url_input = self.query_one("#base-url-input", Input) - - api_key_input.value = config.api_key - model_input.value = config.model - base_url_input.value = config.base_url - - self.query_one("#persona-update-max", Input).value = str(config.persona_update_max) - self.query_one("#task-update-max", Input).value = str(config.task_update_max) - self.query_one("#memory-query-max", Input).value = str(config.memory_query_max) - self.query_one("#memory-update-max", Input).value = str(config.memory_update_max) - - self.query_one("#web-username-input", Input).value = config.web_username - self.query_one("#web-password-input", Input).value = config.web_password - - from textual.widgets import Checkbox - try: - self.query_one("#enable-web-checkbox", Checkbox).value = config.enable_web - except: - pass - self.query_one("#web-port-input", Input).value = str(config.web_port) - except Exception: - pass diff --git a/ui/widgets/cypher_query_box.py b/ui/widgets/cypher_query_box.py deleted file mode 100644 index 789db02..0000000 --- a/ui/widgets/cypher_query_box.py +++ /dev/null @@ -1,57 +0,0 @@ -"""Cypher查询框组件""" - -from textual.containers import Container, Horizontal -from textual.widgets import Static, TextArea, Button -from textual.app import ComposeResult -from textual.message import Message - - -class CypherQueryBox(Container): - """快捷Cypher查询输入框""" - - class ExecuteQuery(Message): - """执行查询事件""" - def __init__(self, query: str) -> None: - self.query = query - super().__init__() - - def compose(self) -> ComposeResult: - """构建查询框""" - yield Static("F4:执行Cypher查询", classes="query-title") - yield TextArea( - placeholder="输入Cypher查询语句...", - id="cypher-textarea" - ) - with Horizontal(classes="query-buttons"): - yield Button("执行", id="execute-button", variant="primary") - yield Button("清空", id="clear-button") - - def on_button_pressed(self, event: Button.Pressed) -> None: - """处理按钮点击""" - if event.button.id == "execute-button": - self._execute_query() - elif event.button.id == "clear-button": - self._clear_query() - - def on_key(self, event) -> None: - """处理按键事件""" - if event.key == "enter" and event.ctrl: - event.stop() - self._execute_query() - - def _execute_query(self) -> None: - """执行查询""" - textarea = self.query_one("#cypher-textarea", TextArea) - query = textarea.text.strip() - if query: - self.post_message(self.ExecuteQuery(query)) - - def _clear_query(self) -> None: - """清空查询""" - textarea = self.query_one("#cypher-textarea", TextArea) - textarea.clear() - - def focus(self) -> None: - """聚焦查询框""" - textarea = self.query_one("#cypher-textarea", TextArea) - textarea.focus() diff --git a/ui/widgets/input_box.py b/ui/widgets/input_box.py deleted file mode 100644 index 03335be..0000000 --- a/ui/widgets/input_box.py +++ /dev/null @@ -1,72 +0,0 @@ -"""输入框组件""" - -from textual.containers import Container, Horizontal -from textual.widgets import TextArea, Button -from textual.message import Message - - -class InputBox(Container): - """输入框组件""" - - class SendMessage(Message): - """发送消息事件""" - def __init__(self, content: str) -> None: - self.content = content - super().__init__() - - class ClearHistory(Message): - """清空聊天记录事件""" - def __init__(self) -> None: - super().__init__() - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self._history: list[str] = [] - self._history_index: int = -1 - - def compose(self): - """构建输入框""" - yield TextArea( - placeholder="输入消息... (Enter换行)", - id="input-textarea" - ) - with Horizontal(classes="input-buttons"): - yield Button("清空", id="clear-button", variant="default") - yield Button("发送", id="send-button", variant="primary") - - def on_mount(self) -> None: - """组件挂载时""" - # 设置焦点 - textarea = self.query_one(TextArea) - textarea.focus() - - def on_button_pressed(self, event: Button.Pressed) -> None: - """处理按钮点击""" - if event.button.id == "send-button": - self._send_message() - elif event.button.id == "clear-button": - self.post_message(self.ClearHistory()) - - def on_key(self, event) -> None: - """处理按键事件""" - if event.key == "enter" and event.ctrl: - self._send_message() - event.stop() - - def _send_message(self) -> None: - """发送消息""" - textarea = self.query_one(TextArea) - content = textarea.text.strip() - if content: - # 保存到历史 - self._history.append(content) - self._history_index = len(self._history) - # 发送消息 - self.post_message(self.SendMessage(content)) - # 清空输入框 - textarea.clear() - - def focus(self) -> None: - """聚焦输入框""" - textarea = self.query_one(TextArea) - textarea.focus() diff --git a/ui/widgets/left_panel.py b/ui/widgets/left_panel.py deleted file mode 100644 index 6cdc22b..0000000 --- a/ui/widgets/left_panel.py +++ /dev/null @@ -1,23 +0,0 @@ -"""左侧面板""" - -from textual.containers import Container -from textual.app import ComposeResult -from .message_history import MessageHistory -from .input_box import InputBox - - -class LeftPanel(Container): - """左侧主面板""" - - def compose(self) -> ComposeResult: - """构建左侧面板""" - yield MessageHistory() - yield InputBox() - - def get_message_history(self) -> MessageHistory: - """获取消息历史组件""" - return self.query_one(MessageHistory) - - def get_input_box(self) -> InputBox: - """获取输入框组件""" - return self.query_one(InputBox) diff --git a/ui/widgets/message_history.py b/ui/widgets/message_history.py deleted file mode 100644 index e755551..0000000 --- a/ui/widgets/message_history.py +++ /dev/null @@ -1,57 +0,0 @@ -"""消息历史组件""" - -from textual.containers import ScrollableContainer -from textual.message import Message -from .message_widget import MessageWidget -from ..models.message import Message as MessageModel - - -class MessageHistory(ScrollableContainer): - """消息历史区域""" - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self._messages: list[MessageModel] = [] - - def compose(self): - """构建消息历史""" - for message in self._messages: - yield MessageWidget(message) - - def add_message(self, message: MessageModel) -> None: - """添加新消息""" - self._messages.append(message) - # 添加新组件 - message_widget = MessageWidget(message) - self.mount(message_widget) - # 滚动到最新消息 - self.scroll_to_widget(message_widget, animate=False) - - def update_latest_message(self, content: str) -> None: - """更新最新消息的内容""" - if self.children: - latest_widget = self.children[-1] - if isinstance(latest_widget, MessageWidget): - latest_widget.update_content(content) - # 确保滚动到最新消息 - self.scroll_to_widget(latest_widget, animate=False) - - def clear_messages(self) -> None: - """清空消息历史""" - self._messages.clear() - # 移除所有子组件 - for child in self.children: - child.remove() - - def get_latest_message(self) -> MessageModel | None: - """获取最新消息""" - if self._messages: - return self._messages[-1] - return None - - def toggle_latest_tool_details(self) -> None: - """切换最新消息的工具详情""" - if self.children: - latest_widget = self.children[-1] - if isinstance(latest_widget, MessageWidget): - latest_widget.toggle_tool_details() diff --git a/ui/widgets/message_widget.py b/ui/widgets/message_widget.py deleted file mode 100644 index a6929de..0000000 --- a/ui/widgets/message_widget.py +++ /dev/null @@ -1,108 +0,0 @@ -"""消息组件""" - -from textual.containers import Container, Vertical -from textual.widgets import Static -from textual.message import Message -from textual.css.query import NoMatches -from ..models.message import Message as MessageModel - - -class MessageWidget(Container): - """单条消息组件""" - - def __init__(self, message: MessageModel, **kwargs): - super().__init__(**kwargs) - self._message = message - self._show_tool_details = False - self._content_widget = None # 保存内容组件的引用 - self._tool_details_container = None # 保存工具详情容器引用 - - def compose(self): - """构建消息组件""" - # 消息头 - role_emoji = "🟠" if self._message.role == "user" else "🔵" - timestamp_str = self._message.timestamp.strftime("%H:%M:%S") - yield Static( - f"{role_emoji} {timestamp_str}", - classes="message-header" - ) - - # 消息内容 - 保存引用以便后续更新 - self._content_widget = Static( - self._message.content, - classes="message-content" - ) - yield self._content_widget - - # 工具调用指示器 - if self._message.tool_calls: - tool_count = len(self._message.tool_calls) - toggle_hint = "(F3折叠)" if self._show_tool_details else "(F3展开)" - yield Static( - f"[工具:{tool_count}次] {toggle_hint}", - classes="tool-indicator" - ) - - # 工具调用详情容器 - 始终创建,但根据状态显示/隐藏 - self._tool_details_container = Vertical(classes="tool-details") - with self._tool_details_container: - for i, tool_call in enumerate(self._message.tool_calls, 1): - yield Static( - f"工具 {i}: {tool_call.name}", - classes="tool-name" - ) - yield Static( - f"参数: {tool_call.arguments}", - classes="tool-args" - ) - - # 显示执行结果 - if self._message.tool_results: - for result in self._message.tool_results: - if result.tool_call_id == tool_call.id: - # 显示完整结果,不截断 - result_text = result.result - # 如果结果太长,只显示前1000字符,但提供完整信息 - if len(result_text) > 1000: - result_text = result_text[:1000] + f"\n... (共{len(result.result)}字符,按F3查看完整内容)" - yield Static( - f"结果: {result_text}", - classes="tool-result" - ) - - # 根据状态设置初始显示/隐藏 - if not self._show_tool_details: - self._tool_details_container.styles.display = "none" - - def update_content(self, new_content: str) -> None: - """更新消息内容""" - self._message.content = new_content - if self._content_widget: - self._content_widget.update(new_content) - - def toggle_tool_details(self) -> None: - """切换工具详情显示状态""" - if self._message.tool_calls and self._tool_details_container: - self._show_tool_details = not self._show_tool_details - - # 切换显示/隐藏 - if self._show_tool_details: - self._tool_details_container.styles.display = "block" - else: - self._tool_details_container.styles.display = "none" - - # 更新指示器文字 - self._update_indicator() - - # 刷新布局 - self.refresh(layout=True) - - def _update_indicator(self) -> None: - """更新工具调用指示器文字""" - try: - indicator = self.query_one(".tool-indicator", Static) - tool_count = len(self._message.tool_calls) - toggle_hint = "(F3折叠)" if self._show_tool_details else "(F3展开)" - indicator.update(f"[工具:{tool_count}次] {toggle_hint}") - except NoMatches: - pass diff --git a/ui/widgets/operation_log.py b/ui/widgets/operation_log.py deleted file mode 100644 index 7806f97..0000000 --- a/ui/widgets/operation_log.py +++ /dev/null @@ -1,66 +0,0 @@ -"""操作日志组件""" - -from datetime import datetime -from textual.containers import ScrollableContainer -from textual.widgets import Static -from ..models.log_entry import LogEntry - - -class OperationLog(ScrollableContainer): - """图操作日志区域""" - - def __init__(self, max_entries: int = 100, **kwargs): - super().__init__(**kwargs) - self._logs: list[LogEntry] = [] - self._max_entries = max_entries - - def compose(self): - """构建日志区域""" - if not self._logs: - yield Static("暂无操作日志", classes="log-empty") - - def add_log(self, entry: LogEntry) -> None: - """添加日志(插入到顶部)""" - # 限制日志数量 - if len(self._logs) >= self._max_entries: - self._logs.pop() - # 移除最旧的组件 - if self.children: - self.children[-1].remove() - - # 插入到列表开头 - self._logs.insert(0, entry) - - # 创建日志显示组件 - log_widget = self._create_log_widget(entry) - - # 挂载到顶部 - self.mount(log_widget, before=0 if self.children else None) - - # 滚动到顶部 - self.scroll_to(0, animate=False) - - def _create_log_widget(self, entry: LogEntry) -> Static: - """创建日志显示组件""" - timestamp_str = entry.timestamp.strftime("%H:%M:%S") - text = ( - f"[{timestamp_str}] {entry.tool_name}\n" - f" 参数: {entry.args_summary}\n" - f" 结果: {entry.result_summary}\n" - f" 耗时: {entry.duration:.2f}s" - ) - return Static(text, classes="log-entry") - - def clear_logs(self) -> None: - """清空日志""" - self._logs.clear() - for child in self.children: - child.remove() - # 显示空状态 - self.mount(Static("暂无操作日志", classes="log-empty")) - - def get_latest_log(self) -> LogEntry | None: - """获取最新日志""" - if self._logs: - return self._logs[0] - return None diff --git a/ui/widgets/right_panel.py b/ui/widgets/right_panel.py deleted file mode 100644 index 6692e14..0000000 --- a/ui/widgets/right_panel.py +++ /dev/null @@ -1,67 +0,0 @@ -"""右侧面板""" - -from textual.containers import Container, ScrollableContainer -from textual.css.query import NoMatches -from textual.widgets import Static -from textual.app import ComposeResult -from .config_section import ConfigSection -from .operation_log import OperationLog -from .cypher_query_box import CypherQueryBox -from ..models.config import AppConfig - - -class RightPanel(Container): - """右侧边栏""" - - def __init__(self, config: AppConfig | None = None, use_embedded_db: bool = True, **kwargs): - super().__init__(**kwargs) - self._is_collapsed = False - self._config = config or AppConfig() - self._use_embedded_db = use_embedded_db - - def compose(self) -> ComposeResult: - """构建右侧面板""" - yield Static("F2:隐藏侧边栏", classes="sidebar-title") - with ScrollableContainer(): - yield ConfigSection(self._config) - yield OperationLog() - if not self._use_embedded_db: - yield CypherQueryBox() - - def toggle(self) -> None: - """切换折叠/展开""" - self._is_collapsed = not self._is_collapsed - if self._is_collapsed: - self.styles.width = 0 - self.styles.display = "none" - else: - self.styles.width = 35 - self.styles.display = "block" - - def is_collapsed(self) -> bool: - """检查是否折叠""" - return self._is_collapsed - - def get_config_section(self) -> ConfigSection: - """获取配置区组件""" - return self.query_one(ConfigSection) - - def get_operation_log(self) -> OperationLog: - """获取操作日志组件""" - return self.query_one(OperationLog) - - def get_cypher_query_box(self) -> CypherQueryBox | None: - """获取Cypher查询框组件(可能不存在)""" - try: - return self.query_one(CypherQueryBox) - except NoMatches: - return None - - def has_cypher_query_box(self) -> bool: - """检查是否存在Cypher查询框""" - return not self._use_embedded_db - - def update_title(self) -> None: - """更新标题""" - title = self.query_one(Static) - title.update("F2:展开侧边栏" if self._is_collapsed else "F2:隐藏侧边栏") diff --git a/ui/widgets/status_bar.py b/ui/widgets/status_bar.py deleted file mode 100644 index f7168be..0000000 --- a/ui/widgets/status_bar.py +++ /dev/null @@ -1,41 +0,0 @@ -"""状态栏组件""" - -from textual.widgets import Static -from textual.message import Message - - -class StatusBar(Static): - """底部状态栏""" - - class FocusChanged(Message): - """焦点变更事件""" - def __init__(self, focus_name: str) -> None: - self.focus_name = focus_name - super().__init__() - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self._shortcuts = "F1:帮助 F2:侧边栏 F3:工具详情 F5:清屏 F6:退出" - self._license_info = "本项目由jianf设计,以GPLv3形式开源" - self._api_status = "未配置" - self._processing = False - - def on_mount(self) -> None: - """组件挂载时""" - self._update_display() - - def _update_display(self) -> None: - """更新显示""" - status_icon = "●" if self._api_status == "已配置" else "○" - processing_indicator = " [处理中...]" if self._processing else "" - self.update(f"{status_icon} API: {self._api_status}{processing_indicator} | {self._license_info} | {self._shortcuts}") - - def set_api_status(self, configured: bool) -> None: - """设置API状态""" - self._api_status = "已配置" if configured else "未配置" - self._update_display() - - def set_processing(self, processing: bool) -> None: - """设置处理状态""" - self._processing = processing - self._update_display()