pulls.py

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

import argparse
import fnmatch
import os
import subprocess
import sys
from pathlib import Path
from typing import Iterable, List

from rich.console import Console
from rich.table import Table

console = Console()


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Scan un dossier et fait `git pull` sur chaque dépôt Git trouvé, avec statut coloré."
    )
    p.add_argument("root", type=Path, help="Dossier racine à parcourir")
    p.add_argument("-r", "--recursive", action="store_true", help="Parcourir récursivement tous les sous-dossiers")
    p.add_argument("--include-root", action="store_true", help="Inclure le dossier racine si c'est un dépôt Git")
    p.add_argument("-e", "--exclude", action="append", default=[], help="Motif glob pour exclure des dossiers")
    p.add_argument("--exclude-file", type=Path, help="Fichier texte avec motifs d'exclusion (un par ligne)")
    p.add_argument("-n", "--dry-run", action="store_true", help="N'exécute pas réellement `git pull`")
    p.add_argument("-v", "--verbose", action="store_true", help="Mode verbeux")
    p.add_argument("--git", default="git", help="Nom/chemin de l'exécutable git (défaut: git)")
    return p.parse_args()


# --- Fonctions utilitaires --------------------------------------------------


def load_excludes(args: argparse.Namespace) -> List[str]:
    patterns = list(args.exclude or [])
    if args.exclude_file and args.exclude_file.exists():
        for line in args.exclude_file.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            patterns.append(line)
    return patterns


def is_excluded(path: Path, root: Path, patterns: Iterable[str]) -> bool:
    rel = path.relative_to(root).as_posix()
    for pat in patterns:
        if fnmatch.fnmatch(path.name, pat) or fnmatch.fnmatch(rel, pat):
            return True
    return False


def iter_dirs(root: Path, recursive: bool) -> Iterable[Path]:
    if recursive:
        for dpath, dirnames, _ in os.walk(root):
            dirnames[:] = [d for d in dirnames if d != ".git"]
            for d in dirnames:
                yield Path(dpath) / d
    else:
        for child in root.iterdir():
            if child.is_dir() and child.name != ".git":
                yield child


def is_git_repo(path: Path) -> bool:
    return (path / ".git").is_dir()


def run_cmd(cmd: list, cwd: Path) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)


def repo_status(repo: Path, git: str) -> str:
    # On met à jour les refs distantes
    update = run_cmd([git, "remote", "update"], cwd=repo)
    if update.returncode != 0:
        return "REMOTE_ERROR"

    status = run_cmd([git, "status", "-uno"], cwd=repo)
    s = status.stdout.lower()
    if "up to date" in s:
        return "UP_TO_DATE"
    if "behind" in s:
        return "BEHIND"
    if "ahead" in s and "behind" not in s:
        return "AHEAD"
    return "UNKNOWN"


def git_pull(repo: Path, git: str) -> bool:
    proc = run_cmd([git, "pull", "--ff-only"], cwd=repo)
    return proc.returncode == 0


# --- Main -------------------------------------------------------------------


def main() -> int:
    args = parse_args()
    root: Path = args.root.expanduser().resolve()

    if not root.exists() or not root.is_dir():
        console.print(f"[red]Erreur:[/red] {root} n'existe pas ou n'est pas un dossier.", style="bold red")
        return 2

    excludes = load_excludes(args)
    if args.verbose and excludes:
        console.print(f"[cyan]Motifs d'exclusion:[/cyan] {', '.join(excludes)}")

    targets: List[Path] = []
    if args.include_root and is_git_repo(root) and not is_excluded(root, root, excludes):
        targets.append(root)

    for d in iter_dirs(root, args.recursive):
        if is_excluded(d, root, excludes):
            if args.verbose:
                console.print(f"[dim]↷ {d} exclu[/dim]")
            continue
        if is_git_repo(d):
            targets.append(d)
        elif args.verbose:
            console.print(f"[dim]↷ {d} pas un dépôt Git[/dim]")

    if not targets:
        console.print("[green]Aucun dépôt Git trouvé. 🎉[/green]")
        return 0

    table = Table(title="Statut des dépôts Git", show_lines=True)
    table.add_column("Dépôt", style="bold")
    table.add_column("Statut", justify="center")
    table.add_column("Action", justify="center")

    failures = 0

    for repo in sorted(targets):
        status = repo_status(repo, args.git)

        if status == "UP_TO_DATE":
            table.add_row(str(repo), "[green]✅ Déjà à jour[/green]", "—")
            continue

        if status == "AHEAD":
            table.add_row(str(repo), "[yellow]⬆️ En avance[/yellow]", "[dim]Push recommandé[/dim]")
            continue

        if status == "BEHIND":
            if args.dry_run:
                table.add_row(str(repo), "[blue]⬇️ Derrière[/blue]", "[cyan]Dry-run[/cyan]")
                continue
            ok = git_pull(repo, args.git)
            if ok:
                table.add_row(str(repo), "[blue]⬇️ Derrière[/blue]", "[green]Pull effectué[/green]")
            else:
                table.add_row(str(repo), "[blue]⬇️ Derrière[/blue]", "[red]Échec pull[/red]")
                failures += 1
            continue

        if status == "REMOTE_ERROR":
            table.add_row(str(repo), "[red]⚠️ Erreur remote[/red]", "[red]Check remote[/red]")
            failures += 1
            continue

        # UNKNOWN
        table.add_row(str(repo), "[yellow]❓ Inconnu[/yellow]", "[dim]Check manuel[/dim]")

    console.print(table)

    if failures:
        console.print(f"[bold red]Terminé avec {failures} échec(s).[/bold red]")
        return 1

    console.print("[bold green]Tout est à jour. 🧼✨[/bold green]")
    return 0


if __name__ == "__main__":
    sys.exit(main())