用Python实现Docker镜像批量推送(带进度条)

首页

编程语言

热心网友

转载

2026-04-29

背景

在本地调试或搭建漏洞复现环境时，我们经常会用到各种公开的 Docker 镜像，比如 vulhub/、jboss/、portainer/ 这些。但有时候，你可能需要把这些镜像迁移到自己的 Docker Hub 仓库里，原因不外乎这么几个：

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

用Python实现Docker镜像批量推送(带进度条)

防止哪天官方仓库突然删除或者访问不了。
需要在隔离的内网环境里部署。
想长期保存当前正在使用的特定镜像版本。
为了配合自己的 docker-compose.yml 文件进行统一管理。

手动操作的话，就是反复执行 docker tag 和 docker push，镜像一多、体积一大，这个过程就变得特别繁琐。所以，这里分享一个Python 脚本，它能帮你自动化完成这件事，主要功能包括：

✅ 自动读取当前所有容器（包括已经停止的）所使用的镜像。
✅ 去掉原始仓库的前缀（比如 xxx/），然后打上你自己的仓库标签。
✅ 提供实时进度条（基于 tqdm 库），清晰展示每一层镜像的上传进度。
✅ 实现增量上传：如果某层镜像在远程仓库已经存在，系统会自动跳过，不会重复消耗流量和时间。

脚本代码

把下面的代码保存为一个文件，比如就叫 push_images_with_progress.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import docker
import sys
from tqdm import tqdm
# ========== 配置 ==========
DOCKER_USERNAME = "rockmelodies"   # 你的 Docker Hub 用户名
# =========================
client = docker.from_env()
def get_unique_images_from_all_containers():
    """获取所有容器（包括停止的）使用的镜像名（唯一）"""
    containers = client.containers.list(all=True)
    images = set()
    for c in containers:
        # 镜像名格式：repository:tag
        image_full = c.image.tags[0] if c.image.tags else f"{c.image.id[:12]}"
        images.add(image_full)
    return images
def short_image_name(full_name):
    """从 vulhub/nginx:1.11.13 提取 nginx:1.11.13"""
    # 去掉第一个 / 之前的所有内容
    if '/' in full_name:
        return full_name.split('/', 1)[-1]
    return full_name
def tag_and_push(image_full):
    """打标签并推送，显示进度条"""
    short = short_image_name(image_full)
    new_tag = f"{DOCKER_USERNAME}/{short}"
    print(f"\n? 处理镜像: {image_full}")
    print(f"   → 新标签: {new_tag}")
    # 1. 拉取原始镜像（确保本地存在）
    try:
        client.images.get(image_full)
    except docker.errors.ImageNotFound:
        print(f"   ⚠️ 本地未找到 {image_full}，尝试拉取...")
        client.images.pull(image_full)
    # 2. 打标签
    img = client.images.get(image_full)
    img.tag(new_tag)
    # 3. 推送（带进度条）
    print("   ⬆️ 开始推送...")
    layers_progress = {}
    for line in client.images.push(new_tag, stream=True, decode=True):
        if 'id' in line and 'progressDetail' in line:
            layer_id = line['id']
            progress = line['progressDetail']
            total = progress.get('total')
            current = progress.get('current')
            if total and current:
                if layer_id not in layers_progress:
                    layers_progress[layer_id] = tqdm(
                        total=total, unit='B', unit_scale=True,
                        desc=f"    层 {layer_id[:12]}", lea ve=False
                    )
                layers_progress[layer_id].update(current - layers_progress[layer_id].n)
                if current >= total:
                    layers_progress[layer_id].close()
                    del layers_progress[layer_id]
        elif 'status' in line:
            status = line['status']
            if 'already exists' in status.lower():
                tqdm.write(f"   ℹ️ {status}")
    print(f"   ✅ 推送完成: {new_tag}\n")
def main():
    print("? 获取所有容器使用的镜像...")
    images = get_unique_images_from_all_containers()
    if not images:
        print("❌ 没有找到任何容器镜像。")
        sys.exit(1)
    print(f"? 找到 {len(images)} 个唯一镜像:")
    for img in images:
        print(f"   - {img}")
    for img in images:
        try:
            tag_and_push(img)
        except Exception as e:
            print(f"❌ 推送失败 {img}: {e}")
    print("\n? 所有镜像处理完毕！")
if __name__ == "__main__":
    main()

使用步骤

1. 安装依赖

pip install docker tqdm

如果系统提示找不到 pip 命令，可能需要先确保你的 Python 环境配置正确，有时候需要运行一下 python -m ensurepip --upgrade 来修复。

2. 登录 Docker Hub

docker login

根据提示，输入你的 Docker ID 和 个人访问令牌（推荐使用令牌，比密码更安全）或者密码。

3. 修改脚本中的用户名

打开脚本，找到 DOCKER_USERNAME = "rockmelodies" 这一行，把里面的 rockmelodies 替换成你自己的 Docker Hub 用户名。

4. 运行脚本

python3 push_images_with_progress.py

运行效果示例

? 获取所有容器使用的镜像...
? 找到 11 个唯一镜像:
   - vulhub/comfyui:3.39.1-with-manager
   - vulhub/webmin:1.910
   ...
? 处理镜像: vulhub/comfyui:3.39.1-with-manager
   → 新标签: rockmelodies/comfyui:3.39.1-with-manager
   ⬆️ 开始推送...
    层 a1b2c3d4e5f6: 45%|████▌     | 45.2M/100M [00:05<00:06, 8.2MB/s]
   ℹ️ Layer already exists
   ✅ 推送完成: rockmelodies/comfyui:3.39.1-with-manager
? 所有镜像处理完毕！

核心优势

? 增量上传（不会重复推送）

脚本通过 docker.images.push(stream=True) 来获取推送过程的实时流式输出。
Docker 底层采用的是分层存储机制。这意味着，如果远程仓库里已经存在某一层镜像（比如之前推送过），Docker 会直接返回 Layer already exists 的状态并跳过，不会重新传输数据。
所以，这个脚本你完全可以放心多次运行，完全不会浪费流量和时间在重复内容上。

? 精确进度条

利用 tqdm 库，为每一个正在上传的镜像层单独绘制进度条。
实时显示已传输大小、总大小以及当前的传输速度，让你对进度一目了然。
当某一层传输完成后，对应的进度条会自动关闭，界面保持整洁。

? 自动提取镜像列表

通过调用 Docker API，获取所有容器（参数 all=True 确保了已停止的容器也会被包括进来）的信息。
自动进行去重处理，同一个镜像不会因为被多个容器使用而被重复推送。
你不需要手动维护一个镜像列表，这对于容器动态变化的环境特别友好。

?️ 智能打标签

自动去除原始镜像的仓库前缀，比如常见的 vulhub/、jboss/、portainer/。
转换逻辑很直观：vulhub/nginx:1.11.13 → 你的用户名/nginx:1.11.13。
这样生成的标签既简洁，又完全符合 Docker Hub 的命名规范。

常见问题

Q1：推送时提示“denied: requested access to the resource is denied”

A：这通常是因为没有登录 Docker Hub，或者登录凭据已经失效。请先执行 docker login 命令，并仔细检查输入的用户名和令牌（或密码）是否正确。

Q2：脚本运行时卡住不动？

A：推送大型镜像本身就需要时间，网络传输过程中进度条可能会有短暂的停顿，这通常是正常的，后台仍在工作。请耐心等待一会儿。如果长时间（比如超过10分钟）完全没有任何输出和进度更新，那才需要检查一下网络连接是否正常。

Q3：我不想推送所有镜像，只想推送某几个怎么办？

A：有两种思路：一是修改脚本里的 get_unique_images_from_all_containers() 函数，增加一个过滤条件；二是在脚本开头直接定义一个你需要的镜像白名单列表，只处理列表里的镜像。

Q4：推送完成后如何确认镜像在 Docker Hub 上？

A：最直接的方法是登录 Docker Hub 网站（https://hub.docker.com/u/你的用户名）查看你的仓库列表。或者，更“工程师”的做法是，直接在终端执行 docker pull 你的用户名/镜像名:标签 命令来测试拉取，成功就说明推送没问题。

扩展想法

可以增加一个 --dry-run（干跑）模式，只模拟运行并显示将要推送的镜像列表，而不实际执行推送操作，方便预览。
扩展脚本功能，支持同时推送到多个目标仓库，比如除了 Docker Hub，还能推送到阿里云、腾讯云的镜像仓库。
增加推送失败后的自动重试机制，并把详细的操作日志记录到文件里，方便后续排查问题。

总结

总的来说，这个 Python 脚本提供了一种高效的解决方案：只需一条命令，就能把本地所有运行过的 Docker 镜像，完整地迁移到你自己的 Docker Hub 仓库中。整个过程配有实时进度条，透明可控。而且，得益于 Docker 镜像的分层特性，重复运行脚本也不会造成网络和存储资源的浪费。

如果你也在频繁管理多个漏洞复现环境，或者积累了不少自定义镜像需要归档，不妨试试这个脚本。它能让原本繁琐的镜像迁移工作，变得轻松又省心。

来源:https://www.jb51.net/python/363076g3g.htm

免责声明：游乐网为非赢利性网站，所展示的游戏/软件/文章内容均来自于互联网或第三方用户上传分享，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系youleyoucom@outlook.com。

上一篇：Git单分支与多分支的协作流程下一篇：如何在 Gin 中间件中检测后续处理器的执行失败状态