306 lines
12 KiB
Python
306 lines
12 KiB
Python
import os
|
||
import json
|
||
from loguru import logger
|
||
from PySide6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit,
|
||
QPushButton, QComboBox, QMessageBox, QTextEdit,
|
||
QGroupBox, QGridLayout)
|
||
from PySide6.QtCore import Qt
|
||
|
||
from threads import (GitInstallThread, GitCloneThread, ListDirectoryThread,
|
||
SetTimezoneAndRestartThread,
|
||
CheckFirewallThread, OpenPortThread)
|
||
|
||
|
||
class RemoteCommandTab(QWidget):
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.parent = parent
|
||
self.init_ui()
|
||
|
||
def init_ui(self):
|
||
layout = QVBoxLayout()
|
||
|
||
# Git管理组
|
||
git_group = QGroupBox("Git代码管理")
|
||
git_layout = QGridLayout()
|
||
|
||
git_layout.addWidget(QLabel("Git仓库URL:"), 0, 0)
|
||
self.git_url_input = QLineEdit()
|
||
self.git_url_input.setPlaceholderText("https://github.com/username/repo.git")
|
||
git_layout.addWidget(self.git_url_input, 0, 1)
|
||
|
||
git_layout.addWidget(QLabel("项目路径:"), 1, 0)
|
||
self.project_path_input = QLineEdit()
|
||
self.project_path_input.setPlaceholderText("/home/user/project")
|
||
git_layout.addWidget(self.project_path_input, 1, 1)
|
||
|
||
self.install_git_btn = QPushButton("安装Git")
|
||
self.install_git_btn.clicked.connect(self.install_git)
|
||
git_layout.addWidget(self.install_git_btn, 0, 2)
|
||
|
||
self.clone_btn = QPushButton("拉取代码")
|
||
self.clone_btn.clicked.connect(self.clone_git)
|
||
git_layout.addWidget(self.clone_btn, 1, 2)
|
||
|
||
self.list_dir_btn = QPushButton("列出目录")
|
||
self.list_dir_btn.clicked.connect(self.list_directory)
|
||
git_layout.addWidget(self.list_dir_btn, 2, 2)
|
||
|
||
self.set_timezone_btn = QPushButton("设置时区并重启")
|
||
self.set_timezone_btn.clicked.connect(self.set_timezone_and_restart)
|
||
git_layout.addWidget(self.set_timezone_btn, 3, 0, 1, 2)
|
||
|
||
git_group.setLayout(git_layout)
|
||
layout.addWidget(git_group)
|
||
|
||
# 防火墙管理组
|
||
firewall_group = QGroupBox("防火墙管理")
|
||
firewall_layout = QGridLayout()
|
||
|
||
self.check_firewall_btn = QPushButton("检查防火墙")
|
||
self.check_firewall_btn.clicked.connect(self.check_firewall)
|
||
firewall_layout.addWidget(self.check_firewall_btn, 0, 0)
|
||
|
||
firewall_layout.addWidget(QLabel("端口:"), 0, 1)
|
||
self.port_input = QLineEdit("8000")
|
||
firewall_layout.addWidget(self.port_input, 0, 2)
|
||
|
||
self.open_port_btn = QPushButton("开放端口")
|
||
self.open_port_btn.clicked.connect(self.open_port)
|
||
firewall_layout.addWidget(self.open_port_btn, 0, 3)
|
||
|
||
firewall_group.setLayout(firewall_layout)
|
||
layout.addWidget(firewall_group)
|
||
|
||
# 操作输出
|
||
output_group = QGroupBox("操作输出")
|
||
output_layout = QVBoxLayout()
|
||
|
||
self.output_text = QTextEdit()
|
||
self.output_text.setReadOnly(True)
|
||
self.output_text.setPlaceholderText("操作结果将在这里显示...")
|
||
output_layout.addWidget(self.output_text)
|
||
|
||
output_group.setLayout(output_layout)
|
||
layout.addWidget(output_group)
|
||
|
||
layout.addStretch()
|
||
self.setLayout(layout)
|
||
|
||
# 从配置文件加载git配置
|
||
self.load_git_config()
|
||
|
||
def load_git_config(self):
|
||
try:
|
||
# 从config.json文件读取配置
|
||
config_path = os.path.join(os.path.dirname(__file__), 'config.json')
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
|
||
# 获取第一个服务器的配置
|
||
if config and 'servers' in config and len(config['servers']) > 0:
|
||
server_config = config['servers'][0]
|
||
git_url = server_config.get('git_url', '')
|
||
project_path = server_config.get('remote_directory', '')
|
||
|
||
self.git_url_input.setText(git_url)
|
||
self.project_path_input.setText(project_path)
|
||
|
||
logger.info(f"从配置文件加载git配置: git_url={git_url}, project_path={project_path}")
|
||
else:
|
||
logger.warning("配置文件中未找到服务器配置")
|
||
except Exception as e:
|
||
logger.error(f"加载git配置失败: {str(e)}")
|
||
QMessageBox.warning(self, "警告", f"加载git配置失败: {str(e)}")
|
||
|
||
def install_git(self):
|
||
if not self.check_ssh_connection():
|
||
return
|
||
|
||
self.output_text.append("正在安装Git...")
|
||
self.install_git_btn.setEnabled(False)
|
||
|
||
self.git_install_thread = GitInstallThread(self.parent.ssh_client)
|
||
self.git_install_thread.result_ready.connect(self.on_git_install_result)
|
||
self.git_install_thread.start()
|
||
|
||
def on_git_install_result(self, success, message):
|
||
self.install_git_btn.setEnabled(True)
|
||
if success:
|
||
self.output_text.append(f"Git安装成功: {message}")
|
||
logger.info(f"Git安装成功: {message}")
|
||
else:
|
||
self.output_text.append(f"Git安装失败: {message}")
|
||
logger.error(f"Git安装失败: {message}")
|
||
|
||
def clone_git(self):
|
||
"""克隆Git仓库到远程服务器"""
|
||
if not self.check_ssh_connection():
|
||
return
|
||
|
||
git_url = self.git_url_input.text().strip()
|
||
if not git_url:
|
||
QMessageBox.warning(self, "警告", "请输入Git仓库地址")
|
||
return
|
||
|
||
# 获取当前服务器配置
|
||
config = None
|
||
if self.parent and hasattr(self.parent, 'server_connection_tab'):
|
||
config = self.parent.server_connection_tab.get_current_config()
|
||
|
||
if config:
|
||
# 使用remote_directory而不是django_path
|
||
target_path = config.get('remote_directory', '/home/user')
|
||
project_name = config.get('project_name', 'myproject')
|
||
else:
|
||
# 如果没有config，使用默认值
|
||
target_path = '/home/user'
|
||
project_name = 'myproject'
|
||
|
||
# 获取密码
|
||
password = self.get_password()
|
||
if password is None:
|
||
return
|
||
|
||
self.output_text.append(f"正在克隆Git仓库 {git_url} 到 {target_path}...")
|
||
self.clone_git_btn.setEnabled(False)
|
||
self.progress_bar.setVisible(True)
|
||
self.progress_bar.setValue(0)
|
||
|
||
self.clone_thread = CloneGitThread(self.parent.ssh_client, git_url, target_path, project_name, password)
|
||
self.clone_thread.progress_updated.connect(self.update_progress)
|
||
self.clone_thread.result_ready.connect(self.on_clone_git_result)
|
||
self.clone_thread.start()
|
||
|
||
def on_git_clone_result(self, success, message):
|
||
self.clone_btn.setEnabled(True)
|
||
if success:
|
||
self.output_text.append(f"Git克隆成功: {message}")
|
||
logger.info(f"Git克隆成功: {message}")
|
||
else:
|
||
self.output_text.append(f"Git克隆失败: {message}")
|
||
logger.error(f"Git克隆失败: {message}")
|
||
|
||
def list_directory(self):
|
||
if not self.check_ssh_connection():
|
||
return
|
||
|
||
path = self.project_path_input.text().strip()
|
||
if not path:
|
||
QMessageBox.warning(self, "警告", "请输入要列出的目录路径")
|
||
return
|
||
|
||
self.output_text.append(f"正在列出目录 {path}...")
|
||
self.list_dir_btn.setEnabled(False)
|
||
|
||
self.list_dir_thread = ListDirectoryThread(self.parent.ssh_client, path)
|
||
self.list_dir_thread.result_ready.connect(self.on_list_directory_result)
|
||
self.list_dir_thread.start()
|
||
|
||
def on_list_directory_result(self, success, message):
|
||
self.list_dir_btn.setEnabled(True)
|
||
if success:
|
||
self.output_text.append(f"目录列表:\n{message}")
|
||
logger.info(f"目录列表成功:\n{message}")
|
||
else:
|
||
self.output_text.append(f"列出目录失败: {message}")
|
||
logger.error(f"列出目录失败: {message}")
|
||
|
||
def set_timezone_and_restart(self):
|
||
if not self.check_ssh_connection():
|
||
return
|
||
|
||
# 获取密码
|
||
password = self.get_password()
|
||
if password is None:
|
||
return
|
||
|
||
self.output_text.append("正在设置时区为Asia/Shanghai并重启服务器...")
|
||
self.set_timezone_btn.setEnabled(False)
|
||
|
||
self.timezone_thread = SetTimezoneAndRestartThread(self.parent.ssh_client, password)
|
||
self.timezone_thread.result_ready.connect(self.on_set_timezone_and_restart_result)
|
||
self.timezone_thread.start()
|
||
|
||
def on_set_timezone_and_restart_result(self, success, message):
|
||
self.set_timezone_btn.setEnabled(True)
|
||
if success:
|
||
self.output_text.append(f"时区设置成功: {message}")
|
||
logger.info(f"时区设置成功: {message}")
|
||
else:
|
||
self.output_text.append(f"时区设置失败: {message}")
|
||
logger.error(f"时区设置失败: {message}")
|
||
|
||
def check_firewall(self):
|
||
if not self.check_ssh_connection():
|
||
return
|
||
|
||
# 获取密码
|
||
password = self.get_password()
|
||
if password is None:
|
||
return
|
||
|
||
self.output_text.append("正在检查防火墙状态...")
|
||
self.check_firewall_btn.setEnabled(False)
|
||
|
||
self.firewall_thread = CheckFirewallThread(self.parent.ssh_client, password)
|
||
self.firewall_thread.result_ready.connect(self.on_check_firewall_result)
|
||
self.firewall_thread.start()
|
||
|
||
def on_check_firewall_result(self, success, message):
|
||
self.check_firewall_btn.setEnabled(True)
|
||
if success:
|
||
self.output_text.append(f"防火墙状态:\n{message}")
|
||
logger.info(f"防火墙状态检查成功")
|
||
else:
|
||
self.output_text.append(f"防火墙状态检查失败: {message}")
|
||
logger.error(f"防火墙状态检查失败: {message}")
|
||
|
||
def open_port(self):
|
||
if not self.check_ssh_connection():
|
||
return
|
||
|
||
port = self.port_input.text().strip()
|
||
if not port:
|
||
QMessageBox.warning(self, "警告", "请输入要开放的端口号")
|
||
return
|
||
|
||
# 获取密码
|
||
password = self.get_password()
|
||
if password is None:
|
||
return
|
||
|
||
self.output_text.append(f"正在开放端口 {port}...")
|
||
self.open_port_btn.setEnabled(False)
|
||
|
||
self.open_port_thread = OpenPortThread(self.parent.ssh_client, port, password)
|
||
self.open_port_thread.result_ready.connect(self.on_open_port_result)
|
||
self.open_port_thread.start()
|
||
|
||
def on_open_port_result(self, success, message):
|
||
self.open_port_btn.setEnabled(True)
|
||
if success:
|
||
self.output_text.append(f"端口开放成功: {message}")
|
||
logger.info(f"端口开放成功: {message}")
|
||
else:
|
||
self.output_text.append(f"端口开放失败: {message}")
|
||
logger.error(f"端口开放失败: {message}")
|
||
|
||
def get_password(self):
|
||
"""获取服务器连接密码"""
|
||
if not self.parent or not hasattr(self.parent, 'password_input'):
|
||
QMessageBox.warning(self, "警告", "无法获取服务器密码")
|
||
return None
|
||
|
||
password = self.parent.password_input.text()
|
||
if not password:
|
||
QMessageBox.warning(self, "警告", "请先输入服务器密码")
|
||
return None
|
||
|
||
return password
|
||
|
||
def check_ssh_connection(self):
|
||
if not self.parent or not self.parent.ssh_client:
|
||
QMessageBox.warning(self, "警告", "请先连接服务器")
|
||
return False
|
||
return True |