364 lines
16 KiB
Python
364 lines
16 KiB
Python
import os
|
||
import sys
|
||
import time
|
||
from PySide6.QtCore import QThread, Signal
|
||
from loguru import logger
|
||
|
||
class DjangoInstallThread(QThread):
|
||
"""安装Django的线程"""
|
||
result_ready = Signal(bool, str)
|
||
progress_updated = Signal(int)
|
||
|
||
def __init__(self, ssh_client, password):
|
||
super().__init__()
|
||
self.ssh_client = ssh_client
|
||
self.password = password
|
||
|
||
def run(self):
|
||
try:
|
||
self.progress_updated.emit(10)
|
||
|
||
# 检查Django是否已安装
|
||
stdin, stdout, stderr = self.ssh_client.exec_command("python3 -m django --version")
|
||
django_version = stdout.read().decode().strip()
|
||
|
||
if django_version:
|
||
self.result_ready.emit(True, f"Django已安装: {django_version}")
|
||
logger.info(f"Django已安装: {django_version}")
|
||
return
|
||
|
||
self.progress_updated.emit(30)
|
||
|
||
# 尝试使用pip安装
|
||
stdin, stdout, stderr = self.ssh_client.exec_command("pip3 install --break-system-packages django")
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
|
||
if exit_status == 0:
|
||
self.progress_updated.emit(90)
|
||
stdin, stdout, stderr = self.ssh_client.exec_command("python3 -m django --version")
|
||
django_version = stdout.read().decode().strip()
|
||
self.result_ready.emit(True, f"Django安装成功: {django_version}")
|
||
logger.info(f"Django安装成功: {django_version}")
|
||
return
|
||
|
||
self.progress_updated.emit(50)
|
||
|
||
# 如果pip安装失败，尝试使用apt安装，使用-S选项从标准输入读取密码
|
||
stdin, stdout, stderr = self.ssh_client.exec_command(f"echo '{self.password}' | sudo -S apt update && echo '{self.password}' | sudo -S apt install -y python3-django")
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
|
||
if exit_status == 0:
|
||
self.progress_updated.emit(90)
|
||
stdin, stdout, stderr = self.ssh_client.exec_command("python3 -m django --version")
|
||
django_version = stdout.read().decode().strip()
|
||
self.result_ready.emit(True, f"Django安装成功: {django_version}")
|
||
logger.info(f"Django安装成功: {django_version}")
|
||
else:
|
||
error = stderr.read().decode()
|
||
self.result_ready.emit(False, f"Django安装失败: {error}")
|
||
logger.error(f"Django安装失败: {error}")
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
self.result_ready.emit(False, error_msg)
|
||
logger.error(f"Django安装异常: {error_msg}")
|
||
|
||
class DjangoCommandThread(QThread):
|
||
"""执行Django相关命令的线程"""
|
||
output_signal = Signal(str)
|
||
finished_signal = Signal()
|
||
password_request_signal = Signal() # 请求密码的信号
|
||
|
||
def __init__(self, ssh_client, command, log_file=None):
|
||
super().__init__()
|
||
self.ssh_client = ssh_client
|
||
self.command = command
|
||
self.password = None
|
||
self.waiting_for_password = False
|
||
self.log_file = log_file # 日志文件路径，如果提供则将输出重定向到该文件
|
||
|
||
def set_password(self, password):
|
||
self.password = password
|
||
self.waiting_for_password = False
|
||
|
||
def run(self):
|
||
try:
|
||
logger.info(f"执行Django命令: {self.command}")
|
||
|
||
# 如果提供了日志文件路径，修改命令以重定向输出
|
||
if self.log_file:
|
||
self.command = f"{self.command} > {self.log_file} 2>&1"
|
||
self.output_signal.emit(f"命令输出将重定向到日志文件: {self.log_file}")
|
||
|
||
# 如果命令包含sudo，修改为使用-S选项从标准输入读取密码
|
||
if "sudo" in self.command:
|
||
command_with_sudo = self.command.replace("sudo", "sudo -S")
|
||
stdin, stdout, stderr = self.ssh_client.exec_command(command_with_sudo)
|
||
|
||
# 检查是否需要密码
|
||
password_prompt = False
|
||
for line in iter(stderr.readline, ""):
|
||
line_text = line.strip()
|
||
self.output_signal.emit(line_text)
|
||
if "password for" in line_text.lower() or "密码" in line_text:
|
||
password_prompt = True
|
||
break
|
||
|
||
# 如果需要密码，请求用户输入
|
||
if password_prompt:
|
||
self.waiting_for_password = True
|
||
self.password_request_signal.emit()
|
||
|
||
# 等待密码输入
|
||
while self.waiting_for_password:
|
||
self.msleep(100)
|
||
|
||
# 发送密码
|
||
if self.password:
|
||
stdin.write(self.password + "\n")
|
||
stdin.flush()
|
||
self.output_signal.emit("密码已发送")
|
||
else:
|
||
self.output_signal.emit("未提供密码，命令可能失败")
|
||
else:
|
||
stdin, stdout, stderr = self.ssh_client.exec_command(self.command)
|
||
|
||
# 对于apt命令，使用特殊处理以显示进度
|
||
if "apt install" in self.command:
|
||
# 使用非交互模式并显示进度
|
||
self.output_signal.emit("正在安装软件包，请稍候...")
|
||
|
||
# 读取输出和错误，实时显示
|
||
import select
|
||
import time
|
||
|
||
# 检查是否需要密码
|
||
password_sent = False
|
||
|
||
while not stdout.channel.exit_status_ready():
|
||
# 检查是否有数据可读
|
||
r, w, e = select.select([stdout.channel, stderr.channel], [], [], 0.1)
|
||
|
||
if stdout.channel in r:
|
||
output = stdout.channel.recv(1024).decode('utf-8', errors='replace')
|
||
if output:
|
||
self.output_signal.emit(output.strip())
|
||
logger.info(f"命令输出: {output.strip()}")
|
||
|
||
if stderr.channel in r:
|
||
error = stderr.channel.recv(1024).decode('utf-8', errors='replace')
|
||
if error:
|
||
# 检查是否需要密码
|
||
if ("password for" in error.lower() or "密码" in error) and not password_sent:
|
||
self.output_signal.emit("检测到需要输入密码")
|
||
self.waiting_for_password = True
|
||
self.password_request_signal.emit()
|
||
|
||
# 等待密码输入
|
||
while self.waiting_for_password:
|
||
self.msleep(100)
|
||
|
||
# 发送密码
|
||
if self.password:
|
||
stdin.write(self.password + "\n")
|
||
stdin.flush()
|
||
password_sent = True
|
||
self.output_signal.emit("密码已发送")
|
||
else:
|
||
self.output_signal.emit("未提供密码，命令可能失败")
|
||
else:
|
||
self.output_signal.emit(error.strip())
|
||
logger.error(f"命令错误: {error.strip()}")
|
||
|
||
# 短暂休眠以避免过度占用CPU
|
||
time.sleep(0.01)
|
||
|
||
# 读取剩余输出
|
||
while True:
|
||
r, w, e = select.select([stdout.channel, stderr.channel], [], [], 0.1)
|
||
if not r:
|
||
break
|
||
|
||
if stdout.channel in r:
|
||
output = stdout.channel.recv(1024).decode('utf-8', errors='replace')
|
||
if output:
|
||
self.output_signal.emit(output.strip())
|
||
logger.info(f"命令输出: {output.strip()}")
|
||
|
||
if stderr.channel in r:
|
||
error = stderr.channel.recv(1024).decode('utf-8', errors='replace')
|
||
if error:
|
||
self.output_signal.emit(error.strip())
|
||
logger.error(f"命令错误: {error.strip()}")
|
||
else:
|
||
# 对于非apt命令，使用原有的行读取方式
|
||
# 读取输出
|
||
for line in iter(stdout.readline, ""):
|
||
self.output_signal.emit(line.strip())
|
||
logger.info(f"命令输出: {line.strip()}")
|
||
|
||
# 读取错误
|
||
for line in iter(stderr.readline, ""):
|
||
line_text = line.strip()
|
||
if "password for" not in line_text.lower() and "密码" not in line_text: # 避免重复显示密码提示
|
||
self.output_signal.emit(f"错误: {line_text}")
|
||
logger.error(f"命令错误: {line_text}")
|
||
|
||
# 检查退出状态
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
if exit_status == 0:
|
||
self.output_signal.emit("命令执行成功")
|
||
logger.info(f"命令执行成功: {self.command}")
|
||
else:
|
||
self.output_signal.emit(f"命令执行失败，退出状态: {exit_status}")
|
||
logger.error(f"命令执行失败，退出状态: {exit_status}")
|
||
except Exception as e:
|
||
error_msg = f"执行命令时出错: {str(e)}"
|
||
self.output_signal.emit(error_msg)
|
||
logger.error(error_msg)
|
||
finally:
|
||
self.finished_signal.emit()
|
||
|
||
class UploadSettingsThread(QThread):
|
||
"""上传settings.py文件的线程"""
|
||
result_ready = Signal(bool, str)
|
||
password_request_signal = Signal() # 请求密码的信号
|
||
|
||
def __init__(self, ssh_client, django_path, settings_content):
|
||
super().__init__()
|
||
self.ssh_client = ssh_client
|
||
self.django_path = django_path
|
||
self.settings_content = settings_content
|
||
self.password = None
|
||
self.waiting_for_password = False
|
||
|
||
def set_password(self, password):
|
||
self.password = password
|
||
self.waiting_for_password = False
|
||
|
||
def run(self):
|
||
try:
|
||
# 查找settings.py文件
|
||
stdin, stdout, stderr = self.ssh_client.exec_command(f"find {self.django_path} -name settings.py")
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
|
||
if exit_status != 0:
|
||
self.result_ready.emit(False, "未找到settings.py文件")
|
||
logger.error("未找到settings.py文件")
|
||
return
|
||
|
||
settings_path = stdout.read().decode().strip()
|
||
|
||
# 创建临时文件
|
||
temp_file = "/tmp/settings_upload.py"
|
||
sftp = self.ssh_client.open_sftp()
|
||
|
||
with sftp.file(temp_file, 'w') as f:
|
||
f.write(self.settings_content)
|
||
|
||
# 在覆盖前备份原文件（带时间戳）
|
||
import datetime
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
backup_path = f"{settings_path}.backup_{timestamp}"
|
||
|
||
# 备份原文件
|
||
stdin, stdout, stderr = self.ssh_client.exec_command(f"sudo -S cp {settings_path} {backup_path}")
|
||
|
||
# 检查是否需要密码
|
||
password_sent = False
|
||
while True:
|
||
if stderr.channel.recv_ready():
|
||
error = stderr.channel.recv(1024).decode('utf-8', errors='replace')
|
||
if error:
|
||
# 检查是否需要密码
|
||
if ("password for" in error.lower() or "密码" in error) and not password_sent:
|
||
self.waiting_for_password = True
|
||
self.password_request_signal.emit()
|
||
|
||
# 等待密码输入
|
||
while self.waiting_for_password:
|
||
self.msleep(100)
|
||
|
||
# 发送密码
|
||
if self.password:
|
||
stdin.write(self.password + "\n")
|
||
stdin.flush()
|
||
password_sent = True
|
||
else:
|
||
self.result_ready.emit(False, "未提供密码，备份失败")
|
||
logger.error("未提供密码，备份失败")
|
||
sftp.close()
|
||
return
|
||
else:
|
||
logger.error(f"备份错误: {error.strip()}")
|
||
|
||
if stdout.channel.exit_status_ready():
|
||
break
|
||
|
||
# 短暂休眠以避免过度占用CPU
|
||
time.sleep(0.01)
|
||
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
|
||
if exit_status != 0:
|
||
error = stderr.read().decode()
|
||
self.result_ready.emit(False, f"备份原文件失败: {error}")
|
||
logger.error(f"备份原文件失败: {error}")
|
||
sftp.close()
|
||
return
|
||
|
||
logger.info(f"原文件已备份至: {backup_path}")
|
||
|
||
# 移动临时文件到目标位置
|
||
stdin, stdout, stderr = self.ssh_client.exec_command(f"sudo -S mv {temp_file} {settings_path}")
|
||
|
||
# 检查是否需要密码
|
||
password_sent = False
|
||
while True:
|
||
if stderr.channel.recv_ready():
|
||
error = stderr.channel.recv(1024).decode('utf-8', errors='replace')
|
||
if error:
|
||
# 检查是否需要密码
|
||
if ("password for" in error.lower() or "密码" in error) and not password_sent:
|
||
self.waiting_for_password = True
|
||
self.password_request_signal.emit()
|
||
|
||
# 等待密码输入
|
||
while self.waiting_for_password:
|
||
self.msleep(100)
|
||
|
||
# 发送密码
|
||
if self.password:
|
||
stdin.write(self.password + "\n")
|
||
stdin.flush()
|
||
password_sent = True
|
||
else:
|
||
self.result_ready.emit(False, "未提供密码，移动文件失败")
|
||
logger.error("未提供密码，移动文件失败")
|
||
sftp.close()
|
||
return
|
||
else:
|
||
logger.error(f"移动文件错误: {error.strip()}")
|
||
|
||
if stdout.channel.exit_status_ready():
|
||
break
|
||
|
||
# 短暂休眠以避免过度占用CPU
|
||
time.sleep(0.01)
|
||
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
|
||
if exit_status == 0:
|
||
self.result_ready.emit(True, f"settings.py上传成功，原文件已备份为: {os.path.basename(backup_path)}")
|
||
logger.info(f"settings.py上传成功，原文件已备份为: {os.path.basename(backup_path)}")
|
||
else:
|
||
error = stderr.read().decode()
|
||
self.result_ready.emit(False, error)
|
||
logger.error(f"settings.py上传失败: {error}")
|
||
|
||
sftp.close()
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
self.result_ready.emit(False, error_msg)
|
||
logger.error(f"settings.py上传异常: {error_msg}") |