home-docker/scripts/check_versions.py

#!/usr/bin/env python3
"""Script to query Docker registries for latest versions."""
import subprocess
import json
import re
import sys
import time

# Active images with pinned versions (not commented out, not :latest, not :edge)
IMAGES = [
    ("dockerhub", "library/caddy", "2", "mirror.gcr.io/caddy:2"),
    ("dockerhub", "linuxserver/jackett", "0.24.1124", "mirror.gcr.io/linuxserver/jackett:0.24.1124"),
    ("dockerhub", "linuxserver/qbittorrent", "version-5.1.2-r4", "mirror.gcr.io/linuxserver/qbittorrent:version-5.1.2-r4"),
    ("dockerhub", "n8nio/n8n", "2.15.0", "mirror.gcr.io/n8nio/n8n:2.15.0"),
    ("dockerhub", "library/postgres", "14", "mirror.gcr.io/postgres:14"),
    ("dockerhub", "qdrant/qdrant", "v1.16-unprivileged", "mirror.gcr.io/qdrant/qdrant:v1.16-unprivileged"),
    ("ghcr", "kube-vip/kube-vip", "v0.4.4", "ghcr.io/kube-vip/kube-vip:v0.4.4"),
    ("quay", "oauth2-proxy/oauth2-proxy", "v7.6.0", "quay.io/oauth2-proxy/oauth2-proxy:v7.6.0"),
]

def safe_version_sort_key(tag_str):
    parts = tag_str.lstrip('v').split('.')
    nums = []
    for p in parts:
        try:
            nums.append(int(re.match(r'(\d+)', p).group(1)))
        except (ValueError, AttributeError, TypeError):
            nums.append(0)
    return nums

def fetch_dockerhub_tag_details(image_name, max_items=200):
    try:
        result = subprocess.run(
            ["curl", "-s", "-L", "-f", 
             f"https://hub.docker.com/v2/repositories/{image_name}/tags?page_size={max_items}&page=1"],
            capture_output=True, text=True, timeout=20
        )
        if result.returncode != 0:
            return None, f"curl failed: {result.stderr[:300]}"
        
        data = json.loads(result.stdout)
        tags_detail = []
        for tag_data in data.get("results", []):
            tag_name = tag_data.get("name", "")
            if tag_name and not tag_name.startswith("sha256:"):
                tags_detail.append({
                    "name": tag_name,
                    "last_updated": tag_data.get("last_updated", ""),
                })
        return tags_detail, None
    except Exception as e:
        return None, str(e)

def fetch_ghcr_tags(image_name):
    url = f"https://ghcr.io/v2/{image_name}/tags/list"
    result = subprocess.run(
        ["curl", "-s", "-L", "-f", url],
        capture_output=True, text=True, timeout=20
    )
    if result.returncode != 0:
        return None, f"GHCR API failed: {result.stderr[:300]}"
    
    try:
        data = json.loads(result.stdout)
        tags = data.get("tags", [])
        return [t for t in tags if not t.startswith("sha256:")], None
    except Exception as e:
        return None, str(e)

def fetch_quay_tags(image_name):
    url = f"https://quay.io/api/v1/repository/{image_name}?includeTags=true&limit=100"
    result = subprocess.run(
        ["curl", "-s", "-L", "-f", url],
        capture_output=True, text=True, timeout=20
    )
    if result.returncode != 0:
        return None, f"Quay API failed: {result.stderr[:300]}"
    
    try:
        data = json.loads(result.stdout)
        tags = list(data.get("tags", {}).keys())
        return tags, None
    except Exception as e:
        return None, str(e)

results = []

for registry_type, image_name, current_tag, display_name in IMAGES:
    print(f"\n{'='*70}")
    print(f"CHECKING: {display_name}")
    print(f"{'='*70}")
    
    tags_detail = None
    err = None
    
    time.sleep(0.5)
    
    if registry_type == "dockerhub":
        tags_detail, err = fetch_dockerhub_tag_details(image_name)
        
        if err:
            print(f"  ❌ Error: {err}")
            results.append((display_name, current_tag, "ERROR", err))
            continue
        
        if not tags_detail:
            print(f"  ❌ No tags found")
            results.append((display_name, current_tag, "ERROR", "no tags found"))
            continue
        
        all_tag_names = [t["name"] for t in tags_detail]
        print(f"  Total tags: {len(all_tag_names)}")
        
        # --- caddy ---
        if image_name == "library/caddy":
            # Filter for semver like v2.8.x, v2.9.x etc.
            specific = [t for t in all_tag_names 
                      if re.match(r'^v?\d+\.\d+\.\d+$', t)]
            specific.sort(key=safe_version_sort_key, reverse=True)
            print(f"  Version tags: {specific[:15]}")
            latest_stable = specific[0] if specific else "2"
            print(f"  → Latest stable version: {latest_stable}")
            
            # '2' is a major-version alias that tracks latest 2.x
            results.append((display_name, current_tag, latest_stable, 
                          f"Tag '2' tracks latest 2.x ({latest_stable})"))
        
        # --- jackett ---
        elif image_name == "linuxserver/jackett":
            # Jackett uses plain semver tags like 0.24.1124
            jackett_stable = [t for t in all_tag_names 
                            if re.match(r'^\d+\.\d+\.\d+$', t)]
            jackett_stable.sort(key=safe_version_sort_key, reverse=True)
            print(f"  Stable tags (latest first): {jackett_stable[:10]}")
            latest_stable = jackett_stable[0] if jackett_stable else "unknown"
            print(f"  → Latest stable version: {latest_stable}")
            results.append((display_name, current_tag, latest_stable, ""))
        
        # --- qbittorrent ---
        elif image_name == "linuxserver/qbittorrent":
            qbit_tags = [t for t in all_tag_names if re.match(r'version-', t)]
            print(f"  'version-*' tags: {qbit_tags[:15]}")
            qbit_tags.sort(key=safe_version_sort_key, reverse=True)
            latest_stable = qbit_tags[0] if qbit_tags else "unknown"
            print(f"  → Latest 'version-*' tag: {latest_stable}")
            results.append((display_name, current_tag, latest_stable, ""))
        
        # --- n8n ---
        elif image_name == "n8nio/n8n":
            n8n_tags = [t for t in all_tag_names 
                      if re.match(r'^\d+\.\d+\.\d+$', t)
                      and not re.search(r'(alpha|beta|rc|dev|nightly)', t, re.I)]
            n8n_tags.sort(key=safe_version_sort_key, reverse=True)
            print(f"  Stable tags (latest first): {n8n_tags[:10]}")
            latest_stable = n8n_tags[0] if n8n_tags else "unknown"
            print(f"  → Latest stable version: {latest_stable}")
            results.append((display_name, current_tag, latest_stable, ""))
        
        # --- postgres ---
        elif image_name == "library/postgres":
            major_tags = [t for t in all_tag_names 
                        if re.match(r'^\d+$', t) and int(t) >= 10]
            major_tags.sort(key=int, reverse=True)
            print(f"  Major version tags: {major_tags[:10]}")
            latest_stable = major_tags[0] if major_tags else "unknown"
            print(f"  → Latest major version: {latest_stable}")
            results.append((display_name, current_tag, latest_stable, ""))
        
        # --- qdrant ---
        elif image_name == "qdrant/qdrant":
            # Unprivileged tags
            unpriv = [t for t in all_tag_names if 'unprivileged' in t.lower()]
            unpriv.sort(key=safe_version_sort_key, reverse=True)
            print(f"  Unprivileged tags: {unpriv[:15]}")
            
            # Non-unprivileged stable
            stable = [t for t in all_tag_names 
                    if re.match(r'^v?\d+\.\d+', t)
                    and 'unprivileged' not in t.lower()
                    and not re.search(r'(alpha|beta|rc|dev|nightly)', t, re.I)]
            stable.sort(key=safe_version_sort_key, reverse=True)
            print(f"  Stable version tags: {stable[:10]}")
            
            latest_stable = unpriv[0] if unpriv else "unknown"
            print(f"  → Latest unprivileged: {latest_stable}")
            results.append((display_name, current_tag, latest_stable, ""))
        
        else:
            stable = [t for t in all_tag_names 
                    if re.match(r'^v?\d+\.\d+', t)
                    and not re.search(r'(alpha|beta|rc|dev|nightly)', t, re.I)]
            stable.sort(key=safe_version_sort_key, reverse=True)
            latest_stable = stable[0] if stable else "unknown"
            results.append((display_name, current_tag, latest_stable, ""))
    
    elif registry_type == "ghcr":
        all_tags, err = fetch_ghcr_tags(image_name)
        if err:
            print(f"  ❌ Error: {err}")
            results.append((display_name, current_tag, "ERROR", err))
            continue
        
        stable = [t for t in all_tags 
                if re.match(r'^v?\d+\.\d+', t)
                and not re.search(r'(alpha|beta|rc|dev|nightly)', t, re.I)]
        stable.sort(key=safe_version_sort_key, reverse=True)
        latest_stable = stable[0] if stable else all_tags[0] if all_tags else "unknown"
        print(f"  Tags: {all_tags[:20]}")
        print(f"  → Latest stable version: {latest_stable}")
        results.append((display_name, current_tag, latest_stable, ""))
    
    elif registry_type == "quay":
        all_tags, err = fetch_quay_tags(image_name)
        if err:
            print(f"  ❌ Error: {err}")
            results.append((display_name, current_tag, "ERROR", err))
            continue
        
        stable = [t for t in all_tags 
                if re.match(r'^v?\d+\.\d+', t)
                and not re.search(r'(alpha|beta|rc|dev|nightly)', t, re.I)]
        stable.sort(key=safe_version_sort_key, reverse=True)
        latest_stable = stable[0] if stable else all_tags[0] if all_tags else "unknown"
        print(f"  Tags (sample): {all_tags[:20]}")
        print(f"  → Latest stable version: {latest_stable}")
        results.append((display_name, current_tag, latest_stable, ""))

# Print summary
print("\n\n" + "="*70)
print("SUMMARY")
print("="*70)

for display_name, current_tag, latest_tag, note in results:
    print(f"\n  {display_name}")
    print(f"    Current: {current_tag}")
    
    if latest_tag == "ERROR":
        print(f"    Status: ❌ {note}")
        continue
    
    # Normalize for comparison
    cur = current_tag.lstrip('v').lower()
    lat = latest_tag.lstrip('v').lower()
    
    if cur == lat:
        print(f"    Latest:  {latest_tag}  ✓")
    else:
        print(f"    Latest:  {latest_tag}")
        print(f"    Status:  ⚠ UPDATE AVAILABLE")

print()