skopeo同步镜像案例

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import subprocess
import sys
import os

# ==========================================
# ⚙️ 配置区域
# ==========================================
# 1. 认证信息
SRC_CREDS = "qiaoning:123456"
DEST_CREDS = "admin:123456"

# 2. 镜像地址
SRC_IMAGE = "docker://hub.qiaoning.com/jdaip-system/train/jupyter-llamafactory0.9.2-torch2.5-py3.11-rocm6.2.0-ubuntu24.04-openeuler22-amd64:v1"
DEST_IMAGE = "docker://docker.io.com/jdaip-system/train/jupyter-llamafactory0.9.2-torch2.5-py3.11-rocm6.2.0-ubuntu24.04-openeuler22-amd64:v1"

# 3. 本地纯文本日志路径（用于留存审计，会自动去掉动态刷新的乱码）
LOG_FILE_PATH = "./skopeo_sync.log"

# ==========================================
# 🛠️ 核心执行函数
# ==========================================
def run_skopeo_live(cmd, timeout=7200):
    """
    运行 Skopeo 并实时、原汁原味地在终端展示其动态进度条，同时记录纯净日志。
    """
    print(f"▶️ 开始执行同步命令，预计耗时较长，请耐心等待...")
    
    log_file = open(LOG_FILE_PATH, "a", encoding="utf-8")
    try:
        # 使用 Popen 启动子进程
        # stderr=subprocess.STDOUT 把错误流和标准输出合并，因为 skopeo 的进度条很多时候在 stderr 里
        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1 # 行缓冲/块缓冲自适应
        )

        output_buffer = []
        
        # 实时按字符/行读取输出。
        # 因为进度条包含 '\r'（回车不换行），我们需要按单个字符或块读取来完美还原终端动态效果
        while True:
            # 每次读取一个字符，兼容 \r 动态进度条
            char = process.stdout.read(1)
            if not char and process.poll() is not None:
                break
            
            if char:
                # 1. 实时输出到你的终端屏幕上，完美还原 [======>-----] 进度条
                sys.stdout.write(char)
                sys.stdout.flush()
                output_buffer.append(char)
                
                # 2. 写入本地日志文件（去掉回车符 \r，防止日志文件里全是重叠乱码）
                if char != '\r':
                    log_file.write(char)
                    if char == '\n':
                        log_file.flush()

        # 等待进程完全结束
        returncode = process.wait(timeout=timeout)
        full_output = "".join(output_buffer).strip()

        if returncode == 0:
            print("\n\n✅ 镜像迁移成功！")
            return returncode, full_output, ""
        else:
            print(f"\n\n❌ 镜像迁移失败，错误码: {returncode}")
            return returncode, "", f"Skopeo exited with code {returncode}"

    except subprocess.TimeoutExpired:
        process.kill()
        error_msg = f"\n\n🚨 传输超时！已超过设置的 {timeout} 秒限制。"
        print(error_msg)
        return -1, "", error_msg
    except Exception as e:
        if 'process' in locals():
            process.kill()
        error_msg = f"\n\n🚨 脚本运行异常: {str(e)}"
        print(error_msg)
        return -1, "", error_msg
    finally:
        log_file.close()

# ==========================================
# 🏁 脚本入口
# ==========================================
if __name__ == "__main__":
    # 组装 skopeo 列表参数（不使用 shell=True，防止特殊字符引发解析漏洞）
    skopeo_cmd = [
        "skopeo", "copy",
        "--src-creds", SRC_CREDS,
        "--dest-creds", DEST_CREDS,
        "--timeout", "120m",           # Skopeo 内部的底层网络超时
        SRC_IMAGE,
        DEST_IMAGE
    ]
    
    print("=" * 60)
    print(f"源 镜 像: {SRC_IMAGE}")
    print(f"目标镜像: {DEST_IMAGE}")
    print(f"本地日志: {os.path.abspath(LOG_FILE_PATH)}")
    print("=" * 60)

    # 执行同步，Python 侧逻辑最大超时限制设为 2.5 小时（9000秒），略大于 skopeo 的 120m
    exit_code, stdout, stderr = run_skopeo_live(skopeo_cmd, timeout=9000)
    
    sys.exit(exit_code)

公共执行命令函数

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import subprocess
import sys
import os

def run_command_live(cmd, log_file_path=None, timeout=3600):
    """
    通用函数：实时流式执行系统命令，支持动态进度条还原（兼容\r），同时支持输出到日志文件。
    
    :param cmd: 列表格式的命令，例如 ['docker', 'pull', 'alpine'] 或 ['ping', '-c', '4', 'baidu.com']
    :param log_file_path: 可选，本地纯文本日志保存路径（会自动过滤掉进度条刷新导致的乱码）
    :param timeout: 全局绝对超时时间（秒），默认 1 小时
    :return: (returncode, full_output, error_msg)
    """
    # 如果指定了日志路径，则以追加模式打开
    log_file = open(log_file_path, "a", encoding="utf-8") if log_file_path else None
    
    try:
        # 启动子进程，合并 stdout 和 stderr，采用自适应缓冲
        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1
        )

        output_buffer = []
        
        # 实时按字符读取，这是兼容 \r 动态进度条（不刷屏）的核心设计
        while True:
            char = process.stdout.read(1)
            # 如果没有字符了，且子进程已经退出，则跳出循环
            if not char and process.poll() is not None:
                break
            
            if char:
                # 1. 原汁原味地实时同步到当前终端屏幕
                sys.stdout.write(char)
                sys.stdout.flush()
                output_buffer.append(char)
                
                # 2. 如果启用了日志，同步写入文件（过滤掉 \r，确保生成的日志文件纯净无乱码）
                if log_file and char != '\r':
                    log_file.write(char)
                    if char == '\n':
                        log_file.flush()

        # 等待进程完全结束并获取状态码
        returncode = process.wait(timeout=timeout)
        full_output = "".join(output_buffer).strip()

        if returncode == 0:
            return returncode, full_output, ""
        else:
            return returncode, full_output, f"Command failed with exit code {returncode}"

    except subprocess.TimeoutExpired:
        process.kill()
        err = f"Execution timed out after {timeout} seconds."
        return -1, "", err
    except Exception as e:
        if 'process' in locals():
            process.kill()
        return -1, "", str(e)
    finally:
        if log_file:
            log_file.close()

# ==========================================
# 🏁 通用调用示例
# ==========================================
if __name__ == "__main__":
    LOG_FILE = "./system_execution.log"
    
    # 示例 1：测试普通的换行输出命令 (列出当前目录)
    print("--- 测试命令 1: ls -la ---")
    cmd1 = ["ls", "-la"]
    run_command_live(cmd1)

    print("\n" + "="*40 + "\n")

    # 示例 2：测试带动态进度条的命令 (用 curl 下载文件)
    print("--- 测试命令 2: curl 下载 (带进度条) ---")
    cmd2 = ["curl", "-L", "https://speed.hetzner.de/100MB.bin", "-o", "/tmp/100MB.bin"]
    
    # 执行它，日志保存在 LOG_FILE 中，允许跑 10 分钟
    code, stdout, stderr = run_command_live(cmd2, log_file_path=LOG_FILE, timeout=600)
    
    if code == 0:
        print("\n🎉 命令执行成功！")
    else:
        print(f"\n❌ 命令执行失败，原因: {stderr}")