"""
PyQt5 自定义控件
- FileBrowser: 远程文件浏览 + 上传/下载/删除
- MonitorPanel: CPU/内存/磁盘/网络实时监控面板
- AIChatPanel: AI Agent 对话面板
"""
import os
import time
from pathlib import Path
from datetime import datetime
from typing import Optional, List, Dict
from PyQt5.QtCore import Qt, pyqtSignal, QSize
from PyQt5.QtGui import QFont, QColor, QIcon, QPixmap, QPainter, QBrush
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QLineEdit,
QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView,
QFileDialog, QMessageBox, QProgressBar, QTreeWidget, QTreeWidgetItem,
QTextEdit, QSplitter, QFrame, QSizePolicy, QGroupBox, QFormLayout,
QComboBox, QToolButton, QStyle, QApplication, QInputDialog,
QListWidget, QListWidgetItem, QTabWidget,
)
from core.ssh_client import SSHConnection
from core.monitor import SystemMonitor
from core.manager import ConnectionManager
from .workers import (
ListDirWorker, UploadWorker, DownloadWorker, MonitorWorker, AIWorker,
)
# ============================================================
# 文件浏览面板
# ============================================================
class FileBrowser(QWidget):
"""远程 SFTP 文件浏览：路径栏 + 工具栏 + 表格 + 状态栏"""
def __init__(self, manager: ConnectionManager, parent=None):
super().__init__(parent)
self.manager = manager
self.current_host_id: Optional[str] = None
self.current_path: str = "/"
self.cwd_history: List[str] = [] # 简单的前进/后退栈
self.list_worker: Optional[ListDirWorker] = None
self._build_ui()
def _build_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(8, 8, 8, 8)
layout.setSpacing(6)
# 路径栏
nav = QHBoxLayout()
self.btn_back = QPushButton("◀")
self.btn_back.setFixedWidth(32)
self.btn_back.clicked.connect(self._go_back)
self.btn_up = QPushButton("▲")
self.btn_up.setFixedWidth(32)
self.btn_up.clicked.connect(self._go_up)
self.path_edit = QLineEdit("/")
self.path_edit.returnPressed.connect(self._go_to_path)
self.btn_refresh = QPushButton("刷新")
self.btn_refresh.clicked.connect(lambda: self._refresh())
nav.addWidget(self.btn_back)
nav.addWidget(self.btn_up)
nav.addWidget(self.path_edit, 1)
nav.addWidget(self.btn_refresh)
layout.addLayout(nav)
# 工具栏
toolbar = QHBoxLayout()
self.btn_upload = QPushButton("⬆ 上传")
self.btn_upload.clicked.connect(self._upload)
self.btn_download = QPushButton("⬇ 下载")
self.btn_download.clicked.connect(self._download)
self.btn_mkdir = QPushButton("新建目录")
self.btn_mkdir.clicked.connect(self._mkdir)
self.btn_delete = QPushButton("删除")
self.btn_delete.clicked.connect(self._delete)
self.btn_rename = QPushButton("重命名")
self.btn_rename.clicked.connect(self._rename)
for w in (self.btn_upload, self.btn_download, self.btn_mkdir,
self.btn_delete, self.btn_rename):
toolbar.addWidget(w)
toolbar.addStretch(1)
self.progress = QProgressBar()
self.progress.setFixedWidth(180)
self.progress.setVisible(False)
toolbar.addWidget(self.progress)
layout.addLayout(toolbar)
# 文件表格
self.table = QTableWidget(0, 4)
self.table.setHorizontalHeaderLabels(["名称", "大小", "修改时间", "类型"])
self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch)
self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeToContents)
self.table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeToContents)
self.table.horizontalHeader().setSectionResizeMode(3, QHeaderView.ResizeToContents)
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.table.setAlternatingRowColors(True)
self.table.doubleClicked.connect(self._on_double_clicked)
layout.addWidget(self.table, 1)
# 状态栏
self.status_label = QLabel("未连接")
self.status_label.setStyleSheet("color: #888;")
layout.addWidget(self.status_label)
# ------- 连接管理 -------
def set_host(self, host_id: str):
self.current_host_id = host_id
self.current_path = "/"
self.path_edit.setText("/")
self.cwd_history.clear()
conn = self.manager.get_connection(host_id)
if conn and conn.connected:
self.status_label.setText(f"已连接: {conn.username}@{conn.host}")
self._refresh()
else:
self.status_label.setText("主机未连接，无法浏览")
self.table.setRowCount(0)
def refresh(self):
self._refresh()
# ------- 内部操作 -------
def _conn(self) -> Optional[SSHConnection]:
if not self.current_host_id:
return None
c = self.manager.get_connection(self.current_host_id)
if not c or not c.connected:
QMessageBox.warning(self, "未连接", "请先在「主机」面板连接当前主机")
return None
return c
def _go_back(self):
if len(self.cwd_history) > 1:
self.cwd_history.pop()
self.current_path = self.cwd_history[-1]
self.path_edit.setText(self.current_path)
self._refresh()
def _go_up(self):
p = self.current_path.rstrip("/")
if not p:
return
parent = os.path.dirname(p) or "/"
self.current_path = parent
self.path_edit.setText(parent)
self._refresh()
def _go_to_path(self):
path = self.path_edit.text().strip() or "/"
self.current_path = path
self._refresh()
def _refresh(self):
conn = self._conn()
if not conn:
return
if self.list_worker and self.list_worker.isRunning():
return
self.status_label.setText(f"加载中: {self.current_path}")
self.list_worker = ListDirWorker(conn, self.current_path)
self.list_worker.finished_with.connect(self._on_list_done)
self.list_worker.start()
def _on_list_done(self, path: str, entries: list):
if path != self.current_path:
return # 用户已跳转
self.table.setRowCount(0)
# 当前目录放第一行
cur = QTableWidgetItem(f"📁 .")
self.table.insertRow(0)
self.table.setItem(0, 0, cur)
self.table.setItem(0, 1, QTableWidgetItem("-"))
self.table.setItem(0, 2, QTableWidgetItem("-"))
self.table.setItem(0, 3, QTableWidgetItem("dir"))
for e in entries:
row = self.table.rowCount()
self.table.insertRow(row)
icon = "📁" if e["is_dir"] else "📄"
self.table.setItem(row, 0, QTableWidgetItem(f"{icon} {e['name']}"))
self.table.setItem(row, 1, QTableWidgetItem(
"-" if e["is_dir"] else SystemMonitor.format_bytes(e["size"])
))
try:
ts = time.strftime("%Y-%m-%d %H:%M", time.localtime(e["mtime"]))
except Exception:
ts = "-"
self.table.setItem(row, 2, QTableWidgetItem(ts))
self.table.setItem(row, 3, QTableWidgetItem("dir" if e["is_dir"] else "file"))
if self.current_path not in self.cwd_history or self.cwd_history[-1] != self.current_path:
self.cwd_history.append(self.current_path)
self.status_label.setText(f"路径: {self.current_path} · 共 {len(entries)} 项")
def _on_double_clicked(self, idx):
row = idx.row()
if row == 0:
self._go_up()
return
name_item = self.table.item(row, 0)
if not name_item:
return
# 去掉 emoji 前缀
name = name_item.text().lstrip("📁📄 ").strip()
type_item = self.table.item(row, 3)
is_dir = type_item and type_item.text() == "dir"
if is_dir:
new_path = self.current_path.rstrip("/") + "/" + name
if not new_path.startswith("/"):
new_path = "/" + new_path
self.current_path = new_path
self.path_edit.setText(new_path)
self._refresh()
else:
self._download_for_item(name)
def _selected_remote_path(self) -> Optional[str]:
rows = self.table.selectionModel().selectedRows()
if not rows:
QMessageBox.information(self, "提示", "请先选中一个文件或目录")
return None
row = rows[0].row()
if row == 0:
return None
name_item = self.table.item(row, 0)
name = name_item.text().lstrip("📁📄 ").strip()
return self.current_path.rstrip("/") + "/" + name
def _upload(self):
conn = self._conn()
if not conn:
return
files, _ = QFileDialog.getOpenFileNames(self, "选择要上传的文件")
if not files:
return
for local in files:
base = os.path.basename(local)
remote = self.current_path.rstrip("/") + "/" + base
self._start_upload(conn, local, remote)
def _start_upload(self, conn, local, remote):
self.progress.setVisible(True)
self.progress.setValue(0)
worker = UploadWorker(conn, local, remote)
worker.progress.connect(lambda d, t: self.progress.setValue(
int(d / t * 100) if t else 0))
worker.finished_with.connect(
lambda ok, msg, w=worker: self._on_upload_done(ok, msg, w))
worker.start()
self._active_workers = getattr(self, "_active_workers", [])
self._active_workers.append(worker)
def _on_upload_done(self, ok, msg, worker):
self.progress.setVisible(False)
if ok:
self.status_label.setText(msg)
self._refresh()
else:
QMessageBox.critical(self, "上传失败", msg)
if worker in getattr(self, "_active_workers", []):
self._active_workers.remove(worker)
def _download(self):
rp = self._selected_remote_path()
if rp:
self._download_for_path(rp)
def _download_for_item(self, name: str):
rp = self.current_path.rstrip("/") + "/" + name
self._download_for_path(rp)
def _download_for_path(self, remote: str):
conn = self._conn()
if not conn:
return
default_name = os.path.basename(remote) or "download"
local, _ = QFileDialog.getSaveFileName(self, "保存到", default_name)
if not local:
return
self.progress.setVisible(True)
self.progress.setValue(0)
worker = DownloadWorker(conn, remote, local)
worker.progress.connect(lambda d, t: self.progress.setValue(
int(d / t * 100) if t else 0))
worker.finished_with.connect(
lambda ok, msg, w=worker: self._on_download_done(ok, msg, w))
worker.start()
self._active_workers = getattr(self, "_active_workers", [])
self._active_workers.append(worker)
def _on_download_done(self, ok, msg, worker):
self.progress.setVisible(False)
if ok:
self.status_label.setText(msg)
else:
QMessageBox.critical(self, "下载失败", msg)
if worker in getattr(self, "_active_workers", []):
self._active_workers.remove(worker)
def _mkdir(self):
conn = self._conn()
if not conn:
return
name, ok = QInputDialog.getText(self, "新建目录", "目录名:")
if not ok or not name.strip():
return
remote = self.current_path.rstrip("/") + "/" + name.strip()
ok2, msg = conn.mkdir(remote)
if ok2:
self._refresh()
else:
QMessageBox.critical(self, "创建失败", msg)
def _delete(self):
conn = self._conn()
if not conn:
return
rp = self._selected_remote_path()
if not rp:
return
if QMessageBox.question(
self, "确认删除",
f"确定要删除 {rp} 吗？\n目录将递归删除需用命令执行。",
QMessageBox.Yes | QMessageBox.No
) != QMessageBox.Yes:
return
ok, msg = conn.remove(rp)
if ok:
self._refresh()
else:
QMessageBox.critical(self, "删除失败", msg)
def _rename(self):
conn = self._conn()
if not conn:
return
rp = self._selected_remote_path()
if not rp:
return
new_name, ok = QInputDialog.getText(
self, "重命名", "新名称:", text=os.path.basename(rp))
if not ok or not new_name.strip():
return
new_path = self.current_path.rstrip("/") + "/" + new_name.strip()
ok2, msg = conn.rename(rp, new_path)
if ok2:
self._refresh()
else:
QMessageBox.critical(self, "重命名失败", msg)
# ============================================================
# 监控面板
# ============================================================
class MonitorPanel(QWidget):
"""实时监控：CPU、内存、磁盘、网络"""
def __init__(self, manager: ConnectionManager, parent=None):
super().__init__(parent)
self.manager = manager
self.current_host_id: Optional[str] = None
self.worker: Optional[MonitorWorker] = None
self._last_net: Dict[str, tuple] = {} # iface -> (rx, ts)
self._build_ui()
def _build_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(10)
# 顶部：主机信息 + 控制
head = QHBoxLayout()
self.host_label = QLabel("未连接")
self.host_label.setStyleSheet("font-size: 14pt; font-weight: bold;")
head.addWidget(self.host_label)
head.addStretch(1)
self.interval_combo = QComboBox()
self.interval_combo.addItems(["1 秒", "2 秒", "3 秒", "5 秒", "10 秒"])
self.interval_combo.setCurrentIndex(2)
head.addWidget(QLabel("刷新:"))
head.addWidget(self.interval_combo)
self.btn_toggle = QPushButton("开始监控")
self.btn_toggle.setCheckable(True)
self.btn_toggle.toggled.connect(self._on_toggle)
head.addWidget(self.btn_toggle)
layout.addLayout(head)
# 主机信息
self.info_label = QLabel("-")
self.info_label.setStyleSheet("color: #666;")
layout.addWidget(self.info_label)
# CPU + 内存行
grid = QHBoxLayout()
grid.addWidget(self._build_card("CPU 使用率", "cpu_card"))
grid.addWidget(self._build_card("内存", "mem_card"))
layout.addLayout(grid)
# 负载 + 启动时间
grid2 = QHBoxLayout()
grid2.addWidget(self._build_card("系统负载", "load_card"))
grid2.addWidget(self._build_card("启动时间", "uptime_card"))
layout.addLayout(grid2)
# 磁盘 + 网络
grid3 = QHBoxLayout()
self.disk_group = self._build_table_card("磁盘", ["挂载点", "已用/总大小", "使用率", "进度"])
self.net_group = self._build_table_card("网络", ["网卡", "↓ 接收", "↑ 发送", "速率"])
grid3.addWidget(self.disk_group)
grid3.addWidget(self.net_group)
layout.addLayout(grid3, 1)
def _build_card(self, title: str, name: str) -> QGroupBox:
box = QGroupBox(title)
v = QVBoxLayout(box)
big = QLabel("0%")
big.setObjectName(f"{name}_value")
big.setStyleSheet("font-size: 28pt; font-weight: bold;")
big.setAlignment(Qt.AlignCenter)
v.addWidget(big)
sub = QLabel("-")
sub.setObjectName(f"{name}_sub")
sub.setAlignment(Qt.AlignCenter)
sub.setStyleSheet("color: #888;")
v.addWidget(sub)
return box
def _build_table_card(self, title: str, headers: list) -> QGroupBox:
box = QGroupBox(title)
v = QVBoxLayout(box)
table = QTableWidget(0, len(headers))
table.setHorizontalHeaderLabels(headers)
table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
table.verticalHeader().setVisible(False)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
v.addWidget(table)
return box
def _value_label(self, name: str) -> QLabel:
return self.findChild(QLabel, f"{name}_value")
def _sub_label(self, name: str) -> QLabel:
return self.findChild(QLabel, f"{name}_sub")
def set_host(self, host_id: str):
"""切换主机时停止旧监控并刷新 UI"""
self._stop_worker()
self.current_host_id = host_id
if not host_id:
self.host_label.setText("未连接")
return
h = self.manager.get_host(host_id)
if h:
self.host_label.setText(f"{h.get('name', h.get('host'))} ({h.get('host')})")
if self.btn_toggle.isChecked():
self._start_worker()
def _on_toggle(self, checked: bool):
if checked:
self.btn_toggle.setText("停止监控")
self._start_worker()
else:
self.btn_toggle.setText("开始监控")
self._stop_worker()
def _start_worker(self):
if not self.current_host_id:
return
conn = self.manager.get_connection(self.current_host_id)
if not conn or not conn.connected:
self.info_label.setText("⚠ 当前主机未连接")
return
idx = self.interval_combo.currentIndex()
interval = [1, 2, 3, 5, 10][idx]
self._stop_worker()
self.worker = MonitorWorker(conn, interval=interval)
self.worker.sample_ready.connect(self._on_sample)
self.worker.error.connect(lambda m: self.info_label.setText(f"⚠ {m}"))
self.worker.start()
self.info_label.setText(f"已启动监控，每 {interval} 秒刷新")
def _stop_worker(self):
if self.worker:
self.worker.stop()
self.worker.wait(2000)
self.worker = None
self._last_net.clear()
def _on_sample(self, m: dict):
if m.get("error"):
self.info_label.setText(f"⚠ {m['error']}")
return
# 主机
os_info = m.get("os", "")
krn = m.get("kernel", "")
host = m.get("hostname", "")
self.info_label.setText(f"主机: {host} · 系统: {os_info} · 内核: {krn}")
# CPU
cpu = m.get("cpu", 0)
self._value_label("cpu_card").setText(f"{cpu:.1f}%")
cores = m.get("cores", 1)
self._sub_label("cpu_card").setText(f"{cores} 核 CPU")
# 内存
mp = m.get("mem_percent", 0)
mu = SystemMonitor.format_bytes(m.get("mem_used", 0))
mt = SystemMonitor.format_bytes(m.get("mem_total", 0))
self._value_label("mem_card").setText(f"{mp:.1f}%")
self._sub_label("mem_card").setText(f"{mu} / {mt}")
# 负载
l1, l5, l15 = m.get("load1", 0), m.get("load5", 0), m.get("load15", 0)
self._value_label("load_card").setText(f"{l1:.2f}")
self._sub_label("load_card").setText(
f"5分钟: {l5:.2f} · 15分钟: {l15:.2f} (核心数={cores})"
)
# 启动时间
ut = SystemMonitor.format_uptime(m.get("uptime", 0))
self._value_label("uptime_card").setText(ut)
boot_ts = time.time() - m.get("uptime", 0)
self._sub_label("uptime_card").setText(
f"启动于: {time.strftime('%Y-%m-%d %H:%M', time.localtime(boot_ts))}"
)
# 磁盘表
disks = m.get("disks", [])
disk_table: QTableWidget = self.disk_group.findChild(QTableWidget)
disk_table.setRowCount(len(disks))
for i, d in enumerate(disks):
disk_table.setItem(i, 0, QTableWidgetItem(d["mount"]))
used_s = SystemMonitor.format_bytes(d["used"])
total_s = SystemMonitor.format_bytes(d["total"])
disk_table.setItem(i, 1, QTableWidgetItem(f"{used_s} / {total_s}"))
pct = d["percent"]
disk_table.setItem(i, 2, QTableWidgetItem(f"{pct}%"))
bar = QProgressBar()
bar.setValue(min(pct, 100))
bar.setFormat(f"{pct}%")
disk_table.setCellWidget(i, 3, bar)
# 网络表（计算每秒速率）
nets = m.get("net", [])
net_table: QTableWidget = self.net_group.findChild(QTableWidget)
net_table.setRowCount(len(nets))
now = time.time()
for i, n in enumerate(nets):
rx, tx, iface = n["rx"], n["tx"], n["iface"]
rx_s, tx_s = "-", "-"
if iface in self._last_net:
last_rx, last_tx, last_ts = self._last_net[iface]
dt = max(now - last_ts, 0.001)
rx_speed = (rx - last_rx) / dt
tx_speed = (tx - last_tx) / dt
rx_s = f"{SystemMonitor.format_bytes(rx)} ({SystemMonitor.format_bytes(int(rx_speed))}/s)"
tx_s = f"{SystemMonitor.format_bytes(tx)} ({SystemMonitor.format_bytes(int(tx_speed))}/s)"
else:
rx_s = f"{SystemMonitor.format_bytes(rx)}"
tx_s = f"{SystemMonitor.format_bytes(tx)}"
self._last_net[iface] = (rx, tx, now)
net_table.setItem(i, 0, QTableWidgetItem(iface))
net_table.setItem(i, 1, QTableWidgetItem(rx_s))
net_table.setItem(i, 2, QTableWidgetItem(tx_s))
if iface in self._last_net and len(self._last_net[iface]) == 3:
last_rx, last_tx, last_ts = self._last_net[iface]
dt = max(now - last_ts, 0.001)
rx_speed = (rx - last_rx) / dt
tx_speed = (tx - last_tx) / dt
net_table.setItem(i, 3, QTableWidgetItem(
f"↓{SystemMonitor.format_bytes(int(rx_speed))}/s ↑{SystemMonitor.format_bytes(int(tx_speed))}/s"))
else:
net_table.setItem(i, 3, QTableWidgetItem("采样中..."))
# ============================================================
# AI Agent 对话面板
# ============================================================
class AIChatPanel(QWidget):
"""AI Agent 对话面板"""
def __init__(self, manager: ConnectionManager, agent, parent=None):
super().__init__(parent)
self.manager = manager
self.agent = agent
self.worker: Optional[AIWorker] = None
self.current_host_id: Optional[str] = None
self._build_ui()
def _build_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(8, 8, 8, 8)
# 顶部状态
head = QHBoxLayout()
self.status_label = QLabel("AI Agent: 未配置")
self.status_label.setStyleSheet("color: #888;")
head.addWidget(self.status_label)
head.addStretch(1)
self.btn_config = QPushButton("⚙ AI 设置")
self.btn_config.clicked.connect(self._show_config)
head.addWidget(self.btn_config)
self.btn_clear = QPushButton("清空对话")
self.btn_clear.clicked.connect(self._clear)
head.addWidget(self.btn_clear)
layout.addLayout(head)
# 快捷指令
quick = QHBoxLayout()
quick.addWidget(QLabel("快捷:"))
for label, prompt in [
("检查状态", "帮我检查当前主机的运行状态，包括 CPU/内存/磁盘/网络"),
("查日志", "查看最近 100 行系统日志，重点关注 error 和 warning"),
("找大文件", "列出 /var/log 目录下最大的 10 个文件"),
("查端口", "查看当前监听的端口以及对应进程"),
]:
b = QPushButton(label)
b.clicked.connect(lambda _, p=prompt: self.input_edit.setText(p))
quick.addWidget(b)
quick.addStretch(1)
layout.addLayout(quick)
# 对话显示
self.chat_view = QTextEdit()
self.chat_view.setReadOnly(True)
self.chat_view.setStyleSheet("""
QTextEdit {
background: #1e1e1e;
color: #e0e0e0;
font-family: Consolas, 'Courier New', monospace;
font-size: 10pt;
}
""")
layout.addWidget(self.chat_view, 1)
# 输入区
input_layout = QHBoxLayout()
self.input_edit = QLineEdit()
self.input_edit.setPlaceholderText("输入问题，回车发送（Shift+回车换行）...")
self.input_edit.returnPressed.connect(self._send)
self.btn_send = QPushButton("发送")
self.btn_send.clicked.connect(self._send)
input_layout.addWidget(self.input_edit, 1)
input_layout.addWidget(self.btn_send)
layout.addLayout(input_layout)
self._append("系统", "AI Agent 已就绪。请先点击「⚙ AI 设置」配置 API Key。")
def set_host(self, host_id: str):
self.current_host_id = host_id
if host_id:
h = self.manager.get_host(host_id)
if h:
self._append("系统", f"已切换到主机: {h.get('name', h.get('host'))}")
def _append(self, role: str, content: str, color: str = ""):
ts = datetime.now().strftime("%H:%M:%S")
color_map = {
"user": "#4fc3f7", "assistant": "#aed581", "assistant_thinking": "#ffb74d",
"tool_call": "#ba68c8", "tool_result": "#90a4ae", "error": "#ef5350",
"system": "#888",
}
c = color or color_map.get(role, "#e0e0e0")
safe = (content or "").replace("&", "&").replace("<", "<").replace(">", ">")
safe = safe.replace("\n", "
")
role_label_map = {
"user": "我", "assistant": "AI", "assistant_thinking": "AI 思考",
"tool_call": "调用", "tool_result": "结果", "error": "错误", "system": "系统",
}
rl = role_label_map.get(role, role)
self.chat_view.append(
f'[{ts}] '
f'{rl}: '
f'{safe}'
)
# 滚动到底
sb = self.chat_view.verticalScrollBar()
sb.setValue(sb.maximum())
def _send(self):
if self.worker and self.worker.isRunning():
return
text = self.input_edit.text().strip()
if not text:
return
if not self.agent.api_key:
QMessageBox.warning(self, "未配置", "请先在「⚙ AI 设置」中配置 API Key")
return
self._append("user", text)
self.input_edit.clear()
self.btn_send.setEnabled(False)
self.btn_send.setText("思考中...")
conn = self.manager.get_connection(self.current_host_id) if self.current_host_id else None
self.worker = AIWorker(self.agent, text, conn)
self.worker.step.connect(self._append)
self.worker.finished_with.connect(self._on_done)
self.worker.error.connect(self._on_error)
self.worker.start()
def _on_done(self, final: str):
self._append("assistant", final)
self.btn_send.setEnabled(True)
self.btn_send.setText("发送")
def _on_error(self, msg: str):
self._append("error", msg)
self.btn_send.setEnabled(True)
self.btn_send.setText("发送")
def _clear(self):
self.agent.clear_history()
self.chat_view.clear()
self._append("系统", "对话历史已清空")
def _show_config(self):
from .config_dialog import AIConfigDialog
dlg = AIConfigDialog(self.agent, self)
if dlg.exec_():
self.status_label.setText(
f"AI Agent: {self.agent.model} @ {self.agent.base_url}")
self._append("系统", "AI 配置已更新")