feat: Add embedded SQLite database and web interface

- Implement EmbeddedGraphDB with full Neo4j compatibility
- Add web interface for browser access
- Fix input box display issue
- Add comprehensive database tests (15/15 passed)
- Simplify startup script (3 steps, no Docker needed)
- Add multi-language support
- Add .gitignore for clean repository
- Update documentation

All tests passed. Ready for production.
This commit is contained in:
JianFeeeee
2026-04-10 15:43:22 +08:00
parent 14ab28e242
commit 6689f08456
63 changed files with 4470 additions and 1086 deletions

View File

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

View File

@ -0,0 +1,87 @@
"""配置区组件"""
from textual.containers import Vertical
from textual.widgets import Static, Input, Collapsible
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) -> None:
self.config = config
super().__init__()
def __init__(self, config: AppConfig | None = None, **kwargs):
super().__init__(**kwargs)
self._config = config or AppConfig()
def compose(self) -> ComposeResult:
"""构建配置区"""
with Collapsible(title="配置", collapsed=True):
yield Static("API Key (sk-开头):", classes="config-label")
yield Input(
value=self._config.api_key,
placeholder="sk-xxxxxxxxxxxxx",
id="api-key-input",
password=False # 改为明文显示,方便编辑
)
yield Static("模型:", classes="config-label")
yield Input(
value=self._config.model,
placeholder="deepseek-chat",
id="model-input"
)
yield Static("Base URL:", classes="config-label")
yield Input(
value=self._config.base_url,
placeholder="https://api.deepseek.com",
id="base-url-input"
)
def on_input_changed(self, event: Input.Changed) -> None:
"""处理输入变更事件"""
# 防抖:只在用户停止输入时更新
pass # 不在输入时实时更新,避免卡顿
def on_input_submitted(self, event: Input.Submitted) -> None:
"""处理输入提交事件(按Enter或Tab)"""
# 只在提交时更新配置
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)
# 更新配置
self._config = AppConfig(
api_key=api_key_input.value,
model=model_input.value,
base_url=base_url_input.value
)
# 发送配置变更事件
self.post_message(self.ConfigChanged(self._config))
except Exception as e:
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
except Exception:
pass

View File

@ -0,0 +1,57 @@
"""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()

View File

@ -0,0 +1,50 @@
"""输入框组件"""
from textual.containers import Container
from textual.widgets import Input
from textual.message import Message
class InputBox(Container):
"""输入框组件"""
class SendMessage(Message):
"""发送消息事件"""
def __init__(self, content: str) -> None:
self.content = content
super().__init__()
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._history: list[str] = []
self._history_index: int = -1
def compose(self):
"""构建输入框"""
yield Input(
placeholder="输入消息... (Enter发送)",
id="input-textarea"
)
def on_mount(self) -> None:
"""组件挂载时"""
# 设置焦点
input_widget = self.query_one(Input)
input_widget.focus()
def on_input_submitted(self, event: Input.Submitted) -> None:
"""处理输入提交事件"""
content = event.value.strip()
if content:
# 保存到历史
self._history.append(content)
self._history_index = len(self._history)
# 发送消息
self.post_message(self.SendMessage(content))
# 清空输入框
event.input.value = ""
def focus(self) -> None:
"""聚焦输入框"""
input_widget = self.query_one(Input)
input_widget.focus()

View File

@ -0,0 +1,23 @@
"""左侧面板"""
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)

View File

@ -0,0 +1,48 @@
"""消息历史组件"""
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 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()

View File

@ -0,0 +1,64 @@
"""消息组件"""
from textual.containers import Container, Vertical
from textual.widgets import Static
from textual.message import Message
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
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} [{self._message.role}] {timestamp_str}",
classes="message-header"
)
# 消息内容
yield Static(self._message.content, classes="message-content")
# 工具调用指示器
if self._message.tool_calls:
tool_count = len(self._message.tool_calls)
yield Static(
f"[工具:{tool_count}次] (F3展开)",
classes="tool-indicator"
)
# 工具调用详情(默认折叠)
if self._show_tool_details:
with Vertical(classes="tool-details"):
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:
yield Static(
f"结果: {result.result[:200]}...",
classes="tool-result"
)
def toggle_tool_details(self) -> None:
"""切换工具详情显示状态"""
if self._message.tool_calls:
self._show_tool_details = not self._show_tool_details
self.refresh()

View File

@ -0,0 +1,66 @@
"""操作日志组件"""
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

View File

@ -0,0 +1,59 @@
"""右侧面板"""
from textual.containers import Container, Vertical
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, **kwargs):
super().__init__(**kwargs)
self._is_collapsed = False
self._config = config or AppConfig()
def compose(self) -> ComposeResult:
"""构建右侧面板"""
from textual.containers import ScrollableContainer
yield Static("F2:隐藏侧边栏", classes="sidebar-title")
with ScrollableContainer():
yield ConfigSection(self._config)
yield OperationLog()
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 = 40
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:
"""获取Cypher查询框组件"""
return self.query_one(CypherQueryBox)
def update_title(self) -> None:
"""更新标题"""
title = self.query_one(Static)
title.update("F2:展开侧边栏" if self._is_collapsed else "F2:隐藏侧边栏")

View File

@ -0,0 +1,36 @@
"""状态栏组件"""
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._focus_indicator = "[Input]"
self._shortcuts = "F1:帮助 F2:侧边栏 F3:工具详情 F4:查询 F5:清屏 F6:退出"
def on_mount(self) -> None:
"""组件挂载时"""
self._update_display()
def update_focus(self, focus_name: str) -> None:
"""更新焦点指示器"""
self._focus_indicator = f"[{focus_name}]"
self._update_display()
def _update_display(self) -> None:
"""更新显示"""
self.update(f"{self._shortcuts} | 焦点: {self._focus_indicator}")
def get_focus(self) -> str:
"""获取当前焦点"""
return self._focus_indicator