bt-source/panel/script/free_total_update.py

# encoding utf-8
import subprocess
import requests
import os
import logging
import tempfile


class UpdateSiteTotal:
    # 类级别的常量定义，集中管理固定配置
    DEFAULT_USER_AGENT = "BT-Panel/10.0"
    INSTALL_SCRIPT_TIMEOUT = 600  # 安装脚本超时时间(秒)
    VERSION_CHECK_TIMEOUT = 10    # 版本检查超时时间(秒)
    SCRIPT_DOWNLOAD_TIMEOUT = 15  # 脚本下载超时时间(秒)

    site_total_path = '/www/server/site_total'
    download_url = 'https://download.bt.cn/site_total/'
    version_url = download_url + 'version.txt'
    install_sh = download_url + 'install.sh'
    site_total_bin = os.path.join(site_total_path, 'site_total')

    def __init__(self):
        """初始化更新工具
        """

        # 初始化日志配置
        self._init_logger()

    def _init_logger(self):
        """初始化日志配置（封装为独立方法，便于维护）"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )

    # ------------------------------
    # 版本相关方法
    # ------------------------------
    def get_current_version(self):
        """获取当前安装的版本号（返回字符串格式，如"1.0"）"""
        if not self.is_installed():
            self._log_warning(f"未检测到 {self.site_total_bin}，无法获取当前版本")
            return None

        try:
            # 执行版本命令
            result = subprocess.run(
                [self.site_total_bin, 'version'],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                check=True,
                timeout=self.VERSION_CHECK_TIMEOUT
            )
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired, Exception) as e:
            return self._handle_version_cmd_error(e)

        # 解析版本号（调用辅助方法）
        return self._parse_version_output(result)

    def _parse_version_output(self, result):
        """解析版本命令的输出结果，提取版本号"""
        # 合并stdout和stderr，避免版本信息输出到错误流
        output_lines = result.stdout.splitlines() + result.stderr.splitlines()
        
        for line in output_lines:
            if "Version:" in line:
                # 提取版本号并格式化（确保x.y格式）
                raw_version = line.split(":")[-1].strip()
                return self._format_version(raw_version)
        
        self._log_warning(f"无法从输出中解析版本号: {output_lines}")
        return None

    def _format_version(self, raw_version):
        """将原始版本号格式化为x.y的字符串格式"""
        try:
            float_version = float(raw_version)
            return f"{float_version:.1f}"
        except ValueError:
            self._log_error(f"版本号格式异常，原始值: {raw_version}")
            return None

    def _handle_version_cmd_error(self, error):
        """处理版本命令执行中的异常（统一错误处理）"""
        if isinstance(error, subprocess.CalledProcessError):
            self._log_error(f"版本命令执行失败: {error.stderr.strip()}")
        elif isinstance(error, subprocess.TimeoutExpired):
            self._log_error("版本命令执行超时")
        else:
            self._log_error(f"获取版本号时发生错误: {str(error)}")
        return None

    def get_latest_version(self):
        """获取最新版本号（返回字符串格式）"""
        try:
            response = requests.get(
                self.version_url,
                timeout=self.VERSION_CHECK_TIMEOUT,
                headers={"User-Agent": self.DEFAULT_USER_AGENT}
            )
            response.raise_for_status()
            latest_version = response.text.strip()
            
            if not latest_version:
                self._log_warning("获取到的最新版本号为空")
                return None
            return latest_version
        except requests.RequestException as e:
            self._log_error(f"获取最新版本失败: {str(e)}")
            return None

    def check_update_available(self):
        """检查是否有可用更新（基于float格式版本号比较）"""
        current_version_str = self.get_current_version()
        latest_version_str = self.get_latest_version()

        # 检查版本号获取结果
        if not current_version_str:
            return False, "无法获取当前版本"
        if not latest_version_str:
            return False, "无法获取最新版本"

        # 版本号比较
        try:
            current_version = float(current_version_str)
            latest_version = float(latest_version_str)
        except ValueError as e:
            error_msg = f"版本号格式错误（应为x.y）: {str(e)}"
            self._log_error(f"{error_msg}（当前: {current_version_str}, 最新: {latest_version_str}）")
            return False, error_msg

        if latest_version > current_version:
            return True, f"当前版本: {current_version_str}, 最新版本: {latest_version_str}"
        else:
            return False, f"当前版本: {current_version_str}, 最新版本: {latest_version_str}"

    # ------------------------------
    # 安装/更新相关方法
    # ------------------------------
    def is_installed(self):
        """检查site_total是否已安装且可执行"""
        return os.path.exists(self.site_total_bin) and os.access(self.site_total_bin, os.X_OK)

    def install_or_update(self):
        """执行安装或更新操作"""
        # 下载安装脚本
        install_script = self._download_install_script()
        if not install_script:
            return False, "下载安装脚本失败"

        # 执行安装脚本
        return self._execute_install_script(install_script)

    def _download_install_script(self):
        """下载安装脚本（提取为独立方法，职责单一）"""
        try:
            response = requests.get(
                self.install_sh,
                timeout=self.SCRIPT_DOWNLOAD_TIMEOUT,
                headers={"User-Agent": self.DEFAULT_USER_AGENT}
            )
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            self._log_error(f"下载安装脚本失败: {str(e)}")
            return None

    def _execute_install_script(self, script_content):
        """执行安装脚本（提取为独立方法，职责单一）"""
        temp_script = None
        try:
            # 创建临时脚本文件
            temp_script = self._create_temp_script(script_content)
            
            # 执行脚本
            result = subprocess.run(
                ["bash", temp_script],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                timeout=self.INSTALL_SCRIPT_TIMEOUT
            )

            # 检查执行结果
            if result.returncode != 0:
                error_msg = (f"安装脚本执行失败(返回码: {result.returncode})\n"
                             f"错误输出: {result.stderr.strip()[:500]}")
                self._log_error(error_msg)
                return False, error_msg

            if not self.is_installed():
                return False, "安装完成但未找到可执行文件"

            return True, "安装/更新成功"

        except subprocess.TimeoutExpired:
            return False, "安装脚本执行超时"
        except Exception as e:
            return False, f"执行安装过程中出错: {str(e)}"
        finally:
            # 清理临时文件
            self._cleanup_temp_script(temp_script)

    def _create_temp_script(self, content):
        """创建临时脚本文件（辅助方法，封装文件操作）"""
        with tempfile.NamedTemporaryFile(
            mode='w',
            suffix='.sh',
            delete=False,
            dir='/tmp'
        ) as f:
            f.write(content)
            return f.name

    def _cleanup_temp_script(self, script_path):
        """清理临时脚本文件（辅助方法，确保资源释放）"""
        if script_path and os.path.exists(script_path):
            try:
                os.remove(script_path)
            except OSError as e:
                self._log_warning(f"无法删除临时文件 {script_path}: {str(e)}")

    # ------------------------------
    # 主逻辑与工具方法
    # ------------------------------
    def update_if_needed(self):
        """有更新时执行更新，未安装时执行安装（主逻辑入口）"""
        if not self.is_installed():
            self._log_info("未检测到安装，开始执行安装")
            return self.install_or_update()

        has_update, msg = self.check_update_available()
        if not has_update:
            return False, f"无需更新: {msg}"

        self._log_info(f"检测到更新: {msg}，开始执行更新")
        return self.install_or_update()

    # 日志工具方法（封装日志调用，便于统一管理）
    @staticmethod
    def _log_info(msg):
        logging.info(msg)

    @staticmethod
    def _log_warning(msg):
        logging.warning(msg)

    @staticmethod
    def _log_error(msg):
        logging.error(msg)


if __name__ == "__main__":
    updater = UpdateSiteTotal()
    success, message = updater.update_if_needed()
    logging.info(f"操作结果: {'成功' if success else '失败'} - {message}")