kdeps-tools-git_ops

version: 2.0.0
description: 'Run one structured git operation and return its JSON result. Operations: status, diff, log, show, branch, remote, add, commit, checkout, init, clone, push, pull. Requires: operation, and url for clone. Optional: workingDir, paths, message, branch, remote, args, maxCount, dryRun.'
parameters:
  args:
    type: string
    description: Extra space-separated git arguments (show)
  branch:
    type: string
    description: Branch for checkout, push, pull
  dryRun:
    type: boolean
    description: 'Report what a write operation would do without doing it. Default: false'
  maxCount:
    type: integer
    description: 'Commits to list (log). Default: 10'
  message:
    type: string
    description: Commit message (commit)
  operation:
    type: string
    description: status, diff, log, show, branch, remote, add, commit, checkout, init, clone, push or pull
    required: true
  paths:
    type: string
    description: 'Space-separated paths for add, checkout, diff, log (add defaults to .); quote a path with spaces'
  remote:
    type: string
    description: 'Remote for push and pull. Default: origin'
  url:
    type: string
    description: Repository URL (clone)
  workingDir:
    type: string
    description: Repository directory (clone - the target directory); defaults to the current directory
python:
  paths: [workingDir]  # checked against the workspace boundary
  timeout: 5m  # clone, push and pull talk to a remote
  script: |
    import json, os, shlex, subprocess, sys

    a = json.loads(sys.argv[1])

    def arg(k):
        v = a.get(k)
        return "" if v is None else str(v).strip()

    def flag(k):
        v = a.get(k)
        return v is True or str(v).strip().lower() == "true"

    def split(k):
        try:
            return shlex.split(arg(k))
        except ValueError as e:
            sys.exit("git: %s: %s" % (k, e))

    paths, extra, op = split("paths"), split("args"), arg("operation")
    remote, branch, wd = arg("remote") or "origin", arg("branch"), arg("workingDir") or None
    dry = flag("dryRun")
    env = dict(os.environ, GIT_PAGER="cat", GIT_TERMINAL_PROMPT="0")

    def git(*argv, cwd=wd):
        p = subprocess.run(["git", *argv], cwd=cwd, env=env, capture_output=True,
                           text=True, encoding="utf-8", errors="replace")
        if p.returncode != 0:
            msg = "git %s failed: exit status %d" % (argv[0], p.returncode)
            if p.stderr.strip():
                msg += ": " + p.stderr.strip()
            sys.exit(msg)
        return p.stdout

    def lines(out):
        return out.split("\n")

    def status():
        r = {"branch": "", "staged": [], "unstaged": [], "untracked": [], "conflicts": []}
        for line in lines(git("status", "--porcelain", "-b")):
            if line.startswith("##"):
                r["branch"] = line[2:].split("...", 1)[0].strip()
            elif len(line) >= 4:
                xy, f = line[:2], line[3:].strip()
                if xy == "??":
                    r["untracked"].append(f)
                elif "U" in xy:
                    r["conflicts"].append(f)
                elif xy[0] not in " ?":
                    r["staged"].append(f)
                elif xy[1] != " ":
                    r["unstaged"].append(f)
        return r

    def diff():
        out = git("diff", "--no-color", *(["--", *paths] if paths else []))
        return {"diff": out,
                "additions": out.count("\n+") - out.count("\n+++"),
                "deletions": out.count("\n-") - out.count("\n---")}

    def log():
        try:
            n = int(float(arg("maxCount") or 0))
        except ValueError:
            n = 0
        out = git("log", "--max-count=%d" % (n if n > 0 else 10), "--format=%H|%an|%ae|%ai|%s",
                  *(["--", *paths] if paths else []))
        commits = []
        for line in lines(out):
            if line.count("|") >= 4:
                h, an, ae, ad, msg = line.split("|", 4)
                commits.append({"hash": h, "author": an, "email": ae, "date": ad, "message": msg})
        return {"commits": commits, "count": len(commits)}

    def branches():
        r, current = [], ""
        for line in lines(git("branch", "-a")):
            if len(line) < 2:
                continue
            name = line[1:].strip()
            is_remote = name.startswith("remotes/")
            name = name[len("remotes/"):] if is_remote else name
            if line[0] == "*":
                current = name
            r.append({"name": name, "current": line[0] == "*", "remote": is_remote})
        return {"branches": r, "current": current, "count": len(r)}

    def remotes():
        r, seen = [], set()
        for line in lines(git("remote", "-v")):
            f = line.split()
            if len(f) >= 3 and (f[0], f[1]) not in seen:
                seen.add((f[0], f[1]))
                r.append({"name": f[0], "url": f[1], "type": f[2]})
        return {"remotes": r, "count": len(r)}

    def add():
        if dry:
            return {"dryRun": True, "staged": False}
        git("add", *(paths or ["."]))
        return {"staged": True}

    def commit():
        msg = arg("message")
        if not msg:
            sys.exit("git: commit message is required")
        if dry:
            return {"dryRun": True, "message": msg, "committed": False}
        git("commit", "-m", msg)
        return {"message": msg, "committed": True}

    def checkout():
        if dry:
            return {"dryRun": True, "branch": branch, "paths": paths, "switched": False}
        git("checkout", *([branch] if branch else []), *(["--", *paths] if paths else []))
        return {"branch": branch, "paths": paths, "switched": True}

    def init():
        if dry:
            return {"dryRun": True, "initialized": False}
        git("init")
        return {"path": wd or os.getcwd(), "initialized": True}

    def clone():
        url = arg("url")
        if dry:
            return {"dryRun": True, "url": url, "cloned": False}
        # Clone runs from the working directory; workingDir is the target.
        git("clone", url, *([wd] if wd else []), cwd=None)
        return {"url": url, "cloned": True}

    def sync():
        key = op + "ed"
        if dry:
            return {"dryRun": True, "remote": remote, "branch": branch, key: False}
        git(op, remote, *([branch] if branch else []))
        return {"remote": remote, "branch": branch, key: True}

    OPS = {"status": status, "diff": diff, "log": log, "branch": branches, "remote": remotes,
           "add": add, "commit": commit, "checkout": checkout, "init": init, "clone": clone,
           "push": sync, "pull": sync}
    if op == "show":
        result = {"output": git("show", "--no-color", "--format=fuller", *extra)}
    elif op in OPS:
        result = OPS[op]()
    else:
        sys.exit("git: unsupported operation '%s'" % op)
    result["success"] = True
    print(json.dumps(result, indent=2, sort_keys=True, ensure_ascii=False))