agent_hal/skills/git.py

"""
Skill GIT — opérations git sur des dépôts locaux.

Usage LLM :
  SKILL:git ARGS:status [chemin]
  SKILL:git ARGS:log [chemin] [n]
  SKILL:git ARGS:diff [chemin]
  SKILL:git ARGS:add <chemin_repo> | <fichiers>
  SKILL:git ARGS:commit <chemin_repo> | <message>
  SKILL:git ARGS:push [chemin] [remote] [branche]
  SKILL:git ARGS:pull [chemin] [remote] [branche]
  SKILL:git ARGS:clone <url> [destination]
  SKILL:git ARGS:branch [chemin]
  SKILL:git ARGS:checkout <branche> [chemin]
  SKILL:git ARGS:init <chemin>
  SKILL:git ARGS:stash [chemin]
  SKILL:git ARGS:tag <nom> [chemin]
"""
import os
import subprocess

DESCRIPTION = "Opérations git : status, log, diff, add, commit, push, pull, clone, branch, checkout, init"
USAGE = (
    "SKILL:git ARGS:status [path] | log [path] [n] | diff [path] | "
    "add <repo>|<files> | commit <repo>|<msg> | push [path] | pull [path] | "
    "clone <url> [dest] | branch [path] | checkout <branch> [path] | init <path>"
)


def _git(cmd: str, cwd: str = None, timeout: int = 60) -> str:
    try:
        result = subprocess.run(
            f"git {cmd}", shell=True, text=True,
            capture_output=True, timeout=timeout,
            cwd=cwd
        )
        out = (result.stdout + result.stderr).strip()
        return out[:4000] if out else f"(code retour : {result.returncode})"
    except subprocess.TimeoutExpired:
        return f"Timeout ({timeout}s)"
    except Exception as e:
        return str(e)


def _find_git_root(path: str) -> str:
    """Remonte jusqu'à trouver un dépôt git."""
    try:
        result = subprocess.run(
            "git rev-parse --show-toplevel", shell=True, text=True,
            capture_output=True, cwd=path
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception:
        pass
    return path


def run(args: str, context) -> str:
    parts = args.strip().split(None, 1)
    action = parts[0].lower() if parts else ""
    rest   = parts[1] if len(parts) > 1 else ""

    if action == "status":
        path = rest or "."
        cwd = _find_git_root(path)
        return _git("status -sb", cwd=cwd)

    if action == "log":
        parts2 = rest.split()
        path = parts2[0] if parts2 else "."
        n    = parts2[1] if len(parts2) > 1 else "10"
        cwd  = _find_git_root(path)
        return _git(f"log --oneline --graph -n {n}", cwd=cwd)

    if action == "diff":
        parts2 = rest.split(None, 1)
        path   = parts2[0] if parts2 else "."
        extra  = parts2[1] if len(parts2) > 1 else ""
        cwd    = _find_git_root(path)
        return _git(f"diff {extra}", cwd=cwd)

    if action == "add":
        if "|" not in rest:
            # Essaie d'interpréter comme "add <path>"
            cwd = _find_git_root(rest or ".")
            return _git("add -A", cwd=cwd)
        repo, files = rest.split("|", 1)
        cwd = _find_git_root(repo.strip())
        return _git(f"add {files.strip()}", cwd=cwd)

    if action == "commit":
        if "|" not in rest:
            return "Format : commit <chemin_repo> | <message>"
        repo, message = rest.split("|", 1)
        cwd = _find_git_root(repo.strip())
        msg = message.strip().replace('"', '\\"')
        return _git(f'commit -m "{msg}"', cwd=cwd)

    if action == "push":
        parts2 = rest.split()
        path   = parts2[0] if parts2 else "."
        remote = parts2[1] if len(parts2) > 1 else "origin"
        branch = parts2[2] if len(parts2) > 2 else ""
        cwd    = _find_git_root(path)
        return _git(f"push {remote} {branch}", cwd=cwd, timeout=120)

    if action == "pull":
        parts2 = rest.split()
        path   = parts2[0] if parts2 else "."
        remote = parts2[1] if len(parts2) > 1 else "origin"
        branch = parts2[2] if len(parts2) > 2 else ""
        cwd    = _find_git_root(path)
        return _git(f"pull {remote} {branch}", cwd=cwd, timeout=120)

    if action == "clone":
        parts2 = rest.split(None, 1)
        if not parts2:
            return "Format : clone <url> [destination]"
        url  = parts2[0]
        dest = parts2[1] if len(parts2) > 1 else ""
        return _git(f"clone {url} {dest}", timeout=180)

    if action == "branch":
        path = rest or "."
        cwd  = _find_git_root(path)
        return _git("branch -a", cwd=cwd)

    if action == "checkout":
        parts2 = rest.split()
        if not parts2:
            return "Format : checkout <branche> [chemin]"
        branch = parts2[0]
        path   = parts2[1] if len(parts2) > 1 else "."
        cwd    = _find_git_root(path)
        return _git(f"checkout {branch}", cwd=cwd)

    if action == "init":
        path = rest or "."
        os.makedirs(path, exist_ok=True)
        return _git("init", cwd=path)

    if action == "stash":
        path = rest or "."
        cwd  = _find_git_root(path)
        return _git("stash", cwd=cwd)

    if action == "tag":
        parts2 = rest.split()
        if not parts2:
            return "Format : tag <nom> [chemin]"
        name = parts2[0]
        path = parts2[1] if len(parts2) > 1 else "."
        cwd  = _find_git_root(path)
        return _git(f"tag {name}", cwd=cwd)

    return (
        "Action inconnue. Disponible : status, log, diff, add, commit, push, pull, "
        "clone, branch, checkout, init, stash, tag"
    )