镜像转存

有时候需要将镜像从Docker Hub中下载，再转存到私服的镜像仓库中。为完成上述任务，设计了如下的Python脚本,只要按需修改高亮部分即可: 

  1. import os
  2. import subprocess
  3. import requests
  4. import json
  5.  
  6. # 定义变量
  7. public_image = "redis"  # 公共镜像名称
  8. arch = "linux/amd64"  # 架构
  9. tag = "8.0.2"
  10. private_registry = "192.168.*.*:*"  # 私有 Registry 地址（HTTP）
  11.  
  12. # 允许不安全的 registry 连接
  13. os.environ['DOCKER_CONTENT_TRUST'] = '0'
  14.  
  15. public_image_with_tag = f"{public_image}:{tag}"
  16. private_image = f"{private_registry}/{public_image_with_tag}"  # 私有镜像名称
  17.  
  18. # 获取临时目录 - 兼容Windows和Linux
  19. temp_dir = os.getenv("TEMP") or os.getenv("TMPDIR") or "/tmp"
  20. tar_file_path = os.path.join(temp_dir, public_image_with_tag.replace(":", "_") + ".tar")  # 临时 tar 文件路径
  21. tar_file_path = tar_file_path.replace("\\", "_").replace("/", "_")  # 统一处理路径分隔符
  22.  
  23. def run_command(cmd, error_msg):
  24.     """运行命令并检查返回码"""
  25.     result = subprocess.run(cmd, shell=True)
  26.     if result.returncode != 0:
  27.         print(f"错误: {error_msg}")
  28.         exit(1)
  29.  
  30. # 拉取公共镜像
  31. print(f"正在拉取公共镜像: {public_image_with_tag}")
  32. if arch is None:
  33.     run_command(f"docker pull {public_image_with_tag}", f"拉取镜像失败: {public_image_with_tag}")
  34. else:
  35.     print(f"拉取架构为: {arch}")
  36.     run_command(f"docker pull {public_image_with_tag} --platform {arch}", f"拉取镜像失败: {public_image_with_tag}")
  37.  
  38. # 保存镜像为 tar 文件
  39. print(f"正在保存镜像为 tar 文件: {tar_file_path}")
  40. run_command(f"docker save -o {tar_file_path} {public_image_with_tag}", f"保存镜像失败: {public_image_with_tag}")
  41.  
  42. # 加载 tar 文件到 Docker
  43. print(f"正在加载 tar 文件到 Docker: {tar_file_path}")
  44. run_command(f"docker load -i {tar_file_path}", f"加载镜像失败: {tar_file_path}")
  45.  
  46. # 重新打标签（推送到私有 Registry 需要）
  47. print(f"正在重新打标签为: {private_image}")
  48. run_command(f"docker tag {public_image_with_tag} {private_image}", f"重新打标签失败: {public_image} -> {private_image}")
  49.  
  50. # 推送到私有 Registry
  51. print(f"正在推送到私有 Registry: {private_image}")
  52. run_command(f"docker push {private_image}", f"推送镜像失败: {private_image}")
  53.  
  54. # 验证镜像是否成功推送到私有 Registry
  55. print("验证镜像是否成功推送到私有 Registry...")
  56. registry_url = f"http://{private_registry}/v2/{public_image}/tags/list"
  57. print(registry_url)
  58.  
  59. try:
  60.     response = requests.get(registry_url)
  61.     response.raise_for_status()
  62.     result_obj = response.json()
  63.  
  64.     if public_image not in result_obj.get("name", ""):
  65.         print(f"镜像推送验证失败, {result_obj}")
  66.         exit(1)
  67.  
  68.     if tag in result_obj.get("tags", []):
  69.         print(f"镜像已成功推送到私有 Registry: {private_image}")
  70.     else:
  71.         print(f"镜像推送验证失败: {private_image}")
  72.         exit(1)
  73.  
  74. except requests.exceptions.RequestException as e:
  75.     print(f"验证请求失败: {e}")
  76.     exit(1)
  77.  
  78. # # 清理临时文件和本地镜像
  79. print("清理临时文件和本地镜像...")
  80. try:
  81.     os.remove(tar_file_path)
  82. except OSError as e:
  83.     print(f"删除临时文件失败: {e}")
  84.  
  85. run_command(f"docker rmi {public_image_with_tag}", f"删除本地镜像失败: {public_image_with_tag}")
  86. run_command(f"docker rmi {private_image}", f"删除本地镜像失败: {private_image}")
  87.  
  88. print("镜像转存完成！")