328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""配置区组件"""
|
||
|
||
from textual.containers import Vertical
|
||
from textual.widgets import Static, Input, Button, Checkbox
|
||
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 = False, **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-v4-flash",
|
||
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
|
||
|
||
# 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
|
||
|
||
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"
|
||
)
|
||
|
||
sep4 = Static("", classes="config-sep")
|
||
sep4.can_focus = False
|
||
yield sep4
|
||
|
||
# TUI 服务控制 — 仅 admin 可见
|
||
with Vertical(id="admin-tui-service-section"):
|
||
tui_title = Static("━━ TUI 服务 ━━", classes="config-title")
|
||
tui_title.can_focus = False
|
||
yield tui_title
|
||
|
||
tui_hint = Static("控制终端文本界面是否允许连接", classes="config-hint")
|
||
tui_hint.can_focus = False
|
||
yield tui_hint
|
||
|
||
yield Checkbox(
|
||
"启用 TUI 服务",
|
||
value=self._config.enable_tui,
|
||
id="enable-tui-checkbox"
|
||
)
|
||
|
||
def on_mount(self) -> None:
|
||
# 应用管理员区域的可见性
|
||
self._apply_admin_visibility()
|
||
|
||
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
|
||
|
||
web_port = self.query_one("#web-port-input", Input)
|
||
web_port.tab_index = 9
|
||
except Exception:
|
||
pass
|
||
|
||
def on_checkbox_changed(self, event: Checkbox.Changed) -> None:
|
||
"""处理复选框状态变化"""
|
||
try:
|
||
# Web 服务开关
|
||
if event.checkbox.id == "enable-web-checkbox":
|
||
self._update_config_from_ui()
|
||
# 发送配置更新消息
|
||
self.post_message(self.ConfigChanged(self._config, is_tool_limits=False))
|
||
|
||
# TUI 服务开关
|
||
elif event.checkbox.id == "enable-tui-checkbox":
|
||
self._update_config_from_ui()
|
||
# 发送配置更新消息
|
||
self.post_message(self.ConfigChanged(self._config, is_tool_limits=False))
|
||
except Exception:
|
||
pass
|
||
|
||
def _update_config_from_ui(self) -> None:
|
||
"""从UI组件更新配置对象"""
|
||
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)
|
||
|
||
web_checkbox = self.query_one("#enable-web-checkbox", Checkbox)
|
||
web_port_input = self.query_one("#web-port-input", Input)
|
||
|
||
# TUI checkbox 可能不存在（非管理员）
|
||
tui_checkbox = None
|
||
try:
|
||
tui_checkbox = self.query_one("#enable-tui-checkbox", Checkbox)
|
||
except:
|
||
pass
|
||
|
||
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,
|
||
enable_tui=tui_checkbox.value if tui_checkbox else True,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
def on_input_submitted(self, event: Input.Submitted) -> None:
|
||
try:
|
||
self._update_config_from_ui()
|
||
|
||
# 先发送 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
|
||
try:
|
||
tui_section = self.query_one("#admin-tui-service-section")
|
||
tui_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
|
||
|
||
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)
|
||
|
||
# TUI checkbox
|
||
try:
|
||
self.query_one("#enable-tui-checkbox", Checkbox).value = config.enable_tui
|
||
except:
|
||
pass
|
||
except Exception:
|
||
pass
|