jianf/TrulyMEM-TrueHumanMEM
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/TrulyMEM-TrueHumanMEM.git synced 2026-09-21 17:38:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: 删除 TUI/多用户/CLI 代码，准备鸿蒙工程

Browse Source
This commit is contained in:
root
2026-04-28 12:34:26 +08:00
parent c4f7b5a645
commit be6555cca1
29 changed files with 0 additions and 2081 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
core/migrate.py
View File

@ -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()}")

87
trulymem_entry.py
View File

@ -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()

4
ui/__init__.py
View File

@ -1,4 +0,0 @@
from .app import GraphMemoryApp
from .models.config import AppConfig
__all__ = ["GraphMemoryApp", "AppConfig"]

367
ui/app.py
View File

@ -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")

1
ui/handlers/__init__.py
View File

@ -1 +0,0 @@
"""Event Handlers for Graph Memory TUI"""

64
ui/handlers/focus_handler.py
View File

@ -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)

68
ui/handlers/key_handler.py
View File

@ -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)

142
ui/login_screen.py
View File

@ -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)}")

1
ui/models/__init__.py
View File

@ -1 +0,0 @@
"""Data Models for Graph Memory TUI"""

81
ui/models/config.py
View File

@ -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)

30
ui/models/log_entry.py
View File

@ -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

33
ui/models/message.py
View File

@ -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

1
ui/services/__init__.py
View File

@ -1 +0,0 @@
"""Business Services for Graph Memory TUI"""

46
ui/services/config_manager.py
View File

@ -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()

51
ui/services/config_service.py
View File

@ -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

1
ui/styles/__init__.py
View File

@ -1 +0,0 @@
"""Styles for Graph Memory TUI"""

40
ui/styles/app.css
View File

@ -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;
}

139
ui/styles/components.css
View File

@ -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;
}

24
ui/styles/messages.css
View File

@ -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;
}

1
ui/widgets/__init__.py
View File

@ -1 +0,0 @@
"""UI Widgets for Graph Memory TUI"""

274
ui/widgets/config_section.py
View File

@ -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

57
ui/widgets/cypher_query_box.py
View File

@ -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()

72
ui/widgets/input_box.py
View File

@ -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()

23
ui/widgets/left_panel.py
View File

@ -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)

57
ui/widgets/message_history.py
View File

@ -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()

108
ui/widgets/message_widget.py
View File

@ -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

66
ui/widgets/operation_log.py
View File

@ -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

67
ui/widgets/right_panel.py
View File

@ -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:隐藏侧边栏")

41
ui/widgets/status_bar.py
View File

@ -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()