源码:dongdongqiang1/dir - Gitee.com

log.py:
 

import inspect

class Logger:
    """Simple logging utility with adjustable levels at runtime"""
    LEVELS = {'DEBUG': 10, 'INFO': 20, 'WARN': 30, 'ERROR': 40, 'CRIT': 50}

    def __init__(self, level_name='INFO'):
        self.level = self.LEVELS.get(level_name, 20)
        self.crit(f"init log level = {level_name}")

    def set_level(self, level_name):
        self.level = self.LEVELS.get(level_name, 20)
        self.crit(f"set log level = {level_name}")

    def _log(self, level_name, *args, **kwargs):
        """Internal logging method: checks level and outputs if allowed"""
        if self.LEVELS[level_name] < self.level:
            return

        # Get caller's frame (skip current _log and specific level methods)
        frame = inspect.currentframe().f_back.f_back
        funcname = frame.f_code.co_name
        lineno = frame.f_lineno

        if self.LEVELS.get(level_name) >= self.LEVELS.get("WARN"):
            print(f"{funcname} {lineno}: [{level_name}]", *args, **kwargs)
        else:
            print(f"{funcname} {lineno}:", *args, **kwargs)

    def debug(self, *args, **kwargs):
        self._log('DEBUG', *args, **kwargs)

    def info(self, *args, **kwargs):
        self._log('INFO', *args, **kwargs)

    def warn(self, *args, **kwargs):
        self._log('WARN', *args, **kwargs)

    def error(self, *args, **kwargs):
        self._log('ERROR', *args, **kwargs)

    def crit(self, *args, **kwargs):
        self._log('CRIT', *args, **kwargs)
 

version.json:

{
  "version": "1.0.0",
  "update_config_url": "https://raw.giteeusercontent.com/dongdongqiang1/dir/raw/master/tool_box/auto_update/version.json",
  "zip_url": "https://gitee.com/dongdongqiang1/dir/raw/master/tool_box/auto_update/new_version.zip",
  "changelog": "Initial version, supports automatic updates"
}
 

updater.py:

"""
Auto-update module.

On startup, reads local version.json to obtain the current version and
the URL of the remote update configuration. If a newer version is found,
the update ZIP is downloaded, extracted, and a restart script is launched
to replace the application files before relaunching the main program.

All source files are expected to reside in the same directory.
"""

import json
import os
import platform
import shutil
import subprocess
import sys
import urllib.request
import urllib.error
import zipfile

from log import Logger

logger = Logger('INFO')

LOCAL_CONFIG_FILE = "version.json"

# Timeout values in seconds for HTTP requests
CONFIG_TIMEOUT = 10      # timeout when fetching remote version.json
DOWNLOAD_TIMEOUT = 30    # timeout when downloading the update ZIP

# HTTP headers to mimic a browser and avoid being blocked by some hosts
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/120.0.0.0 Safari/537.36",
    "Referer": "https://gitee.com"
}

def _load_local_config() -> tuple:
    """Load local version config.

    Returns:
        A tuple of (current_version, update_config_url).
        Defaults to ("1.0.0", "") if the file is missing or invalid.
    """
    try:
        with open(LOCAL_CONFIG_FILE, 'r', encoding='utf-8') as f:
            cfg = json.load(f)
        return cfg.get("version", "1.0.0"), cfg.get("update_config_url", "")
    except Exception as e:
        logger.error(f"Failed to load local config: {e}")
        return "1.0.0", ""

def _fetch_json(url: str) -> dict:
    """Fetch and parse JSON from a remote URL.

    Args:
        url: The URL to fetch from.

    Returns:
        Parsed JSON as a dictionary.

    Raises:
        Exception: If the request fails or the response is not valid JSON.
    """
    req = urllib.request.Request(url, headers=HEADERS)
    try:
        with urllib.request.urlopen(req, timeout=CONFIG_TIMEOUT) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except Exception as e:
        logger.error(f"Failed to fetch JSON from {url}: {e}")
        raise

def _version_tuple(v: str) -> tuple:
    """Convert a version string to a comparable tuple.

    Example: '1.2.3' -> (1, 2, 3)
    Returns (0, 0, 0) if the string is not a valid dotted numeric version.
    """
    try:
        return tuple(map(int, v.split(".")))
    except Exception:
        return (0, 0, 0)

def _download_file(url: str, dest_path: str) -> None:
    """Download a file from the given URL to a local path.

    Args:
        url: The file URL.
        dest_path: Local path where the content will be saved.

    Raises:
        Exception: If the download fails (error is logged before re-raising).
    """
    req = urllib.request.Request(url, headers=HEADERS)
    try:
        with urllib.request.urlopen(req, timeout=DOWNLOAD_TIMEOUT) as resp, \
                open(dest_path, 'wb') as f:
            f.write(resp.read())
    except Exception as e:
        logger.error(f"Failed to download {url}: {e}")
        raise

def _gen_restart_script(app_dir: str, main_script_path: str) -> str:
    """Generate a platform-specific restart script.

    The script performs the following steps after the main program exits:
      1. Wait briefly for the main process to terminate.
      2. Remove all files except the update artifacts and the script itself.
      3. Move the contents of the 'new_version' folder into the current directory.
      4. Delete the temporary 'new_version' folder and the downloaded ZIP.
      5. Restart the main program.
      6. Self-destruct (delete the script).

    Args:
        app_dir: The application root directory.
        main_script_path: Absolute path to the main Python script.

    Returns:
        The full path to the generated restart script.
    """
    system = platform.system()
    if system == "Windows":
        script_name = "update_restart.bat"
        content = (
            "@echo off\n"
            "timeout /t 2 >nul\n"
            "cd /d \"%~dp0\"\n"
            "for /F \"delims=\" %%i in ('dir /b') do (\n"
            "    if not \"%%i\"==\"new_version.zip\" if not \"%%i\"==\"new_version\" if not \"%%i\"==\"update_restart.bat\" (\n"
            "        rd /s /q \"%%i\" 2>nul || del /q \"%%i\"\n"
            "    )\n"
            ")\n"
            "xcopy /Y /E \".\\new_version\\*\" \".\\\"\n"
            "rd /s /q \"new_version\"\n"
            "del /q \"new_version.zip\"\n"
            f"start \"\" \"{main_script_path}\"\n"
            "cmd /c del /q \"%~f0\"\n"
        )
    else:
        script_name = "update_restart.sh"
        content = (
            "#!/bin/bash\n"
            "sleep 2\n"
            "cd \"$(dirname \"$0\")\"\n"
            "find . -mindepth 1 -not -name 'new_version.zip' -not -name 'new_version' -not -name 'update_restart.sh' -exec rm -rf {} + 2>/dev/null\n"
            "shopt -s dotglob\n"
            "cp -r new_version/* .\n"
            "rm -rf new_version new_version.zip\n"
            f"\"{main_script_path}\" &\n"
            "rm -- \"$0\"\n"
        )

    script_path = os.path.join(app_dir, script_name)
    with open(script_path, 'w', encoding='utf-8') as f:
        f.write(content)

    if system != "Windows":
        os.chmod(script_path, 0o755)   # make the script executable on Unix

    return script_path

def auto_update() -> bool:
    """Check for updates and apply if a newer version is available.

    The remote version configuration is fetched from the URL specified in
    the local version.json. If a newer version exists, the update ZIP is
    downloaded and a restart script is launched.

    Returns:
        True if an update was triggered (the caller should exit immediately).
        False if no update is needed or an error occurred.
    """
    main_script_path = sys.argv[0]
    app_dir = os.path.dirname(os.path.abspath(main_script_path))

    # 1. Load local version configuration (current version and remote config URL)
    current_version, update_config_url = _load_local_config()
    if not update_config_url:
        logger.error("No update_config_url in local config. Skipping update check.")
        return False

    logger.info(f"Current version: {current_version}")
    logger.info("Checking for updates...")

    # 2. Fetch remote version configuration
    try:
        remote_cfg = _fetch_json(update_config_url)
    except Exception:
        return False

    latest_version = remote_cfg.get("version", "1.0.0")
    zip_url = remote_cfg.get("zip_url", "")
    logger.info(f"Remote version: {latest_version}")
    logger.info(f"Download URL: {zip_url}")

    if not zip_url:
        logger.error("No zip_url in remote config.")
        return False

    # 3. Compare versions (semantic version tuples)
    if _version_tuple(current_version) >= _version_tuple(latest_version):
        logger.info("Already up to date.")
        return False

    logger.info(f"New version {latest_version} found. Starting download...")

    # 4. Download the update ZIP
    zip_path = os.path.join(app_dir, "new_version.zip")
    try:
        _download_file(zip_url, zip_path)
        logger.info("Download successful.")
    except Exception as e:
        logger.error(f"Download new version failed: {e}")
        return False

    # 5. Verify ZIP file integrity
    if not zipfile.is_zipfile(zip_path):
        logger.error(f"{zip_path} is not a valid ZIP file, aborting.")
        try:
            os.remove(zip_path)
        except OSError:
            pass
        return False

    # 6. Extract the archive into a temporary folder
    extract_dir = os.path.join(app_dir, "new_version")
    try:
        if os.path.exists(extract_dir):
            shutil.rmtree(extract_dir)
        with zipfile.ZipFile(zip_path, 'r') as zf:
            zf.extractall(extract_dir)
        logger.info(f"Extracted to {extract_dir}")
    except Exception as e:
        logger.warn(f"Extraction warning (non-critical files may be missing): {e}")

    # 7. Create and launch the restart script
    script_path = _gen_restart_script(app_dir, main_script_path)
    logger.info("Restart script created, preparing to exit...")

    if platform.system() == "Windows":
        subprocess.Popen(["cmd", "/c", script_path], shell=True)
    else:
        subprocess.Popen(["bash", script_path])

    return True
 

main.py:

"""
Application entry point.
"""
import sys
import updater

def main():
    # Auto-update will read local version.json internally
    if updater.auto_update():
        print("Update triggered, exiting to restart...")
        sys.exit(0)

    # No update, launch the bussiness function
    import os
    os.system("a.py")

if __name__ == "__main__":
    main()
 

a.py:

import time
def main():
    print("work")
    time.sleep(20)

if __name__ == "__main__":
    main()
 

效果：

猜你想问：
1、这是什么？整体介绍
2、怎么用？注意事项
3、每个文件分别有啥用？有啥功能？作者设计思路？

DeepSeek  回答说：

猜你想问：
1、这是什么？整体介绍
2、怎么用？注意事项
3、每个文件分别有啥用？有啥功能？作者设计思路？

DeepSeek  回答说：

## 1. 这是什么？整体介绍

这是一个**基于 Gitee 仓库的 Python 应用自动更新框架**。它的核心流程是：

1. **启动时检查**：主程序 `main.py` 运行后，立即调用 `updater.auto_update()`。  
2. **读取本地配置**：从 `version.json` 获取当前版本号和远程配置文件的 URL。  
3. **获取远程版本信息**：从远程 `version.json` 拿到最新版本号和更新包下载地址。  
4. **版本比较**：如果远程版本比本地新，就下载 ZIP 更新包，否则正常运行。  
5. **应用更新**：下载完成后解压到临时目录，生成一个**平台相关的重启替换脚本**，然后退出主程序。  
6. **自动替换重启**：重启脚本会等待主程序退出，**清空旧文件、移入新版本文件**，最后重新启动主程序并销毁自身。  

整个过程无需用户干预，也**不依赖任何第三方库**（只用 Python 标准库），非常适合轻量级工具或内部小应用的自动更新场景。

---

## 2. 怎么用？注意事项

### 使用方法

1. **准备文件**  
   将 `main.py`、`updater.py`、`log.py`、`version.json` 以及你的实际业务代码（如 `a.py`）放在**同一目录**下。

2. **配置 version.json**  
   - `update_config_url`：指向你在 Gitee 上放置的远程 `version.json` 的 raw 链接（推荐用 `raw.giteeusercontent.com` 域名，防 403）。  
   - `zip_url`：指向全量更新包 `new_version.zip` 的 raw 下载链接。  
   - 本地 `version.json` 中的 `version` 字段是你**当前程序的版本号**，发布新版本时需同步更新远程的 `version.json` 并将此值改大。

3. **准备远程文件**  
   - 在 Gitee 仓库中维护一个远程 `version.json`，内容格式和本地一致，但 `version` 字段应是最新版本号。  
   - 将新版本程序的所有文件**打包成 `new_version.zip`**（根目录直接放文件，不要嵌套文件夹），上传到 `zip_url` 对应的路径。

4. **运行**  
   执行 `python main.py`。程序会先自动检查更新，若无更新则启动实际业务（这里是运行 `a.py`）。

### 注意事项

- **网络与防盗链**：Gitee 对 `gitee.com/.../raw/...` 链接有防盗链限制。当前代码已在所有请求中加入了 **`User-Agent`** 和 **`Referer: https://gitee.com`**，可以正常下载。如果仍出现 403，建议将 `update_config_url` 的域名换成 `raw.giteeusercontent.com`。
- **ZIP 包内容**：更新是**全量替换**——重启脚本会删除当前目录下除 `new_version.zip`、`new_version` 文件夹和自身以外的所有内容。**因此不要把用户数据、配置文件放在这个目录下**，否则会被误删。
- **版本号格式**：必须为 `x.y.z` 纯数字点分格式（例如 `1.0.0`），其他格式会被视为 `0.0.0`，可能导致比较异常。
- **文件名编码**：如果 ZIP 包内含有中文或特殊字符的文件名，解压可能失败。代码对此做了容错（仅警告后继续），但这类文件将不会更新。
- **权限问题（Linux/macOS）**：生成重启脚本时会自动赋予执行权限（`0o755`），无需额外操作。如果主程序文件本身没有执行权限，请事先设置。
- **运行环境**：Windows 使用 `update_restart.bat`，Linux/macOS 使用 `update_restart.sh`，跨平台兼容。

---

## 3. 每个文件分别有啥用？功能与设计思路

### `log.py` – 简易日志工具

**功能**：提供 `Logger` 类，支持 `DEBUG / INFO / WARN / ERROR / CRIT` 五个级别，并可在运行时动态调整级别。

**设计思路**：
- 完全自包含，不依赖 `logging` 模块，避免复杂配置。
- 通过 `inspect` 获取**调用者的函数名和行号**，方便快速定位问题。
- 高优先级日志（`WARN` 及以上）会显示级别标签，低优先级日志只显示位置，减少视觉干扰。
- 初始化时用 `CRIT` 输出当前日志级别，确保启动配置可见。

---

### `version.json` – 本地版本与更新配置

**功能**：存储当前应用的版本号、远程更新配置地址、ZIP 下载地址和更新日志。

**设计思路**：
- 将“版本信息”和“更新源”从代码中剥离，修改配置无需改动 Python 程序。
- 远程的 `update_config_url` 同理，形成“本地 → 远程配置 → 更新包”的链式结构，方便统一管理。

---

### `updater.py` – 自动更新核心

**功能**：
- 读取本地 `version.json`，获取当前版本和远程配置 URL。
- 请求远程配置（带浏览器伪装头），得到最新版本和 `zip_url`。
- 比较版本，若需要更新则下载 ZIP、校验完整性、解压到 `new_version/`。
- 生成平台相关的重启脚本，并启动该脚本后返回 `True` 通知主程序退出。

**设计思路**：
- **全量更新**：每次用完整 ZIP 包覆盖，避免复杂的增量或差异更新。
- **模拟浏览器请求**：所有 `urllib.request.Request` 都加上了 `User-Agent` 和 `Referer`，防止被 Gitee 拒止，也预留了后续加入 Basic Auth 的空间。
- **安全替换**：不直接覆盖正在运行的程序，而是通过重启脚本在程序退出后完成文件替换，避免文件被占用导致更新失败。
- **跨平台**：根据 `platform.system()` 生成对应的批处理或 shell 脚本，并处理好隐藏文件和执行权限。
- **容错处理**：ZIP 解压失败只警告并继续，防止个别非关键文件损坏导致整个更新中断；下载和远程配置获取失败则立即终止更新。
- **清晰的错误日志**：关键步骤都有日志输出，便于排查问题（默认日志级别设为 `INFO`，正式发布后可保持）。

---

### `main.py` – 应用入口

**功能**：  
调用 `updater.auto_update()`，如果触发更新则立即退出；否则执行实际业务（这里用 `os.system("a.py")` 替代）。

**设计思路**：  
将更新检查放在所有业务逻辑之前，确保每次启动都先尝试更新。更新模块与业务代码完全解耦，你可以把 `os.system("a.py")` 替换成任何启动函数（例如导入自己的 GUI 主界面）。

---

### `a.py` – 模拟业务逻辑

**功能**：  
打印 `work` 并等待 20 秒，仅用于演示更新后业务代码能正常运行。

**设计思路**：  
一个占位文件，实际开发时替换为你自己的功能模块。它的存在是为了验证更新流程不影响后续业务启动。

---

整个框架极其轻量，无外部依赖，你只需专注于维护远程的 `version.json` 和 `new_version.zip`，便能为任何 Python 小工具增加全自动更新能力。