import json import os from loguru import logger from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QComboBox, QMessageBox) from PySide6.QtCore import Qt, Signal # 添加Signal导入 import paramiko from threads import SSHConnectionThread class ServerConnectionTab(QWidget): # 添加服务器切换信号 server_changed = Signal(dict) def __init__(self): super().__init__() self.ssh_client = None self.config = self.load_config() self.init_ui() def load_config(self): config_file = "config.json" if os.path.exists(config_file): try: with open(config_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"读取配置文件失败: {e}") return {"servers": []} return {"servers": []} def save_config(self): try: with open("config.json", 'w', encoding='utf-8') as f: json.dump(self.config, f, ensure_ascii=False, indent=2) logger.info("配置文件保存成功") except Exception as e: logger.error(f"保存配置文件失败: {e}") QMessageBox.critical(self, "错误", f"保存配置文件失败: {e}") def init_ui(self): layout = QVBoxLayout() # 服务器选择 server_layout = QHBoxLayout() server_layout.addWidget(QLabel("服务器别名:")) self.server_combo = QComboBox() self.server_combo.currentTextChanged.connect(self.on_server_selected) server_layout.addWidget(self.server_combo) layout.addLayout(server_layout) # 服务器信息输入 form_layout = QVBoxLayout() # IP地址 ip_layout = QHBoxLayout() ip_layout.addWidget(QLabel("IP地址:")) self.ip_input = QLineEdit() ip_layout.addWidget(self.ip_input) form_layout.addLayout(ip_layout) # 端口 port_layout = QHBoxLayout() port_layout.addWidget(QLabel("端口:")) self.port_input = QLineEdit("22") port_layout.addWidget(self.port_input) form_layout.addLayout(port_layout) # 用户名 username_layout = QHBoxLayout() username_layout.addWidget(QLabel("用户名:")) self.username_input = QLineEdit() username_layout.addWidget(self.username_input) form_layout.addLayout(username_layout) # 密码 password_layout = QHBoxLayout() password_layout.addWidget(QLabel("密码:")) self.password_input = QLineEdit() self.password_input.setEchoMode(QLineEdit.Password) password_layout.addWidget(self.password_input) form_layout.addLayout(password_layout) # 项目名称 project_layout = QHBoxLayout() project_layout.addWidget(QLabel("项目名称:")) self.project_input = QLineEdit() project_layout.addWidget(self.project_input) form_layout.addLayout(project_layout) # Git仓库URL git_url_layout = QHBoxLayout() git_url_layout.addWidget(QLabel("Git仓库URL:")) self.git_url_input = QLineEdit() self.git_url_input.setPlaceholderText("https://github.com/username/repo.git") git_url_layout.addWidget(self.git_url_input) form_layout.addLayout(git_url_layout) # 远程目录 remote_dir_layout = QHBoxLayout() remote_dir_layout.addWidget(QLabel("远程目录:")) self.remote_dir_input = QLineEdit() self.remote_dir_input.setPlaceholderText("/home/user/project") remote_dir_layout.addWidget(self.remote_dir_input) form_layout.addLayout(remote_dir_layout) # Django项目路径 django_path_layout = QHBoxLayout() django_path_layout.addWidget(QLabel("Django项目路径:")) self.django_path_input = QLineEdit() self.django_path_input.setPlaceholderText("/home/user/django_project") django_path_layout.addWidget(self.django_path_input) form_layout.addLayout(django_path_layout) layout.addLayout(form_layout) # 按钮 button_layout = QHBoxLayout() self.save_button = QPushButton("保存配置") self.save_button.clicked.connect(self.save_server_config) self.connect_button = QPushButton("连接服务器") self.connect_button.clicked.connect(self.connect_server) button_layout.addWidget(self.save_button) button_layout.addWidget(self.connect_button) layout.addLayout(button_layout) # 连接状态 self.status_label = QLabel("未连接") self.status_label.setAlignment(Qt.AlignCenter) layout.addWidget(self.status_label) layout.addStretch() self.setLayout(layout) # 加载服务器列表 self.load_server_list() def load_server_list(self): self.server_combo.clear() servers = self.config.get("servers", []) for server in servers: self.server_combo.addItem(server.get("alias", "")) def on_server_selected(self, alias): if not alias: return servers = self.config.get("servers", []) for server in servers: if server.get("alias") == alias: self.ip_input.setText(server.get("host", "")) self.port_input.setText(str(server.get("port", 22))) self.username_input.setText(server.get("username", "")) self.password_input.setText(server.get("password", "")) self.project_input.setText(server.get("project_name", "")) self.git_url_input.setText(server.get("git_url", "")) self.remote_dir_input.setText(server.get("remote_directory", "")) self.django_path_input.setText(server.get("django_path", "")) # 发送服务器切换信号 self.server_changed.emit(server) break def save_server_config(self): alias = self.server_combo.currentText() if not alias: alias = f"服务器_{len(self.config.get('servers', [])) + 1}" self.server_combo.addItem(alias) server_config = { "alias": alias, "host": self.ip_input.text(), "port": int(self.port_input.text()) if self.port_input.text() else 22, "username": self.username_input.text(), "password": self.password_input.text(), "project_name": self.project_input.text(), "git_url": self.git_url_input.text(), "remote_directory": self.remote_dir_input.text(), "django_path": self.django_path_input.text() } servers = self.config.get("servers", []) # 更新或添加服务器配置 found = False for i, server in enumerate(servers): if server.get("alias") == alias: servers[i] = server_config found = True break if not found: servers.append(server_config) self.config["servers"] = servers self.save_config() QMessageBox.information(self, "成功", "配置已保存") def connect_server(self): host = self.ip_input.text() port = int(self.port_input.text()) if self.port_input.text() else 22 username = self.username_input.text() password = self.password_input.text() if not all([host, username, password]): QMessageBox.warning(self, "警告", "请填写完整的连接信息") return self.status_label.setText("正在连接...") self.connect_button.setEnabled(False) self.connection_thread = SSHConnectionThread(host, username, password, port) self.connection_thread.connection_status.connect(self.on_connection_result) self.connection_thread.start() def on_connection_result(self, success, message): self.connect_button.setEnabled(True) if success: self.status_label.setText(f"连接成功: {message}") self.status_label.setStyleSheet("color: green;") # 创建SSH客户端实例供其他标签使用 self.ssh_client = paramiko.SSHClient() self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh_client.connect(self.ip_input.text(), port=int(self.port_input.text()), username=self.username_input.text(), password=self.password_input.text()) else: self.status_label.setText(f"连接失败: {message}") self.status_label.setStyleSheet("color: red;") self.ssh_client = None def get_current_config(self): """获取当前选中的服务器配置""" try: alias = self.server_combo.currentText() if not alias: return None servers = self.config.get("servers", []) for server in servers: if server.get("alias") == alias: return server return None except Exception as e: logger.error(f"获取当前服务器配置失败: {e}") return None