enihsyou/docker-mirror-speedtest.md

## docker-mirror-speedtest.md

      
    Raw
  

              docker-mirror-speedtest.md
            
          
    Docer Registry HTTP v2 API 需要使用 Bearer Token 才能访问，大大增加了访问难度，像 chsrc 这样的静态链接测速工具难以对 DockerHub 测速。
不测速就不知道各个镜像的质量，实际上很多网络上的镜像站点都是通过 Cloudflare Worker 搭建的。
我想要知道真实的下载速度，所以有了这个脚本。

大部分逻辑来自 enihsyou/docker-mirror-hammal 和 Claude 3.5 Sonnet + Gemini 2.5 Pro
Bash 版本兼容于 Linux/macOS，Powershell 版本兼容于 Windows + Powershell 7
镜像列表来自互联网，感谢发布者
本代码以 MIT 协议发布

使用方法

Linux / macOS (Bash)

用以下命令一键执行：
bash <(curl -sL "https://gist.githubusercontent.com/enihsyou/dafa9fca2e0db2623ecf7db500842b87/raw/docker-mirror-speedtest.sh")
Windows (Powershell)

用以下命令一键执行：
irm "https://gist.githubusercontent.com/enihsyou/dafa9fca2e0db2623ecf7db500842b87/raw/docker-mirror-speedtest.ps1" | iex
自定义目标主机

./docker-mirror-speedtest.sh hub2.nat.tf hub1.nat.tf --verbose

.\docker-mirror-speedtest.ps1 hub2.nat.tf -Verbose
示例输出

$ .\scripts\speed-test.ps1
🚀 开始 Docker 镜像站测速...
❌ [docker.1panel.live] 测速异常: 无法获取认证信息 (403 Forbidden)
❌ [docker.1panel.top] 测速异常: 无法获取认证信息 (429 Too Many Requests)
🔑 [docker.m.daocloud.io] 需要认证，正在获取token...
📍 [docker.m.daocloud.io] 获取下载地址...
🏃 [docker.m.daocloud.io] 正在测速...
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   513  100   513    0     0   6772      0 --:--:-- --:--:-- --:--:--  6932
100 20.0M  100 20.0M    0     0  21.6M      0 --:--:-- --:--:-- --:--:-- 40.4M
📊 [docker.m.daocloud.io] 测速结果: DNS解析时间: 0.016131s 连接时间: 0.030032s 总时间: 0.925078s 下载速度: 22670002B/s
❌ [docker.ketches.cn] 测速异常: 无法获取认证信息 (429 Too Many Requests)
🔑 [docker.1ms.run] 需要认证，正在获取token...
📍 [docker.1ms.run] 获取下载地址...
🏃 [docker.1ms.run] 正在测速...
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 20.0M  100 20.0M    0     0  11.1M      0  0:00:01  0:00:01 --:--:-- 11.1M
📊 [docker.1ms.run] 测速结果: DNS解析时间: 0.013340s 连接时间: 0.016066s 总时间: 1.790445s 下载速度: 11713021B/s
🚩 [hub1.nat.tf] 无需认证，直接测速...
🏃 [hub1.nat.tf] 正在测速...
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 41.8M  100 41.8M    0     0  8974k      0  0:00:04  0:00:04 --:--:-- 10.6M
📊 [hub1.nat.tf] 测速结果: DNS解析时间: 0.013273s 连接时间: 0.015449s 总时间: 4.778494s 下载速度: 9189851B/s
🚩 [hub2.nat.tf] 无需认证，直接测速...
🏃 [hub2.nat.tf] 正在测速...
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 41.8M  100 41.8M    0     0  12.3M      0  0:00:03  0:00:03 --:--:-- 12.3M
📊 [hub2.nat.tf] 测速结果: DNS解析时间: 0.009979s 连接时间: 0.013772s 总时间: 3.385906s 下载速度: 12969542B/s
🔑 [hub.rat.dev] 需要认证，正在获取token...
📍 [hub.rat.dev] 获取下载地址...
🏃 [hub.rat.dev] 正在测速...
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   264  100   264    0     0    197      0  0:00:01  0:00:01 --:--:--   197
100 20.0M  100 20.0M    0     0  66379      0  0:05:15  0:05:15 --:--:-- 32764
📊 [hub.rat.dev] 测速结果: DNS解析时间: 0.017731s 连接时间: 0.043667s 总时间: 315.935037s 下载速度: 66379B/s


═══════════════ 测试总结 ═══════════════

% 基本统计
✓ 成功测试: 5
✗ 失败测试: 3
Σ 总计测试: 8

🏆 速度排名:
🥇 [ 1] hub2.nat.tf                      25.71 MB/s
🥈 [ 2] docker.m.daocloud.io              2.92 MB/s
🥉 [ 3] hub1.nat.tf                       1.64 MB/s
   [ 4] docker.1ms.run                  378.74 KB/s
   [ 5] hub.rat.dev                     345.19 KB/s

❌ 失败详情:
  • docker.1panel.live
    无法获取认证信息 (403 Forbidden)
  • docker.1panel.top
    无法获取认证信息 (429 Too Many Requests)
  • docker.ketches.cn
    无法获取认证信息 (429 Too Many Requests)
Bonus Tip

除了用这个脚本进行测速，还有更简单的办法，就是加上完整前缀直接把镜像拉下来。
docker pull docker.gh-proxy.com/nginx
docker pull docker.gh-proxy.com/library/nginx
docker pull docker.gh-proxy.com/docker.io/library/nginx

docker pull docker.1ms.run/library/python
docker pull ghcr.nju.edu.cn/openfaas/queue-worker
docker pull gcr.linkos.org/kubebuilder/kube-rbac-proxy:v0.13.1
这个方法的好处是简单，并且适配任意镜像站，即便是需要 login 授权的；
但缺点是会写入磁盘，同名镜像会缓存，要记得删除下载的镜像；测速也得靠记时器或者人工感知。
v2 版

2025-12 新增了使用 Python 实现的 v2 版，它有更好的扩展性和行为表现
$ uv run .\speedtest.py
                            Speed Rankings
┏━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┳━━━━━━━━━━┓
┃ Rank ┃ Domain               ┃       Speed ┃ Duration (s) ┃     Size ┃
┡━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━╇━━━━━━━━━━┩
│  1   │ docker.m.ixdev.cn    │  68.17 MB/s │         0.61 │ 41.88 MB │
│  2   │ docker.1ms.run       │   5.49 MB/s │         7.62 │ 41.88 MB │
│  3   │ hub2.nat.tf          │   2.11 MB/s │    10.01 (T) │ 21.14 MB │
│  4   │ docker.cattt.net     │   1.76 MB/s │    10.02 (T) │ 17.63 MB │
│  5   │ docker.m.daocloud.io │   1.21 MB/s │    10.01 (T) │ 12.15 MB │
│  6   │ docker.etcd.fun      │ 969.77 KB/s │    10.32 (T) │  9.77 MB │
└──────┴──────────────────────┴─────────────┴──────────────┴──────────┘
                      ✖ Failure Details
┏━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Domain             ┃ Error                                 ┃
┡━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ docker.xuanyuan.me │ Initial probe failed with status 404. │
│ hub1.nat.tf        │ Initial probe failed with status 500. │
│ hub.amingg.com     │ Initial probe failed with status 403. │
│ hub.rat.dev        │ Cannot connect to registry: SSLError  │
└────────────────────┴───────────────────────────────────────┘

$ uv run .\speedtest.py --group ghcr
                         Speed Rankings
┏━━━━━━┳━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━━┳━━━━━━━━━━┓
┃ Rank ┃ Domain          ┃      Speed ┃ Duration (s) ┃     Size ┃
┡━━━━━━╇━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━━╇━━━━━━━━━━┩
│  1   │ ghcr.nju.edu.cn │ 75.53 MB/s │         0.75 │ 56.95 MB │
└──────┴─────────────────┴────────────┴──────────────┴──────────┘

  
## docker-mirror-speedtest.ps1
# 本 Powershell 脚本用于 DockerHub 镜像的测速
# 传入域名列表，自动测速并按速度输出排名
#
# 使用默认配置
# .\docker-mirror-speedtest.ps1
#
# 自定义域名和测试目标
# .\docker-mirror-speedtest.ps1 -Domains @("docker.1ms.run", "hub1.nat.tf")

[CmdletBinding()]
param (
    [Parameter(ValueFromPipeline = $true)]
    [string[]]$Domains = @(
        # list from status monitor sites
        # https://status.anye.xyz/status/docker
        # https://status.1panel.top/status/docker
        "docker.m.daocloud.io",
        "docker.xuanyuan.me",
        "docker.1ms.run",
        "docker.m.ixdev.cn",
        "hub1.nat.tf",
        "hub2.nat.tf",
        "hub.amingg.com",
        "hub.rat.dev",
        "docker.etcd.fun",
        "docker.cattt.net"
    )
)

# 设置输出编码为UTF8，以支持特殊字符
$OutputEncoding = [Console]::OutputEncoding = [System.Text.Encoding]::UTF8

$TestTarget = @{
    Path   = "library/nginx" # size: 43,913,652
    Digest = "sha256:8c18fec49d4bdf0813737297228f0afbdc26ed7f8f3fd7e66048114d47d03168"
}

# 定义颜色输出函数
function Write-ColorOutput {
    param ([string]$Text, [string]$Color = "White")
    Write-Host $Text -ForegroundColor $Color
}

function Parse-WwwAuthenticate {
    param (
        [string]$AuthStr,
        [hashtable]$Target
    )

    $splits = $AuthStr -split '\s+'
    $bearer = $splits[0]
    if (!$bearer -or $bearer.ToLower() -ne 'bearer') {
        throw "Invalid Www-Authenticate: $AuthStr"
    }

    $params = $splits[1..($splits.Length - 1)] -join '' -split ','
    $result = @{}
    foreach ($param in $params) {
        $kvPair = $param -split '=', 2
        if ($kvPair.Length -eq 2) {
            $k = $kvPair[0].Trim()
            $v = $kvPair[1].Trim() -replace '^"|"$', ''
            $result[$k] = $v
        }
    }
    $result["scope"] = "respository:$($Target.Path):pull"
    return $result
}

function Get-DockerToken {
    param (
        [string]$Realm,
        [string]$Service,
        [string]$Scope
    )

    $tokenUrl = [System.UriBuilder]$Realm
    $query = [System.Web.HttpUtility]::ParseQueryString('')
    $query['service'] = $Service
    $query['scope'] = $Scope
    $tokenUrl.Query = $query.ToString()

    $response = Invoke-RestMethod -Uri $tokenUrl.Uri.ToString() -Method Get
    return $response.token
}

# 添加结果收集类
class SpeedTestResult {
    [string]$Domain
    [double]$Speed
    [double]$ConnectTime
    [bool]$Success
    [string]$Error

    [string] GetSpeedString() {
        if (-not $this.Success) { return "N/A" }
        $nums = $this.Speed
        $unit = "B/s"
        if ($nums -gt 1024 * 1024) {
            $nums = [double]([math]::Round($nums / (1024 * 1024), 2))
            $unit = "MB/s"
        } elseif ($nums -gt 1024) {
            $nums = [double]([math]::Round($nums / 1024, 2))
            $unit = "KB/s"
        }
        return "$nums $unit"
    }
}

$global:Results = [System.Collections.ArrayList]::new()

function Invoke-CurlCommand {
    param (
        [Parameter(Mandatory, Position = 0)]
        [string[]]$Args,
        [Parameter(Mandatory, Position = 1)]
        [string]$Url
    )

    # 构建完整参数数组
    $fullArgs = $Args + @(
        # as some site like 1panel.live blocks curl user-agent
        "--user-agent", "docker-mirror-speedtest.ps1",
        "$Url"
    )

    # 调试输出完整命令
    Write-Verbose "即将执行: curl $($fullArgs -join ' ')"
    return & curl $fullArgs
}

function Test-RegistrySpeed {
    param (
        [string]$Domain,
        [hashtable]$Target
    )

    $dsize = 41943040 # 40MB
    $skill = 102400   # 100KB/s

    $result = [SpeedTestResult]::new()
    $result.Domain = $Domain

    try {
        Write-Verbose "开始测试域名: $Domain"
        $probAuthUrl = "https://$Domain/v2/library/alpine/manifests/latest"
        $downloadUrl = "https://$Domain/v2/$($Target.Path)/blobs/$($Target.Digest)"
        Write-Verbose "测试URL: $downloadUrl"

        # 初始请求
        $args = @( "-I", "-L", "--max-time", "8" )
        $response = Invoke-CurlCommand $args $probAuthUrl
        Write-Verbose "收到响应:`n$($response | Out-String)"

        # 检查状态码
        $statusLine = $response | Where-Object { $_ -match "^HTTP/.+ (\d{3}) " } | Select-Object -First 1
        if ($statusLine -match "200") {
            # 直接可访问，无需认证
            Write-ColorOutput "🚩 [$Domain] 无需认证，直接测速..."
            $redirectUrl = $downloadUrl
        } else {
            # 需要认证，解析WWW-Authenticate头
            $authHeader = $response | Where-Object { $_ -match "Www-Authenticate: " } | Select-Object -First 1
            if ($authHeader) {
                Write-ColorOutput "🔑 [$Domain] 需要认证，正在获取token..."
                $authStr = ($authHeader -split ": ")[1]
                Write-Verbose "认证字符串: $authStr"

                $auth = Parse-WwwAuthenticate $authStr -Target $Target
                Write-Verbose "解析后的认证信息: $($auth | ConvertTo-Json)"

                $token = Get-DockerToken -Realm $auth.realm -Service $auth.service -Scope "repository:$($Target.Path)`:pull"
                Write-Verbose "获取到token: $($token.Substring(0, [Math]::Min(20, $token.Length)))..."

                # 使用token获取重定向URL
                Write-ColorOutput "📍 [$Domain] 获取下载地址..."
                $args = @(
                    "-s", "-I",
                    "-H", "Authorization: Bearer $token",
                    "-w", "%{redirect_url}",
                    "-o", "nul")
                $redirectUrl = Invoke-CurlCommand $args $downloadUrl
                Write-Verbose "重定向URL: $redirectUrl"
                if ($null -eq $redirectUrl) {
                    throw "资源重定向失败"
                }
            } else {
                $statusLine = $response | Where-Object { $_ -match "^HTTP/.+ (\d{3}.+)" } | ForEach-Object { $Matches.1 }
                if ($statusLine) {
                    throw "无法获取认证信息 ($statusLine)"
                }
                $statusLine = $response | Where-Object { $_ -match "(curl:.+)" } | ForEach-Object { $Matches.1 }
                if ($statusLine) {
                    throw "其他异常 ($statusLine)"
                }
                throw "未知错误: $($response | Out-String)"
            }
        }

        # 进行速度测试
        Write-ColorOutput "🏃 [$Domain] 正在测速..." "Blue"

        $curl_args = @(
            "-L",
            "-w", "DNS解析时间: %{time_namelookup}s 连接时间: %{time_connect}s 总时间: %{time_total}s 下载速度: %{speed_download}B/s ",
            "-o", "nul",
            "--range", "0-$dsize",
            "--max-time", "15",
            "--speed-time", "3",
            "--speed-limit", "$skill"
        )
        if ($token) {
            $curl_args += @("-H", "Authorization: Bearer $token")
        }

        $resultOutput = Invoke-CurlCommand $curl_args $redirectUrl
        Write-ColorOutput "📊 [$Domain] 测速结果: $resultOutput"

        $resultOutput -match '连接时间: (?<ctime>\d+\.?\d*)s.+下载速度: (?<speed>\d+\.?\d*)B/s' | Out-Null

        # 解析测速结果
        $result.Speed = $Matches.speed
        $result.ConnectTime = $Matches.ctime
        $result.Success = $true
    } catch {
        $result.Error = $_.Exception.Message
        $result.Success = $false
        Write-ColorOutput "❌ [$Domain] 测速异常: $($result.Error)" "Red"
    }

    [void]$global:Results.Add($result)
    return $result
}

function Show-TestSummary {
    $successful = $global:Results | Where-Object { $_.Success }
    $failed = $global:Results | Where-Object { -not $_.Success }

    Write-Host "`n`n═══════════════ 测试总结 ═══════════════" -ForegroundColor Magenta

    # 统计信息
    Write-Host "✓ 成功测试: $($successful.Count)" -ForegroundColor Green
    Write-Host "✗ 失败测试: $($failed.Count)" -ForegroundColor Red
    Write-Host "Σ 总计测试: $($global:Results.Count)" -ForegroundColor Cyan

    # 显示排名
    if ($successful.Count -gt 0) {
        Write-Host "`n🏆 速度排名:" -ForegroundColor Yellow
        $rank = 1
        $successful | Sort-Object Speed -Descending | ForEach-Object {
            $color = switch ($rank) {
                1 { "Yellow" }
                2 { "White" }
                3 { "DarkYellow" }
                default { "Gray" }
            }

            $medal = switch ($rank) {
                1 { "🥇" }
                2 { "🥈" }
                3 { "🥉" }
                default { "  " }
            }

            Write-Host ("{0} [{1,2}] {2,-30} {3,10}" -f $medal,
                $rank,
                "$($_.Domain)",
                $_.GetSpeedString()) -ForegroundColor $color
            $rank++
        }
    }

    # 失败清单
    if ($failed.Count -gt 0) {
        Write-Host "`n❌ 失败详情:" -ForegroundColor Red
        $failed | ForEach-Object {
            Write-Host "  • $($_.Domain)" -ForegroundColor DarkRed
            Write-Host "    $($_.Error)" -ForegroundColor DarkGray
        }
    }
}

# 主程序入口
Write-ColorOutput "🚀 开始 Docker 镜像站测速..." "Cyan"
Write-Verbose "测试域名列表: $($Domains -join ', ')"

try {
    foreach ($domain in $Domains) {
        Test-RegistrySpeed -Domain $domain -Target $TestTarget | Out-Null
    }
} finally {
    # 完成后显示总结
    Show-TestSummary
}

## docker-mirror-speedtest.sh
#!/bin/bash
#
# 本 Bash 脚本用于 DockerHub 镜像的测速
# 传入域名列表，自动测速并按速度输出排名
#
# 使用方法:
#   ./docker-mirror-speedtest.sh [选项] [域名1 域名2 ...]
#
# 选项:
#   -v, --verbose    启用详细输出模式
#
# 示例:
#   使用默认配置:
#   ./docker-mirror-speedtest.sh
#
#   启用详细输出并测试指定域名:
#   ./docker-mirror-speedtest.sh --verbose docker.1ms.run hub1.nat.tf
#

# --- 配置 ---

# 默认域名列表
# list from status monitor sites
# https://status.anye.xyz/status/docker
# https://status.1panel.top/status/docker
DEFAULT_DOMAINS=(
    "docker.m.daocloud.io"
    "docker.xuanyuan.me"
    "docker.1ms.run"
    "docker.m.ixdev.cn"
    "hub1.nat.tf"
    "hub2.nat.tf"
    "hub.amingg.com"
    "hub.rat.dev"
    "docker.etcd.fun"
    "docker.cattt.net"
)

# 测试目标
TEST_TARGET_PATH="library/nginx"
TEST_TARGET_DIGEST="sha256:8c18fec49d4bdf0813737297228f0afbdc26ed7f8f3fd7e66048114d47d03168"

# --- 全局变量 ---
DOMAINS=()
VERBOSE=false
SUCCESSFUL_RESULTS=()
FAILED_RESULTS=()

# 颜色定义
COLOR_RED='\033[0;31m'
COLOR_GREEN='\033[0;32m'
COLOR_YELLOW='\033[0;33m'
COLOR_BLUE='\033[0;34m'
COLOR_MAGENTA='\033[0;35m'
COLOR_CYAN='\033[0;36m'
COLOR_WHITE='\033[0;97m'
COLOR_GRAY='\033[0;90m'
COLOR_NC='\033[0m' # No Color

# --- 函数定义 ---

# 打印带颜色的输出
# 参数1: 文本
# 参数2: 颜色 (Red, Green, Yellow, Blue, Magenta, Cyan, White, Gray)
write_color_output() {
    local text="$1"
    local color_name="$2"
    local color_code="${COLOR_NC}"
    case "$color_name" in
        "Red") color_code="${COLOR_RED}" ;;
        "Green") color_code="${COLOR_GREEN}" ;;
        "Yellow") color_code="${COLOR_YELLOW}" ;;
        "Blue") color_code="${COLOR_BLUE}" ;;
        "Magenta") color_code="${COLOR_MAGENTA}" ;;
        "Cyan") color_code="${COLOR_CYAN}" ;;
        "White") color_code="${COLOR_WHITE}" ;;
        "Gray") color_code="${COLOR_GRAY}" ;;
    esac
    printf "%b%s%b\n" "${color_code}" "${text}" "${COLOR_NC}"
}

# 详细日志输出函数
# 仅在 VERBOSE 模式下打印
log_verbose() {
    if [ "$VERBOSE" = true ]; then
        write_color_output "DEBUG: $*" "Gray"
    fi
}

# 解析命令行参数
parse_args() {
    local custom_domains=()
    for arg in "$@"; do
        case "$arg" in
            -v|--verbose)
                VERBOSE=true
                shift
                ;;
            *)
                custom_domains+=("$arg")
                shift
                ;;
        esac
    done

    if [ ${#custom_domains[@]} -gt 0 ]; then
        DOMAINS=("${custom_domains[@]}")
    else
        DOMAINS=("${DEFAULT_DOMAINS[@]}")
    fi
}

# 格式化速度
# 参数1: 速度 (B/s)
format_speed() {
    local speed_bps
    speed_bps=$(printf "%.0f" "$1")
    if (( speed_bps > 1024 * 1024 )); then
        printf "%.2f MB/s" "$(bc <<< "scale=2; $speed_bps / (1024*1024)")"
    elif (( speed_bps > 1024 )); then
        printf "%.2f KB/s" "$(bc <<< "scale=2; $speed_bps / 1024")"
    else
        printf "%d B/s" "$speed_bps"
    fi
}

# 主测速函数
# 参数1: 域名
test_registry_speed() {
    local domain="$1"
    local dsize=41943040 # 40MB
    local skill=102400   # 100KB/s
    local token=""
    local redirect_url=""
    local response
    local http_status
    local auth_header
    local error_message

    local prob_auth_url="https://$domain/v2/library/alpine/manifests/latest"
    local download_url="https://$domain/v2/$TEST_TARGET_PATH/blobs/$TEST_TARGET_DIGEST"
    log_verbose "[$domain] 探测URL: $prob_auth_url"

    # 初始请求获取认证信息
    response=$(curl --user-agent "docker-mirror-speedtest.sh" -I -L --max-time 8 "$prob_auth_url")
    escaped_verbose_response=$(printf "%s\n%s" "[$domain] 探测响应:" "$response")
    log_verbose "$escaped_verbose_response"

    http_status=$(echo "$response" | grep -oP '^HTTP/[\d\.]+\s+\K\d{3}' | tail -n 1)

    if [[ "$http_status" == "200" ]]; then
        write_color_output "🚩 [$domain] 无需认证，直接测速..."
        redirect_url="$download_url"
    else
        auth_header=$(echo "$response" | grep -i "Www-Authenticate")

        if [ -n "$auth_header" ]; then
            write_color_output "🔑 [$domain] 需要认证，正在获取token..."
            log_verbose "[$domain] 认证头: $auth_header"

            local realm service scope
            realm=$(echo "$auth_header" | grep -oP 'realm="([^"]+)"' | cut -d'"' -f2)
            service=$(echo "$auth_header" | grep -oP 'service="([^"]+)"' | cut -d'"' -f2)
            scope="repository:$TEST_TARGET_PATH:pull"

            if [ -z "$realm" ]; then
                error_message="无法解析认证 Realm"
                FAILED_RESULTS+=("$domain|$error_message")
                write_color_output "❌ [$domain] 测速异常: $error_message" "Red"
                return
            fi

            log_verbose "[$domain] Realm='$realm', Service='$service', Scope='$scope'"

            local token_url="$realm?service=$service&scope=$scope"
            local token_response
            token_response=$(curl -sL "$token_url")
            token=$(echo "$token_response" | grep -oP '"token":"\K[^" ]+')

            if [ -z "$token" ]; then
                error_message="获取 Token 失败"
                FAILED_RESULTS+=("$domain|$error_message")
                write_color_output "❌ [$domain] 测速异常: $error_message" "Red"
                log_verbose "[$domain] Token响应: $token_response"
                return
            fi
            log_verbose "[$domain] 成功获取Token."

            write_color_output "📍 [$domain] 获取下载地址..."
            redirect_url=$(curl -s -I -H "Authorization: Bearer $token" -w "%{redirect_url}" -o /dev/null "$download_url")

            if [ -z "$redirect_url" ]; then
                error_message="资源重定向失败"
                FAILED_RESULTS+=("$domain|$error_message")
                write_color_output "❌ [$domain] 测速异常: $error_message" "Red"
                return
            fi
            log_verbose "[$domain] 重定向URL: $redirect_url"
        else
            error_message=$(echo "$response" | grep -oP '^HTTP/[\d\.]+\s+\K.+\b')
            if [ -z "$error_message" ]; then
                error_message=$(echo "$response" | grep -oP 'curl: \(\d+\) \K.+')
            fi
            if [ -z "$error_message" ]; then
                error_message="未知错误"
            fi
            FAILED_RESULTS+=("$domain|无法获取认证信息 ($error_message)")
            write_color_output "❌ [$domain] 测速异常: 无法获取认证信息 ($error_message)" "Red"
            return
        fi
    fi

    write_color_output "🏃 [$domain] 正在测速..." "Blue"

    local curl_args=(
        "-L"
        "-w" "DNS解析时间: %{time_namelookup}s 连接时间: %{time_connect}s 总时间: %{time_total}s 下载速度: %{speed_download}B/s"
        "-o" "/dev/null"
        "--range" "0-$dsize"
        "--max-time" "15"
        "--speed-time" "3"
        "--speed-limit" "$skill"
    )
    if [ -n "$token" ]; then
        curl_args+=("-H" "Authorization: Bearer $token")
    fi

    log_verbose "[$domain] 执行测速命令: curl ${curl_args[*]} $redirect_url"
    local result_output
    result_output=$(curl "${curl_args[@]}" "$redirect_url")

    write_color_output "📊 [$domain] 测速结果: $result_output"

    local speed
    speed=$(echo "$result_output" | grep -oP '下载速度: \K[\d\.]+')

    if [ -n "$speed" ] && [ "$(printf "%.0f" "$speed")" -gt 0 ]; then
        SUCCESSFUL_RESULTS+=("$domain|$speed")
    else
        FAILED_RESULTS+=("$domain|速度测试失败或速度为0")
        write_color_output "❌ [$domain] 测速异常: 速度测试失败或速度为0" "Red"
    fi
}

# 显示总结
show_test_summary() {
    local successful_count=${#SUCCESSFUL_RESULTS[@]}
    local failed_count=${#FAILED_RESULTS[@]}
    local total_count=$((successful_count + failed_count))

    printf "\n"
    write_color_output "═══════════════ 测试总结 ═══════════════" "Magenta"

    write_color_output "✓ 成功测试: $successful_count" "Green"
    write_color_output "✗ 失败测试: $failed_count" "Red"
    write_color_output "Σ 总计测试: $total_count" "Cyan"

    if [ "$successful_count" -gt 0 ]; then
        printf "\n"
        write_color_output "🏆 速度排名:" "Yellow"

        # 排序并显示
        # 使用 printf 将每个元素逐行输出，再用 mapfile/readarray 安全读取到数组，避免命令替换拆分问题
        local sorted_results=()
        mapfile -t sorted_results < <(printf "%s\n" "${SUCCESSFUL_RESULTS[@]}" | sort -t'|' -k2 -nr)

        local rank=1
        for result in "${sorted_results[@]}"; do
            local domain speed speed_str color medal
            domain=$(echo "$result" | cut -d'|' -f1)
            speed=$(echo "$result" | cut -d'|' -f2)
            speed_str=$(format_speed "$speed")

            case "$rank" in
                1) color="${COLOR_YELLOW}" ; medal="🥇" ;;
                2) color="${COLOR_WHITE}" ; medal="🥈" ;;
                3) color="${COLOR_GRAY}" ; medal="🥉" ;;
                *) color="${COLOR_NC}" ; medal="  " ;;
            esac

            printf "%b%s [%2d] %-30s %12s%b\n" "$color" "$medal" "$rank" "$domain" "$speed_str" "${COLOR_NC}"
            ((rank++))
        done
    fi

    if [ "$failed_count" -gt 0 ]; then
        printf "\n"
        write_color_output "❌ 失败详情:" "Red"
        for result in "${FAILED_RESULTS[@]}"; do
            local domain error
            domain=$(echo "$result" | cut -d'|' -f1)
            error=$(echo "$result" | cut -d'|' -f2)
            write_color_output "  • $domain" "Red"
            write_color_output "    $error" "Gray"
        done
    fi
}

# --- 主程序 ---
main() {
    parse_args "$@"

    log_verbose "详细模式已启用."
    log_verbose "待测试域名: ${DOMAINS[*]}"

    write_color_output "🚀 开始 Docker 镜像站测速..." "Cyan"
    for domain in "${DOMAINS[@]}"; do
        test_registry_speed "$domain"
    done

    show_test_summary
}

main "$@"

## requirements.txt
requests
rich

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

import argparse
import re
import sys
import time
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

import requests
from rich.console import Console, Group
from rich.live import Live
from rich.panel import Panel
from rich.progress import (
    BarColumn,
    DownloadColumn,
    Progress,
    TextColumn,
    TimeElapsedColumn,
    TransferSpeedColumn,
)
from rich.table import Table

# --- Configuration ---


@dataclass
class TestTargetConfig:
    """Configuration for a specific test target."""

    probe_image: str
    test_image: str
    test_digest: str


TEST_DOMAINS: Dict[str, List[str]] = {
    "default": [
        "docker.m.daocloud.io",
        "docker.xuanyuan.me",
        "docker.1ms.run",
        "docker.m.ixdev.cn",
        "hub1.nat.tf",
        "hub2.nat.tf",
        "hub.amingg.com",
        "hub.rat.dev",
        "docker.etcd.fun",
        "docker.cattt.net",
    ],
    "ghcr": [
        "ghcr.nju.edu.cn",
    ],
}

# Configuration for test targets. This allows specifying different images
# for different domains if needed.
# The 'default' key is used for any domain not explicitly listed.
TEST_CONFIG: Dict[str, TestTargetConfig] = {
    "default": TestTargetConfig(
        probe_image="library/alpine:latest",
        test_image="library/nginx",
        test_digest="sha256:8c18fec49d4bdf0813737297228f0afbdc26ed7f8f3fd7e66048114d47d03168",
    ),
    "ghcr": TestTargetConfig(
        probe_image="hasscc/hacn:latest",
        test_image="hasscc/hacn",
        test_digest="sha256:e02a65517b4cc86d5c652f5b6601d5c51d446c55fd63dbf87b35448dd910bc6c",  # 59,716,988 B
    ),
}


# --- Main Speed Test Logic ---


class SpeedTester:
    """
    Encapsulates the logic for testing the speed of a single Docker registry mirror.
    """

    def __init__(
        self,
        domain: str,
        config: TestTargetConfig,
        console: Console,
        progress: Progress,
        verbose: bool = False,
        timeout: int = 10,
    ):
        self.domain = domain
        self.config = config
        self.console = console
        self.progress = progress
        self.verbose = verbose
        self.timeout = timeout
        self.auth_token: Optional[str] = None
        self.session = requests.Session()
        self.session.headers.update(
            {"User-Agent": "Docker-Mirror-Speedtest-Python/1.0"}
        )

    def _log_verbose(self, message: str):
        """Prints a message if verbose mode is enabled."""
        if self.verbose:
            self.console.print(f"[dim]DEBUG: {message}[/dim]")

    def _get_auth_token(self) -> Tuple[Optional[str], Optional[str]]:
        """
        Probes the registry to see if authentication is needed and, if so, fetches a bearer token.
        Returns a tuple of (token, error_message).
        """
        probe_image = self.config.probe_image
        [pimage, ptag] = probe_image.split(":")
        probe_url = f"https://{self.domain}/v2/{pimage}/manifests/{ptag}"
        self._log_verbose(f"Probing for auth at {probe_url}")

        try:
            response = self.session.head(probe_url, allow_redirects=True, timeout=10)
            self._log_verbose(f"Probe response status: {response.status_code}")
            self._log_verbose(f"Probe response headers: {response.headers}")

            if response.status_code == 200:
                self.console.print(f"✔ [{self.domain}] No authentication required.")
                return None, None

            if (
                response.status_code in (401, 403)
                and "Www-Authenticate" in response.headers
            ):
                self.console.print(
                    f"🔑 [{self.domain}] Authentication required, fetching token..."
                )
                auth_header = response.headers["Www-Authenticate"]
                self._log_verbose(f"Auth header: {auth_header}")

                realm_match = re.search('realm="([^"]+)"', auth_header)
                service_match = re.search('service="([^"]+)"', auth_header)

                if not realm_match:
                    return None, "Could not parse realm from Www-Authenticate header."

                realm = realm_match.group(1)
                service = service_match.group(1) if service_match else ""
                scope = f"repository:{self.config.test_image}:pull"

                auth_params = {"service": service, "scope": scope}

                self._log_verbose(
                    f"Requesting token from {realm} with params {auth_params}"
                )
                token_response = self.session.get(realm, params=auth_params, timeout=10)
                token_response.raise_for_status()

                token = token_response.json().get("token")
                if not token:
                    return None, "Failed to retrieve token from auth server."

                self._log_verbose("Successfully obtained auth token.")
                return token, None

            return None, f"Initial probe failed with status {response.status_code}."

        except requests.exceptions.RequestException as e:
            self._log_verbose(f"Auth request exception: {e}")
            return None, f"Cannot connect to registry: {e.__class__.__name__}"

    def run_test(self) -> Dict:
        """
        Runs the full speed test for the domain.
        Returns a dictionary with the results.
        """
        try:
            token, error = self._get_auth_token()
            if error:
                return {"domain": self.domain, "status": "failed", "error": error}
            if token:
                self.auth_token = token
                self.session.headers.update(
                    {"Authorization": f"Bearer {self.auth_token}"}
                )
        except Exception as e:
            return {
                "domain": self.domain,
                "status": "failed",
                "error": f"Authentication error: {e}",
            }

        test_image = self.config.test_image
        test_digest = self.config.test_digest
        download_url = f"https://{self.domain}/v2/{test_image}/blobs/{test_digest}"
        self._log_verbose(f"Starting download from {download_url}")

        task_id = None
        timed_out = False
        try:
            with self.session.get(download_url, stream=True, timeout=self.timeout + 5) as r:
                self._log_verbose(f"Download response headers: {r.headers}")
                r.raise_for_status()
                total_size = int(r.headers.get("content-length", 0))

                task_id = self.progress.add_task(
                    f"⬇️ [bold]{self.domain}[/]",
                    total=total_size,
                    start=True,
                )

                download_start_time = time.time()
                bytes_downloaded = 0

                for chunk in r.iter_content(chunk_size=8192):
                    self.progress.update(task_id, advance=len(chunk))
                    bytes_downloaded += len(chunk)

                    elapsed_time = time.time() - download_start_time
                    if elapsed_time > self.timeout:
                        timed_out = True
                        break

                download_time = time.time() - download_start_time

            if download_time > 0:
                speed_Bps = bytes_downloaded / download_time
            else:
                speed_Bps = 0

            return {
                "domain": self.domain,
                "status": "success",
                "speed_Bps": speed_Bps,
                "size_bytes": bytes_downloaded,
                "duration_sec": download_time,
                "timed_out": timed_out,
            }

        except requests.exceptions.RequestException as e:
            self._log_verbose(f"Download exception: {e}")
            error_msg = f"Download failed: {e.__class__.__name__}"
            if e.response is not None:
                error_msg += f" (Status: {e.response.status_code})"
            return {"domain": self.domain, "status": "failed", "error": error_msg}
        except Exception as e:
            self._log_verbose(f"An unexpected error occurred: {e}")
            return {
                "domain": self.domain,
                "status": "failed",
                "error": f"An unexpected error occurred: {e}",
            }
        finally:
            if task_id is not None:
                self.progress.remove_task(task_id)


# --- UI and Formatting ---


def format_speed(speed_Bps: float) -> str:
    """Formats speed in B/s to a human-readable string (MB/s, KB/s, B/s)."""
    if speed_Bps > 1024 * 1024:
        return f"{speed_Bps / (1024 * 1024):.2f} MB/s"
    if speed_Bps > 1024:
        return f"{speed_Bps / 1024:.2f} KB/s"
    return f"{speed_Bps:.2f} B/s"


def display_summary(results: List[Dict], console: Console):
    """Displays a final summary of all test results."""
    successful = sorted(
        [r for r in results if r["status"] == "success"],
        key=lambda r: r["speed_Bps"],
        reverse=True,
    )
    failed = [r for r in results if r["status"] == "failed"]

    console.print("\n" + "=" * 20 + " Test Summary " + "=" * 20, style="bold magenta")

    stats_table = Table.grid(padding=(0, 2))
    stats_table.add_column()
    stats_table.add_column()
    stats_table.add_row(
        "[green]■[/green] [bold green]Successful tests:[/]", str(len(successful))
    )
    stats_table.add_row("[red]■[/red] [bold red]Failed tests:[/]", str(len(failed)))
    stats_table.add_row("[cyan]■[/cyan] [bold cyan]Total tests:[/]", str(len(results)))
    console.print(Panel(stats_table, title="Statistics", border_style="cyan"))

    if successful:
        table = Table(
            title="Speed Rankings", title_style="bold yellow", border_style="green"
        )
        table.add_column("Rank", style="cyan", justify="center")
        table.add_column("Domain", style="bold white")
        table.add_column("Speed", style="green", justify="right")
        table.add_column("Duration (s)", style="dim", justify="right")
        table.add_column("Size", style="dim", justify="right")

        for i, res in enumerate(successful, 1):
            size_mb = f"{res['size_bytes'] / (1024 * 1024):.2f} MB"
            duration_str = f"{res['duration_sec']:.2f}"
            if res.get("timed_out", False):
                duration_str += " (T)"

            table.add_row(
                str(i),
                res["domain"],
                format_speed(res["speed_Bps"]),
                duration_str,
                size_mb,
            )
        console.print(table)

    if failed:
        fail_table = Table(
            title="Failure Details", title_style="bold red", border_style="red"
        )
        fail_table.add_column("Domain", style="bold white")
        fail_table.add_column("Error", style="yellow")
        for res in failed:
            fail_table.add_row(res["domain"], res["error"])
        console.print(fail_table)


# --- Main Execution ---


def main():
    """Main function to parse arguments and run the speed tests."""
    parser = argparse.ArgumentParser(
        description="Test the speed of Docker registry mirrors.",
        formatter_class=argparse.RawTextHelpFormatter,
    )
    parser.add_argument(
        "-g",
        "--group",
        nargs="?",
        choices=TEST_DOMAINS.keys(),
        help="Predefined group of domains to test (overrides individual domains).",
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=int,
        default=10,
        help="Maximum download time in seconds for each test. Default is 10s.",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose output for debugging.",
    )
    parser.add_argument(
        "domains",
        nargs="*",
        help="One or more domain names to test. If not provided, a default list is used.",
    )
    args = parser.parse_args()

    console = Console()
    console.print("🚀 [bold cyan]Starting Docker mirror speed test...[/bold cyan]")

    results = []

    # Define two separate progress bars
    overall_progress = Progress(
        TextColumn("[bold blue]Overall Progress"),
        BarColumn(),
        "[progress.percentage]{task.percentage:>3.0f}%",
        TextColumn("({task.completed} of {task.total})"),
        TimeElapsedColumn(),
    )
    download_progress = Progress(
        TextColumn("[progress.description]{task.description}", justify="left"),
        BarColumn(bar_width=None),
        "[progress.percentage]{task.percentage:>3.1f}%",
        "•",
        DownloadColumn(),
        "•",
        TransferSpeedColumn(),
        "•",
        TimeElapsedColumn(),
        transient=True,
    )

    # Group them for coordinated display
    progress_group = Group(
        download_progress,
        overall_progress,
    )

    test_group = args.group or "default"
    test_domains = args.domains or TEST_DOMAINS[test_group]

    with Live(progress_group, console=console, refresh_per_second=24, vertical_overflow="visible") as live:
        overall_task = overall_progress.add_task(
            "Testing domains...", total=len(test_domains)
        )

        for domain in test_domains:
            live.console.rule(f"Testing {domain}")
            config = TEST_CONFIG[test_group]
            tester = SpeedTester(
                domain, config, console, download_progress, args.verbose, args.timeout
            )
            result = tester.run_test()
            results.append(result)
            overall_progress.update(overall_task, advance=1)

            if result["status"] == "success":
                live.console.print(
                    f"✔ [{result['domain']}] Result: [bold green]{format_speed(result['speed_Bps'])}[/bold green]"
                )
            else:
                live.console.print(
                    f"✖ [{result['domain']}] Failed: [bold red]{result['error']}[/bold red]"
                )

    display_summary(results, console)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nAborted by user.")
        sys.exit(0)
	# 本 Powershell 脚本用于 DockerHub 镜像的测速
	# 传入域名列表，自动测速并按速度输出排名
	#
	# 使用默认配置
	# .\docker-mirror-speedtest.ps1
	#
	# 自定义域名和测试目标
	# .\docker-mirror-speedtest.ps1 -Domains @("docker.1ms.run", "hub1.nat.tf")

	[CmdletBinding()]
	param (
	[Parameter(ValueFromPipeline = $true)]
	[string[]]$Domains = @(
	# list from status monitor sites
	# https://status.anye.xyz/status/docker
	# https://status.1panel.top/status/docker
	"docker.m.daocloud.io",
	"docker.xuanyuan.me",
	"docker.1ms.run",
	"docker.m.ixdev.cn",
	"hub1.nat.tf",
	"hub2.nat.tf",
	"hub.amingg.com",
	"hub.rat.dev",
	"docker.etcd.fun",
	"docker.cattt.net"
	)
	)

	# 设置输出编码为UTF8，以支持特殊字符
	$OutputEncoding = [Console]::OutputEncoding = [System.Text.Encoding]::UTF8

	$TestTarget = @{
	Path = "library/nginx" # size: 43,913,652
	Digest = "sha256:8c18fec49d4bdf0813737297228f0afbdc26ed7f8f3fd7e66048114d47d03168"
	}

	# 定义颜色输出函数
	function Write-ColorOutput {
	param ([string]$Text, [string]$Color = "White")
	Write-Host $Text -ForegroundColor $Color
	}

	function Parse-WwwAuthenticate {
	param (
	[string]$AuthStr,
	[hashtable]$Target
	)

	$splits = $AuthStr -split '\s+'
	$bearer = $splits[0]
	if (!$bearer -or $bearer.ToLower() -ne 'bearer') {
	throw "Invalid Www-Authenticate: $AuthStr"
	}

	$params = $splits[1..($splits.Length - 1)] -join '' -split ','
	$result = @{}
	foreach ($param in $params) {
	$kvPair = $param -split '=', 2
	if ($kvPair.Length -eq 2) {
	$k = $kvPair[0].Trim()
	$v = $kvPair[1].Trim() -replace '^"\|"$', ''
	$result[$k] = $v
	}
	}
	$result["scope"] = "respository:$($Target.Path):pull"
	return $result
	}

	function Get-DockerToken {
	param (
	[string]$Realm,
	[string]$Service,
	[string]$Scope
	)

	$tokenUrl = [System.UriBuilder]$Realm
	$query = [System.Web.HttpUtility]::ParseQueryString('')
	$query['service'] = $Service
	$query['scope'] = $Scope
	$tokenUrl.Query = $query.ToString()

	$response = Invoke-RestMethod -Uri $tokenUrl.Uri.ToString() -Method Get
	return $response.token
	}

	# 添加结果收集类
	class SpeedTestResult {
	[string]$Domain
	[double]$Speed
	[double]$ConnectTime
	[bool]$Success
	[string]$Error

	[string] GetSpeedString() {
	if (-not $this.Success) { return "N/A" }
	$nums = $this.Speed
	$unit = "B/s"
	if ($nums -gt 1024 * 1024) {
	$nums = [double]([math]::Round($nums / (1024 * 1024), 2))
	$unit = "MB/s"
	} elseif ($nums -gt 1024) {
	$nums = [double]([math]::Round($nums / 1024, 2))
	$unit = "KB/s"
	}
	return "$nums $unit"
	}
	}

	$global:Results = [System.Collections.ArrayList]::new()

	function Invoke-CurlCommand {
	param (
	[Parameter(Mandatory, Position = 0)]
	[string[]]$Args,
	[Parameter(Mandatory, Position = 1)]
	[string]$Url
	)

	# 构建完整参数数组
	$fullArgs = $Args + @(
	# as some site like 1panel.live blocks curl user-agent
	"--user-agent", "docker-mirror-speedtest.ps1",
	"$Url"
	)

	# 调试输出完整命令
	Write-Verbose "即将执行: curl $($fullArgs -join ' ')"
	return & curl $fullArgs
	}

	function Test-RegistrySpeed {
	param (
	[string]$Domain,
	[hashtable]$Target
	)

	$dsize = 41943040 # 40MB
	$skill = 102400 # 100KB/s

	$result = [SpeedTestResult]::new()
	$result.Domain = $Domain

	try {
	Write-Verbose "开始测试域名: $Domain"
	$probAuthUrl = "https://$Domain/v2/library/alpine/manifests/latest"
	$downloadUrl = "https://$Domain/v2/$($Target.Path)/blobs/$($Target.Digest)"
	Write-Verbose "测试URL: $downloadUrl"

	# 初始请求
	$args = @( "-I", "-L", "--max-time", "8" )
	$response = Invoke-CurlCommand $args $probAuthUrl
	Write-Verbose "收到响应:`n$($response \| Out-String)"

	# 检查状态码
	$statusLine = $response \| Where-Object { $_ -match "^HTTP/.+ (\d{3}) " } \| Select-Object -First 1
	if ($statusLine -match "200") {
	# 直接可访问，无需认证
	Write-ColorOutput "🚩 [$Domain] 无需认证，直接测速..."
	$redirectUrl = $downloadUrl
	} else {
	# 需要认证，解析WWW-Authenticate头
	$authHeader = $response \| Where-Object { $_ -match "Www-Authenticate: " } \| Select-Object -First 1
	if ($authHeader) {
	Write-ColorOutput "🔑 [$Domain] 需要认证，正在获取token..."
	$authStr = ($authHeader -split ": ")[1]
	Write-Verbose "认证字符串: $authStr"

	$auth = Parse-WwwAuthenticate $authStr -Target $Target
	Write-Verbose "解析后的认证信息: $($auth \| ConvertTo-Json)"

	$token = Get-DockerToken -Realm $auth.realm -Service $auth.service -Scope "repository:$($Target.Path)`:pull"
	Write-Verbose "获取到token: $($token.Substring(0, [Math]::Min(20, $token.Length)))..."

	# 使用token获取重定向URL
	Write-ColorOutput "📍 [$Domain] 获取下载地址..."
	$args = @(
	"-s", "-I",
	"-H", "Authorization: Bearer $token",
	"-w", "%{redirect_url}",
	"-o", "nul")
	$redirectUrl = Invoke-CurlCommand $args $downloadUrl
	Write-Verbose "重定向URL: $redirectUrl"
	if ($null -eq $redirectUrl) {
	throw "资源重定向失败"
	}
	} else {
	$statusLine = $response \| Where-Object { $_ -match "^HTTP/.+ (\d{3}.+)" } \| ForEach-Object { $Matches.1 }
	if ($statusLine) {
	throw "无法获取认证信息 ($statusLine)"
	}
	$statusLine = $response \| Where-Object { $_ -match "(curl:.+)" } \| ForEach-Object { $Matches.1 }
	if ($statusLine) {
	throw "其他异常 ($statusLine)"
	}
	throw "未知错误: $($response \| Out-String)"
	}
	}

	# 进行速度测试
	Write-ColorOutput "🏃 [$Domain] 正在测速..." "Blue"

	$curl_args = @(
	"-L",
	"-w", "DNS解析时间: %{time_namelookup}s 连接时间: %{time_connect}s 总时间: %{time_total}s 下载速度: %{speed_download}B/s ",
	"-o", "nul",
	"--range", "0-$dsize",
	"--max-time", "15",
	"--speed-time", "3",
	"--speed-limit", "$skill"
	)
	if ($token) {
	$curl_args += @("-H", "Authorization: Bearer $token")
	}

	$resultOutput = Invoke-CurlCommand $curl_args $redirectUrl
	Write-ColorOutput "📊 [$Domain] 测速结果: $resultOutput"

	$resultOutput -match '连接时间: (?<ctime>\d+\.?\d)s.+下载速度: (?<speed>\d+\.?\d)B/s' \| Out-Null

	# 解析测速结果
	$result.Speed = $Matches.speed
	$result.ConnectTime = $Matches.ctime
	$result.Success = $true
	} catch {
	$result.Error = $_.Exception.Message
	$result.Success = $false
	Write-ColorOutput "❌ [$Domain] 测速异常: $($result.Error)" "Red"
	}

	[void]$global:Results.Add($result)
	return $result
	}

	function Show-TestSummary {
	$successful = $global:Results \| Where-Object { $_.Success }
	$failed = $global:Results \| Where-Object { -not $_.Success }

	Write-Host "`n`n═══════════════ 测试总结 ═══════════════" -ForegroundColor Magenta

	# 统计信息
	Write-Host "✓ 成功测试: $($successful.Count)" -ForegroundColor Green
	Write-Host "✗ 失败测试: $($failed.Count)" -ForegroundColor Red
	Write-Host "Σ 总计测试: $($global:Results.Count)" -ForegroundColor Cyan

	# 显示排名
	if ($successful.Count -gt 0) {
	Write-Host "`n🏆 速度排名:" -ForegroundColor Yellow
	$rank = 1
	$successful \| Sort-Object Speed -Descending \| ForEach-Object {
	$color = switch ($rank) {
	1 { "Yellow" }
	2 { "White" }
	3 { "DarkYellow" }
	default { "Gray" }
	}

	$medal = switch ($rank) {
	1 { "🥇" }
	2 { "🥈" }
	3 { "🥉" }
	default { " " }
	}

	Write-Host ("{0} [{1,2}] {2,-30} {3,10}" -f $medal,
	$rank,
	"$($_.Domain)",
	$_.GetSpeedString()) -ForegroundColor $color
	$rank++
	}
	}

	# 失败清单
	if ($failed.Count -gt 0) {
	Write-Host "`n❌ 失败详情:" -ForegroundColor Red
	$failed \| ForEach-Object {
	Write-Host " • $($_.Domain)" -ForegroundColor DarkRed
	Write-Host " $($_.Error)" -ForegroundColor DarkGray
	}
	}
	}

	# 主程序入口
	Write-ColorOutput "🚀 开始 Docker 镜像站测速..." "Cyan"
	Write-Verbose "测试域名列表: $($Domains -join ', ')"

	try {
	foreach ($domain in $Domains) {
	Test-RegistrySpeed -Domain $domain -Target $TestTarget \| Out-Null
	}
	} finally {
	# 完成后显示总结
	Show-TestSummary
	}
	#!/bin/bash
	#
	# 本 Bash 脚本用于 DockerHub 镜像的测速
	# 传入域名列表，自动测速并按速度输出排名
	#
	# 使用方法:
	# ./docker-mirror-speedtest.sh [选项] [域名1 域名2 ...]
	#
	# 选项:
	# -v, --verbose 启用详细输出模式
	#
	# 示例:
	# 使用默认配置:
	# ./docker-mirror-speedtest.sh
	#
	# 启用详细输出并测试指定域名:
	# ./docker-mirror-speedtest.sh --verbose docker.1ms.run hub1.nat.tf
	#

	# --- 配置 ---

	# 默认域名列表
	# list from status monitor sites
	# https://status.anye.xyz/status/docker
	# https://status.1panel.top/status/docker
	DEFAULT_DOMAINS=(
	"docker.m.daocloud.io"
	"docker.xuanyuan.me"
	"docker.1ms.run"
	"docker.m.ixdev.cn"
	"hub1.nat.tf"
	"hub2.nat.tf"
	"hub.amingg.com"
	"hub.rat.dev"
	"docker.etcd.fun"
	"docker.cattt.net"
	)

	# 测试目标
	TEST_TARGET_PATH="library/nginx"
	TEST_TARGET_DIGEST="sha256:8c18fec49d4bdf0813737297228f0afbdc26ed7f8f3fd7e66048114d47d03168"

	# --- 全局变量 ---
	DOMAINS=()
	VERBOSE=false
	SUCCESSFUL_RESULTS=()
	FAILED_RESULTS=()

	# 颜色定义
	COLOR_RED='\033[0;31m'
	COLOR_GREEN='\033[0;32m'
	COLOR_YELLOW='\033[0;33m'
	COLOR_BLUE='\033[0;34m'
	COLOR_MAGENTA='\033[0;35m'
	COLOR_CYAN='\033[0;36m'
	COLOR_WHITE='\033[0;97m'
	COLOR_GRAY='\033[0;90m'
	COLOR_NC='\033[0m' # No Color

	# --- 函数定义 ---

	# 打印带颜色的输出
	# 参数1: 文本
	# 参数2: 颜色 (Red, Green, Yellow, Blue, Magenta, Cyan, White, Gray)
	write_color_output() {
	local text="$1"
	local color_name="$2"
	local color_code="${COLOR_NC}"
	case "$color_name" in
	"Red") color_code="${COLOR_RED}" ;;
	"Green") color_code="${COLOR_GREEN}" ;;
	"Yellow") color_code="${COLOR_YELLOW}" ;;
	"Blue") color_code="${COLOR_BLUE}" ;;
	"Magenta") color_code="${COLOR_MAGENTA}" ;;
	"Cyan") color_code="${COLOR_CYAN}" ;;
	"White") color_code="${COLOR_WHITE}" ;;
	"Gray") color_code="${COLOR_GRAY}" ;;
	esac
	printf "%b%s%b\n" "${color_code}" "${text}" "${COLOR_NC}"
	}

	# 详细日志输出函数
	# 仅在 VERBOSE 模式下打印
	log_verbose() {
	if [ "$VERBOSE" = true ]; then
	write_color_output "DEBUG: $*" "Gray"
	fi
	}

	# 解析命令行参数
	parse_args() {
	local custom_domains=()
	for arg in "$@"; do
	case "$arg" in
	-v\|--verbose)
	VERBOSE=true
	shift
	;;
	*)
	custom_domains+=("$arg")
	shift
	;;
	esac
	done

	if [ ${#custom_domains[@]} -gt 0 ]; then
	DOMAINS=("${custom_domains[@]}")
	else
	DOMAINS=("${DEFAULT_DOMAINS[@]}")
	fi
	}

	# 格式化速度
	# 参数1: 速度 (B/s)
	format_speed() {
	local speed_bps
	speed_bps=$(printf "%.0f" "$1")
	if (( speed_bps > 1024 * 1024 )); then
	printf "%.2f MB/s" "$(bc <<< "scale=2; $speed_bps / (1024*1024)")"
	elif (( speed_bps > 1024 )); then
	printf "%.2f KB/s" "$(bc <<< "scale=2; $speed_bps / 1024")"
	else
	printf "%d B/s" "$speed_bps"
	fi
	}

	# 主测速函数
	# 参数1: 域名
	test_registry_speed() {
	local domain="$1"
	local dsize=41943040 # 40MB
	local skill=102400 # 100KB/s
	local token=""
	local redirect_url=""
	local response
	local http_status
	local auth_header
	local error_message

	local prob_auth_url="https://$domain/v2/library/alpine/manifests/latest"
	local download_url="https://$domain/v2/$TEST_TARGET_PATH/blobs/$TEST_TARGET_DIGEST"
	log_verbose "[$domain] 探测URL: $prob_auth_url"

	# 初始请求获取认证信息
	response=$(curl --user-agent "docker-mirror-speedtest.sh" -I -L --max-time 8 "$prob_auth_url")
	escaped_verbose_response=$(printf "%s\n%s" "[$domain] 探测响应:" "$response")
	log_verbose "$escaped_verbose_response"

	http_status=$(echo "$response" \| grep -oP '^HTTP/[\d\.]+\s+\K\d{3}' \| tail -n 1)

	if [[ "$http_status" == "200" ]]; then
	write_color_output "🚩 [$domain] 无需认证，直接测速..."
	redirect_url="$download_url"
	else
	auth_header=$(echo "$response" \| grep -i "Www-Authenticate")

	if [ -n "$auth_header" ]; then
	write_color_output "🔑 [$domain] 需要认证，正在获取token..."
	log_verbose "[$domain] 认证头: $auth_header"

	local realm service scope
	realm=$(echo "$auth_header" \| grep -oP 'realm="([^"]+)"' \| cut -d'"' -f2)
	service=$(echo "$auth_header" \| grep -oP 'service="([^"]+)"' \| cut -d'"' -f2)
	scope="repository:$TEST_TARGET_PATH:pull"

	if [ -z "$realm" ]; then
	error_message="无法解析认证 Realm"
	FAILED_RESULTS+=("$domain\|$error_message")
	write_color_output "❌ [$domain] 测速异常: $error_message" "Red"
	return
	fi

	log_verbose "[$domain] Realm='$realm', Service='$service', Scope='$scope'"

	local token_url="$realm?service=$service&scope=$scope"
	local token_response
	token_response=$(curl -sL "$token_url")
	token=$(echo "$token_response" \| grep -oP '"token":"\K[^" ]+')

	if [ -z "$token" ]; then
	error_message="获取 Token 失败"
	FAILED_RESULTS+=("$domain\|$error_message")
	write_color_output "❌ [$domain] 测速异常: $error_message" "Red"
	log_verbose "[$domain] Token响应: $token_response"
	return
	fi
	log_verbose "[$domain] 成功获取Token."

	write_color_output "📍 [$domain] 获取下载地址..."
	redirect_url=$(curl -s -I -H "Authorization: Bearer $token" -w "%{redirect_url}" -o /dev/null "$download_url")

	if [ -z "$redirect_url" ]; then
	error_message="资源重定向失败"
	FAILED_RESULTS+=("$domain\|$error_message")
	write_color_output "❌ [$domain] 测速异常: $error_message" "Red"
	return
	fi
	log_verbose "[$domain] 重定向URL: $redirect_url"
	else
	error_message=$(echo "$response" \| grep -oP '^HTTP/[\d\.]+\s+\K.+\b')
	if [ -z "$error_message" ]; then
	error_message=$(echo "$response" \| grep -oP 'curl: \(\d+\) \K.+')
	fi
	if [ -z "$error_message" ]; then
	error_message="未知错误"
	fi
	FAILED_RESULTS+=("$domain\|无法获取认证信息 ($error_message)")
	write_color_output "❌ [$domain] 测速异常: 无法获取认证信息 ($error_message)" "Red"
	return
	fi
	fi

	write_color_output "🏃 [$domain] 正在测速..." "Blue"

	local curl_args=(
	"-L"
	"-w" "DNS解析时间: %{time_namelookup}s 连接时间: %{time_connect}s 总时间: %{time_total}s 下载速度: %{speed_download}B/s"
	"-o" "/dev/null"
	"--range" "0-$dsize"
	"--max-time" "15"
	"--speed-time" "3"
	"--speed-limit" "$skill"
	)
	if [ -n "$token" ]; then
	curl_args+=("-H" "Authorization: Bearer $token")
	fi

	log_verbose "[$domain] 执行测速命令: curl ${curl_args[*]} $redirect_url"
	local result_output
	result_output=$(curl "${curl_args[@]}" "$redirect_url")

	write_color_output "📊 [$domain] 测速结果: $result_output"

	local speed
	speed=$(echo "$result_output" \| grep -oP '下载速度: \K[\d\.]+')

	if [ -n "$speed" ] && [ "$(printf "%.0f" "$speed")" -gt 0 ]; then
	SUCCESSFUL_RESULTS+=("$domain\|$speed")
	else
	FAILED_RESULTS+=("$domain\|速度测试失败或速度为0")
	write_color_output "❌ [$domain] 测速异常: 速度测试失败或速度为0" "Red"
	fi
	}

	# 显示总结
	show_test_summary() {
	local successful_count=${#SUCCESSFUL_RESULTS[@]}
	local failed_count=${#FAILED_RESULTS[@]}
	local total_count=$((successful_count + failed_count))

	printf "\n"
	write_color_output "═══════════════ 测试总结 ═══════════════" "Magenta"

	write_color_output "✓ 成功测试: $successful_count" "Green"
	write_color_output "✗ 失败测试: $failed_count" "Red"
	write_color_output "Σ 总计测试: $total_count" "Cyan"

	if [ "$successful_count" -gt 0 ]; then
	printf "\n"
	write_color_output "🏆 速度排名:" "Yellow"

	# 排序并显示
	# 使用 printf 将每个元素逐行输出，再用 mapfile/readarray 安全读取到数组，避免命令替换拆分问题
	local sorted_results=()
	mapfile -t sorted_results < <(printf "%s\n" "${SUCCESSFUL_RESULTS[@]}" \| sort -t'\|' -k2 -nr)

	local rank=1
	for result in "${sorted_results[@]}"; do
	local domain speed speed_str color medal
	domain=$(echo "$result" \| cut -d'\|' -f1)
	speed=$(echo "$result" \| cut -d'\|' -f2)
	speed_str=$(format_speed "$speed")

	case "$rank" in
	1) color="${COLOR_YELLOW}" ; medal="🥇" ;;
	2) color="${COLOR_WHITE}" ; medal="🥈" ;;
	3) color="${COLOR_GRAY}" ; medal="🥉" ;;
	*) color="${COLOR_NC}" ; medal=" " ;;
	esac

	printf "%b%s [%2d] %-30s %12s%b\n" "$color" "$medal" "$rank" "$domain" "$speed_str" "${COLOR_NC}"
	((rank++))
	done
	fi

	if [ "$failed_count" -gt 0 ]; then
	printf "\n"
	write_color_output "❌ 失败详情:" "Red"
	for result in "${FAILED_RESULTS[@]}"; do
	local domain error
	domain=$(echo "$result" \| cut -d'\|' -f1)
	error=$(echo "$result" \| cut -d'\|' -f2)
	write_color_output " • $domain" "Red"
	write_color_output " $error" "Gray"
	done
	fi
	}

	# --- 主程序 ---
	main() {
	parse_args "$@"

	log_verbose "详细模式已启用."
	log_verbose "待测试域名: ${DOMAINS[*]}"

	write_color_output "🚀 开始 Docker 镜像站测速..." "Cyan"
	for domain in "${DOMAINS[@]}"; do
	test_registry_speed "$domain"
	done

	show_test_summary
	}

	main "$@"
	#!/usr/bin/env python3
	# -- coding: utf-8 --

	import argparse
	import re
	import sys
	import time
	from dataclasses import dataclass
	from typing import Dict, List, Optional, Tuple

	import requests
	from rich.console import Console, Group
	from rich.live import Live
	from rich.panel import Panel
	from rich.progress import (
	BarColumn,
	DownloadColumn,
	Progress,
	TextColumn,
	TimeElapsedColumn,
	TransferSpeedColumn,
	)
	from rich.table import Table

	# --- Configuration ---


	@dataclass
	class TestTargetConfig:
	"""Configuration for a specific test target."""

	probe_image: str
	test_image: str
	test_digest: str


	TEST_DOMAINS: Dict[str, List[str]] = {
	"default": [
	"docker.m.daocloud.io",
	"docker.xuanyuan.me",
	"docker.1ms.run",
	"docker.m.ixdev.cn",
	"hub1.nat.tf",
	"hub2.nat.tf",
	"hub.amingg.com",
	"hub.rat.dev",
	"docker.etcd.fun",
	"docker.cattt.net",
	],
	"ghcr": [
	"ghcr.nju.edu.cn",
	],
	}

	# Configuration for test targets. This allows specifying different images
	# for different domains if needed.
	# The 'default' key is used for any domain not explicitly listed.
	TEST_CONFIG: Dict[str, TestTargetConfig] = {
	"default": TestTargetConfig(
	probe_image="library/alpine:latest",
	test_image="library/nginx",
	test_digest="sha256:8c18fec49d4bdf0813737297228f0afbdc26ed7f8f3fd7e66048114d47d03168",
	),
	"ghcr": TestTargetConfig(
	probe_image="hasscc/hacn:latest",
	test_image="hasscc/hacn",
	test_digest="sha256:e02a65517b4cc86d5c652f5b6601d5c51d446c55fd63dbf87b35448dd910bc6c", # 59,716,988 B
	),
	}


	# --- Main Speed Test Logic ---


	class SpeedTester:
	"""
	Encapsulates the logic for testing the speed of a single Docker registry mirror.
	"""

	def __init__(
	self,
	domain: str,
	config: TestTargetConfig,
	console: Console,
	progress: Progress,
	verbose: bool = False,
	timeout: int = 10,
	):
	self.domain = domain
	self.config = config
	self.console = console
	self.progress = progress
	self.verbose = verbose
	self.timeout = timeout
	self.auth_token: Optional[str] = None
	self.session = requests.Session()
	self.session.headers.update(
	{"User-Agent": "Docker-Mirror-Speedtest-Python/1.0"}
	)

	def _log_verbose(self, message: str):
	"""Prints a message if verbose mode is enabled."""
	if self.verbose:
	self.console.print(f"[dim]DEBUG: {message}[/dim]")

	def _get_auth_token(self) -> Tuple[Optional[str], Optional[str]]:
	"""
	Probes the registry to see if authentication is needed and, if so, fetches a bearer token.
	Returns a tuple of (token, error_message).
	"""
	probe_image = self.config.probe_image
	[pimage, ptag] = probe_image.split(":")
	probe_url = f"https://{self.domain}/v2/{pimage}/manifests/{ptag}"
	self._log_verbose(f"Probing for auth at {probe_url}")

	try:
	response = self.session.head(probe_url, allow_redirects=True, timeout=10)
	self._log_verbose(f"Probe response status: {response.status_code}")
	self._log_verbose(f"Probe response headers: {response.headers}")

	if response.status_code == 200:
	self.console.print(f"✔ [{self.domain}] No authentication required.")
	return None, None

	if (
	response.status_code in (401, 403)
	and "Www-Authenticate" in response.headers
	):
	self.console.print(
	f"🔑 [{self.domain}] Authentication required, fetching token..."
	)
	auth_header = response.headers["Www-Authenticate"]
	self._log_verbose(f"Auth header: {auth_header}")

	realm_match = re.search('realm="([^"]+)"', auth_header)
	service_match = re.search('service="([^"]+)"', auth_header)

	if not realm_match:
	return None, "Could not parse realm from Www-Authenticate header."

	realm = realm_match.group(1)
	service = service_match.group(1) if service_match else ""
	scope = f"repository:{self.config.test_image}:pull"

	auth_params = {"service": service, "scope": scope}

	self._log_verbose(
	f"Requesting token from {realm} with params {auth_params}"
	)
	token_response = self.session.get(realm, params=auth_params, timeout=10)
	token_response.raise_for_status()

	token = token_response.json().get("token")
	if not token:
	return None, "Failed to retrieve token from auth server."

	self._log_verbose("Successfully obtained auth token.")
	return token, None

	return None, f"Initial probe failed with status {response.status_code}."

	except requests.exceptions.RequestException as e:
	self._log_verbose(f"Auth request exception: {e}")
	return None, f"Cannot connect to registry: {e.__class__.__name__}"

	def run_test(self) -> Dict:
	"""
	Runs the full speed test for the domain.
	Returns a dictionary with the results.
	"""
	try:
	token, error = self._get_auth_token()
	if error:
	return {"domain": self.domain, "status": "failed", "error": error}
	if token:
	self.auth_token = token
	self.session.headers.update(
	{"Authorization": f"Bearer {self.auth_token}"}
	)
	except Exception as e:
	return {
	"domain": self.domain,
	"status": "failed",
	"error": f"Authentication error: {e}",
	}

	test_image = self.config.test_image
	test_digest = self.config.test_digest
	download_url = f"https://{self.domain}/v2/{test_image}/blobs/{test_digest}"
	self._log_verbose(f"Starting download from {download_url}")

	task_id = None
	timed_out = False
	try:
	with self.session.get(download_url, stream=True, timeout=self.timeout + 5) as r:
	self._log_verbose(f"Download response headers: {r.headers}")
	r.raise_for_status()
	total_size = int(r.headers.get("content-length", 0))

	task_id = self.progress.add_task(
	f"⬇️ [bold]{self.domain}[/]",
	total=total_size,
	start=True,
	)

	download_start_time = time.time()
	bytes_downloaded = 0

	for chunk in r.iter_content(chunk_size=8192):
	self.progress.update(task_id, advance=len(chunk))
	bytes_downloaded += len(chunk)

	elapsed_time = time.time() - download_start_time
	if elapsed_time > self.timeout:
	timed_out = True
	break

	download_time = time.time() - download_start_time

	if download_time > 0:
	speed_Bps = bytes_downloaded / download_time
	else:
	speed_Bps = 0

	return {
	"domain": self.domain,
	"status": "success",
	"speed_Bps": speed_Bps,
	"size_bytes": bytes_downloaded,
	"duration_sec": download_time,
	"timed_out": timed_out,
	}

	except requests.exceptions.RequestException as e:
	self._log_verbose(f"Download exception: {e}")
	error_msg = f"Download failed: {e.__class__.__name__}"
	if e.response is not None:
	error_msg += f" (Status: {e.response.status_code})"
	return {"domain": self.domain, "status": "failed", "error": error_msg}
	except Exception as e:
	self._log_verbose(f"An unexpected error occurred: {e}")
	return {
	"domain": self.domain,
	"status": "failed",
	"error": f"An unexpected error occurred: {e}",
	}
	finally:
	if task_id is not None:
	self.progress.remove_task(task_id)


	# --- UI and Formatting ---


	def format_speed(speed_Bps: float) -> str:
	"""Formats speed in B/s to a human-readable string (MB/s, KB/s, B/s)."""
	if speed_Bps > 1024 * 1024:
	return f"{speed_Bps / (1024 * 1024):.2f} MB/s"
	if speed_Bps > 1024:
	return f"{speed_Bps / 1024:.2f} KB/s"
	return f"{speed_Bps:.2f} B/s"


	def display_summary(results: List[Dict], console: Console):
	"""Displays a final summary of all test results."""
	successful = sorted(
	[r for r in results if r["status"] == "success"],
	key=lambda r: r["speed_Bps"],
	reverse=True,
	)
	failed = [r for r in results if r["status"] == "failed"]

	console.print("\n" + "=" * 20 + " Test Summary " + "=" * 20, style="bold magenta")

	stats_table = Table.grid(padding=(0, 2))
	stats_table.add_column()
	stats_table.add_column()
	stats_table.add_row(
	"[green]■[/green] [bold green]Successful tests:[/]", str(len(successful))
	)
	stats_table.add_row("[red]■[/red] [bold red]Failed tests:[/]", str(len(failed)))
	stats_table.add_row("[cyan]■[/cyan] [bold cyan]Total tests:[/]", str(len(results)))
	console.print(Panel(stats_table, title="Statistics", border_style="cyan"))

	if successful:
	table = Table(
	title="Speed Rankings", title_style="bold yellow", border_style="green"
	)
	table.add_column("Rank", style="cyan", justify="center")
	table.add_column("Domain", style="bold white")
	table.add_column("Speed", style="green", justify="right")
	table.add_column("Duration (s)", style="dim", justify="right")
	table.add_column("Size", style="dim", justify="right")

	for i, res in enumerate(successful, 1):
	size_mb = f"{res['size_bytes'] / (1024 * 1024):.2f} MB"
	duration_str = f"{res['duration_sec']:.2f}"
	if res.get("timed_out", False):
	duration_str += " (T)"

	table.add_row(
	str(i),
	res["domain"],
	format_speed(res["speed_Bps"]),
	duration_str,
	size_mb,
	)
	console.print(table)

	if failed:
	fail_table = Table(
	title="Failure Details", title_style="bold red", border_style="red"
	)
	fail_table.add_column("Domain", style="bold white")
	fail_table.add_column("Error", style="yellow")
	for res in failed:
	fail_table.add_row(res["domain"], res["error"])
	console.print(fail_table)


	# --- Main Execution ---


	def main():
	"""Main function to parse arguments and run the speed tests."""
	parser = argparse.ArgumentParser(
	description="Test the speed of Docker registry mirrors.",
	formatter_class=argparse.RawTextHelpFormatter,
	)
	parser.add_argument(
	"-g",
	"--group",
	nargs="?",
	choices=TEST_DOMAINS.keys(),
	help="Predefined group of domains to test (overrides individual domains).",
	)
	parser.add_argument(
	"-t",
	"--timeout",
	type=int,
	default=10,
	help="Maximum download time in seconds for each test. Default is 10s.",
	)
	parser.add_argument(
	"-v",
	"--verbose",
	action="store_true",
	help="Enable verbose output for debugging.",
	)
	parser.add_argument(
	"domains",
	nargs="*",
	help="One or more domain names to test. If not provided, a default list is used.",
	)
	args = parser.parse_args()

	console = Console()
	console.print("🚀 [bold cyan]Starting Docker mirror speed test...[/bold cyan]")

	results = []

	# Define two separate progress bars
	overall_progress = Progress(
	TextColumn("[bold blue]Overall Progress"),
	BarColumn(),
	"[progress.percentage]{task.percentage:>3.0f}%",
	TextColumn("({task.completed} of {task.total})"),
	TimeElapsedColumn(),
	)
	download_progress = Progress(
	TextColumn("[progress.description]{task.description}", justify="left"),
	BarColumn(bar_width=None),
	"[progress.percentage]{task.percentage:>3.1f}%",
	"•",
	DownloadColumn(),
	"•",
	TransferSpeedColumn(),
	"•",
	TimeElapsedColumn(),
	transient=True,
	)

	# Group them for coordinated display
	progress_group = Group(
	download_progress,
	overall_progress,
	)

	test_group = args.group or "default"
	test_domains = args.domains or TEST_DOMAINS[test_group]

	with Live(progress_group, console=console, refresh_per_second=24, vertical_overflow="visible") as live:
	overall_task = overall_progress.add_task(
	"Testing domains...", total=len(test_domains)
	)

	for domain in test_domains:
	live.console.rule(f"Testing {domain}")
	config = TEST_CONFIG[test_group]
	tester = SpeedTester(
	domain, config, console, download_progress, args.verbose, args.timeout
	)
	result = tester.run_test()
	results.append(result)
	overall_progress.update(overall_task, advance=1)

	if result["status"] == "success":
	live.console.print(
	f"✔ [{result['domain']}] Result: [bold green]{format_speed(result['speed_Bps'])}[/bold green]"
	)
	else:
	live.console.print(
	f"✖ [{result['domain']}] Failed: [bold red]{result['error']}[/bold red]"
	)

	display_summary(results, console)


	if __name__ == "__main__":
	try:
	main()
	except KeyboardInterrupt:
	print("\nAborted by user.")
	sys.exit(0)