scw_registry_prune.py

import requests
import argparse
import os
import logging
import time
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configuration du logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)

SUPPORTED_REGIONS = ["fr-par", "nl-ams", "pl-waw"]
DEFAULT_REGION = "fr-par"


def get_api_base(region):
    return f"https://api.scaleway.com/registry/v1/regions/{region}"


def get_headers(token):
    return {
        "X-Auth-Token": token,
        "Content-Type": "application/json"
    }


def create_session_with_retry(retries=3, backoff_factor=0.5):
    """Crée une session HTTP avec retry automatique."""
    session = requests.Session()
    retry_strategy = Retry(
        total=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "DELETE"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session


def parse_datetime(s):
    """Parse une date ISO 8601 en datetime UTC."""
    return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc)


def format_date(dt):
    """Formate une date au format DD/MM/YYYY HHhMM."""
    if isinstance(dt, str):
        dt = parse_datetime(dt)
    return dt.strftime("%d/%m/%Y %Hh%M")


def group_tags_by_digest(tags):
    """
    Groupe les tags par digest (checksum).
    Retourne une liste de groupes, chaque groupe contenant:
    - digest: le checksum de l'image
    - tags: liste des tags pointant vers ce digest
    - newest_date: date du tag le plus récent du groupe
    """
    digest_groups = defaultdict(list)

    for tag in tags:
        digest = tag.get("digest", tag.get("id"))  # Fallback sur id si pas de digest
        digest_groups[digest].append(tag)

    groups = []
    for digest, group_tags in digest_groups.items():
        # Trouver la date la plus récente parmi les tags du groupe
        newest_date = max(parse_datetime(t["created_at"]) for t in group_tags)
        groups.append({
            "digest": digest,
            "tags": group_tags,
            "newest_date": newest_date
        })

    return groups


def list_namespaces(session, token, api_base):
    resp = session.get(f"{api_base}/namespaces", headers=get_headers(token))
    resp.raise_for_status()
    namespaces = resp.json().get("namespaces", [])
    for ns in namespaces:
        logger.info(f"🧭 Namespace ID: {ns['id']} | Name: {ns['name']} | Region: {ns.get('region')}")
    return namespaces


def list_repositories(session, token, api_base, namespace_id, rate_limit_delay):
    all_repos = []
    page = 1
    while True:
        url = f"{api_base}/images?namespace_id={namespace_id}&page_size=100&page={page}"
        resp = session.get(url, headers=get_headers(token))
        resp.raise_for_status()
        repos = resp.json().get("images", [])
        if not repos:
            break
        all_repos.extend(repos)
        page += 1
        if rate_limit_delay > 0:
            time.sleep(rate_limit_delay)
    return all_repos


def list_images(session, token, api_base, repository_id, rate_limit_delay):
    all_tags = []
    page = 1
    while True:
        url = f"{api_base}/images/{repository_id}/tags?page_size=100&page={page}"
        resp = session.get(url, headers=get_headers(token))
        resp.raise_for_status()
        tags = resp.json().get("tags", [])
        if not tags:
            break
        all_tags.extend(tags)
        page += 1
        if rate_limit_delay > 0:
            time.sleep(rate_limit_delay)
    return all_tags


def delete_image(session, token, api_base, tag_id, rate_limit_delay):
    resp = session.delete(f"{api_base}/tags/{tag_id}", headers=get_headers(token))
    if resp.status_code not in [200, 204]:
        logger.warning(f"⚠️ Failed to delete image {tag_id}: {resp.status_code} {resp.text}")
        return False
    else:
        logger.info(f"🗑️ Deleted image {tag_id}")
        if rate_limit_delay > 0:
            time.sleep(rate_limit_delay)
        return True


def confirm_deletion(count):
    """Demande confirmation avant suppression."""
    while True:
        response = input(f"\nVoulez-vous supprimer {count} image(s) ? [y/N]: ").strip().lower()
        if response in ["y", "yes", "o", "oui"]:
            return True
        elif response in ["n", "no", "non", ""]:
            return False
        else:
            print("Veuillez répondre par 'y' (oui) ou 'n' (non)")


def prune(token, keep, dry_run, min_age_days, region, rate_limit_delay, force):
    api_base = get_api_base(region)
    session = create_session_with_retry()

    logger.info(f"🚀 Démarrage du nettoyage...")

    namespaces = list_namespaces(session, token, api_base)
    cutoff_date = datetime.now(timezone.utc) - timedelta(days=min_age_days)

    # Première passe : collecter toutes les images à supprimer
    images_to_delete = []

    for ns in namespaces:
        logger.info(f"📦 Namespace: {ns['name']}")
        repos = list_repositories(session, token, api_base, ns["id"], rate_limit_delay)

        for repo in repos:
            logger.info(f"  📁 Repo: {repo['name']}")
            tags = list_images(session, token, api_base, repo["id"], rate_limit_delay)

            # Grouper les tags par digest (checksum)
            digest_groups = group_tags_by_digest(tags)
            logger.debug(f"    📊 {len(tags)} tags, {len(digest_groups)} images uniques (par digest)")

            # Trier les groupes par date du tag le plus récent
            sorted_groups = sorted(digest_groups, key=lambda g: g["newest_date"], reverse=True)

            # Garder les N groupes les plus récents
            groups_to_keep = sorted_groups[:keep]
            groups_to_delete = sorted_groups[keep:]

            # Logger les groupes conservés
            for group in groups_to_keep:
                tag_names = ", ".join(t["name"] for t in group["tags"])
                logger.debug(f"✅ [KEEP] digest:{group['digest'][:12]}... tags:[{tag_names}] - {format_date(group['newest_date'])}")

            # Traiter les groupes à supprimer
            for group in groups_to_delete:
                # Vérifier si le groupe est trop récent
                if group["newest_date"] > cutoff_date:
                    tag_names = ", ".join(t["name"] for t in group["tags"])
                    logger.debug(f"⏭️ [SKIP] digest:{group['digest'][:12]}... tags:[{tag_names}] (Too recent)")
                    continue

                # Ajouter tous les tags du groupe à supprimer
                for tag in group["tags"]:
                    images_to_delete.append({
                        "tag": tag,
                        "repo_name": repo["name"],
                        "ns_name": ns["name"],
                        "digest": group["digest"]
                    })

    if not images_to_delete:
        logger.info("✅ Aucune image à supprimer.")
        return

    # Afficher le résumé
    logger.info(f"\n🔍 Images à supprimer: {len(images_to_delete)}")
    for img in images_to_delete:
        tag = img["tag"]
        digest_short = img.get("digest", "")[:12]
        if dry_run:
            logger.info(f"🧪 [DRY-RUN] Would delete: {img['ns_name']}/{img['repo_name']}:{tag['name']} (digest:{digest_short}...) - {format_date(tag['created_at'])}")
        else:
            logger.info(f"🎯 [TO DELETE] {img['ns_name']}/{img['repo_name']}:{tag['name']} (digest:{digest_short}...) - {format_date(tag['created_at'])}")

    if dry_run:
        logger.info("🧪 Mode dry-run: aucune suppression effectuée.")
        return

    # Demander confirmation si pas de --force
    if not force:
        if not confirm_deletion(len(images_to_delete)):
            logger.info("🚫 Suppression annulée par l'utilisateur.")
            return

    # Supprimer les images
    deleted_count = 0
    failed_count = 0

    for img in images_to_delete:
        tag = img["tag"]
        digest_short = img.get("digest", "")[:12]
        logger.info(f"🗑️ [REMOVE] {img['ns_name']}/{img['repo_name']}:{tag['name']} (digest:{digest_short}...) - {format_date(tag['created_at'])}")
        if delete_image(session, token, api_base, tag["id"], rate_limit_delay):
            deleted_count += 1
        else:
            failed_count += 1

    logger.info(f"\n✅ Terminé: {deleted_count} image(s) supprimée(s), {failed_count} échec(s).")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Prune old images in Scaleway Container Registry.")
    parser.add_argument("--token", default=os.getenv("SCW_SECRET_KEY"),
                        help="Scaleway secret token (or use SCW_SECRET_KEY env variable)")
    parser.add_argument("--keep", type=int, default=6,
                        help="Number of most recent unique images (by digest) to keep per repo (minimum: 1)")
    parser.add_argument("--dry-run", action="store_true",
                        help="Simulate deletions without removing images")
    parser.add_argument("--min-age-days", type=int, default=30,
                        help="Minimum age (created_at) in days before an image can be deleted")
    parser.add_argument("--region", choices=SUPPORTED_REGIONS, default=DEFAULT_REGION,
                        help=f"Scaleway region (default: {DEFAULT_REGION})")
    parser.add_argument("--rate-limit-delay", type=float, default=0.1,
                        help="Delay in seconds between API calls (default: 0.1)")
    parser.add_argument("--force", "-f", action="store_true",
                        help="Skip confirmation prompt before deletion")
    parser.add_argument("--verbose", "-v", action="store_true",
                        help="Enable verbose logging (DEBUG level)")
    args = parser.parse_args()

    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    if not args.token:
        logger.error("❌ Scaleway token required (use --token or SCW_SECRET_KEY env variable)")
        exit(1)

    if args.keep < 1:
        logger.error("❌ --keep must be at least 1 to avoid deleting all images")
        exit(1)

    if args.min_age_days < 0:
        logger.error("❌ --min-age-days must be a positive number")
        exit(1)

    # Afficher le récapitulatif des options
    logger.info("=" * 50)
    logger.info("🧼 SCALEWAY REGISTRY PRUNE")
    logger.info("=" * 50)
    logger.info(f"  📍 Region:           {args.region}")
    logger.info(f"  🔢 Keep:             {args.keep} image(s) per repo")
    logger.info(f"  📅 Min age:          {args.min_age_days} days")
    logger.info(f"  ⏱️  Rate limit:       {args.rate_limit_delay}s between calls")
    logger.info(f"  🧪 Dry-run:          {'Yes' if args.dry_run else 'No'}")
    logger.info(f"  🔓 Force:            {'Yes' if args.force else 'No'}")
    logger.info(f"  📝 Verbose:          {'Yes' if args.verbose else 'No'}")
    logger.info("=" * 50)

    prune(args.token, args.keep, args.dry_run, args.min_age_days, args.region, args.rate_limit_delay, args.force)