gitpush.py

#!/usr/bin/env python3
"""
gitpush - minimal pure-Python git pusher for your scripts and code, built on Dulwich.

Point it at a folder, it pushes to GitHub. No git binary required.
No --repo needed: the folder name becomes the repo name, and the repo
is created on GitHub automatically if it doesn't exist.

    pip install dulwich
    export GITHUB_TOKEN=...
    python gitpush.py push ./myscripts
"""

import argparse
import fnmatch
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

try:
    from dulwich import porcelain
    from dulwich.repo import Repo
except ImportError:
    sys.exit("Dulwich not installed. Run:  pip install dulwich")

DEFAULT_BRANCH = "main"

# junk that should never get pushed from a code folder
SKIP_PARTS = {"__pycache__", ".venv", "venv", "node_modules", ".idea", ".vscode"}
SKIP_NAMES = {".DS_Store", "Thumbs.db", ".gitpushignore", "_gitpush_ignore.txt"}
SKIP_SUFFIXES = (".pyc", ".pyo", ".swp")


def is_junk(relpath: str) -> bool:
    p = Path(relpath)
    if set(p.parts) & SKIP_PARTS:
        return True
    if p.name in SKIP_NAMES:
        return True
    return p.name.endswith(SKIP_SUFFIXES)


IGNORE_FILENAMES = (".gitpushignore", "_gitpush_ignore.txt")


IGNORE_TEMPLATE = """\
# gitpush ignore file - one pattern per line, wildcards allowed
# In a project folder: matching files/folders are never pushed.
# In a parent folder:  'pushall' skips matching subfolders entirely.
# Lines starting with # are comments. Examples (remove # to activate):

# models
# experiments
# *.bin
# secret*
"""


def ensure_ignore_file(folder):
    """Create a starter _gitpush_ignore.txt on first run if none exists."""
    root = Path(folder)
    if any((root / name).is_file() for name in IGNORE_FILENAMES):
        return
    target = root / "_gitpush_ignore.txt"
    try:
        target.write_text(IGNORE_TEMPLATE)
        print(f"Created {target} - add patterns there to exclude things from pushes.")
    except OSError:
        pass  # read-only location etc. - not worth failing over


def load_ignore_patterns(root) -> list:
    """Read ignore patterns from .gitpushignore or _gitpush_ignore.txt in a folder."""
    for name in IGNORE_FILENAMES:
        f = Path(root) / name
        if f.is_file():
            return [
                ln.strip() for ln in f.read_text().splitlines()
                if ln.strip() and not ln.strip().startswith("#")
            ]
    return []


def matches_patterns(rel: str, patterns) -> bool:
    if not patterns:
        return False
    p = Path(rel)
    for pat in patterns:
        pat = pat.rstrip("/")
        if (fnmatch.fnmatch(rel, pat)
                or fnmatch.fnmatch(p.name, pat)
                or any(fnmatch.fnmatch(part, pat) for part in p.parts)):
            return True
    return False


def check_root(folder: str):
    """If GITPUSH_ROOT is set, refuse to touch anything outside it."""
    root = os.environ.get("GITPUSH_ROOT", "").strip()
    if not root:
        return
    rootp = Path(root).expanduser().resolve()
    target = Path(folder).expanduser().resolve()
    if target != rootp and rootp not in target.parents:
        sys.exit(
            f"Refusing: {target}\n"
            f"is outside GITPUSH_ROOT ({rootp}).\n"
            f"Move the folder inside it, or unset/adjust GITPUSH_ROOT."
        )


# ---------- helpers ----------

_token_cache = None


def token_from(args) -> str:
    global _token_cache
    if _token_cache:
        return _token_cache
    tok = getattr(args, "token", None) or os.environ.get("GITHUB_TOKEN", "")
    if not tok:
        print("GITHUB_TOKEN not set. (Tip: export GITHUB_TOKEN=... to skip this prompt.)")
        try:
            import getpass
            tok = getpass.getpass("Paste GitHub token: ").strip()
        except Exception:
            tok = input("Paste GitHub token: ").strip()
    if not tok:
        sys.exit("No token given - aborting.")
    _token_cache = tok
    return tok


_username_cache = None


def username_from(tok: str) -> str:
    """GITHUB_USER env var if set, otherwise ask; falls back to the GitHub API."""
    global _username_cache
    if _username_cache:
        return _username_cache
    user = os.environ.get("GITHUB_USER", "").strip()
    if not user:
        print("GITHUB_USER not set. (Tip: export GITHUB_USER=yourname to skip this prompt.)")
        user = input("GitHub username (Enter = look it up from token): ").strip()
    if not user:
        user = gh_api("GET", "/user", tok)["login"]
    _username_cache = user
    return user


def author_identity(tok: str) -> bytes:
    """'username <email>' for commits. Email: GITHUB_EMAIL env, or derived
    from your username as GitHub's noreply address so commits link to your
    profile and count on your contribution graph."""
    user = username_from(tok)
    email = os.environ.get("GITHUB_EMAIL", "").strip()
    if not email:
        try:
            uid = gh_api("GET", f"/users/{user}", tok)["id"]
            email = f"{uid}+{user}@users.noreply.github.com"
        except Exception:
            email = f"{user}@users.noreply.github.com"
    return f"{user} <{email}>".encode()


def prompt_repo_metadata() -> dict:
    """First push of a new repo: ask for optional details, Enter skips each."""
    print("New repo - optional details (press Enter to skip any):")
    meta = {}
    desc = input("  Description: ").strip()
    if desc:
        meta["description"] = desc
    homepage = input("  Website URL: ").strip()
    if homepage:
        meta["homepage"] = homepage
    topics = input("  Topics (comma-separated, e.g. python,cli): ").strip()
    if topics:
        meta["_topics"] = [t.strip().lower().replace(" ", "-") for t in topics.split(",") if t.strip()]
    return meta


def gh_api(method: str, path: str, tok: str, payload: dict = None):
    """Tiny GitHub REST client using only the stdlib."""
    req = urllib.request.Request(
        f"https://api.github.com{path}",
        data=json.dumps(payload).encode() if payload else None,
        method=method,
        headers={
            "Authorization": f"Bearer {tok}",
            "Accept": "application/vnd.github+json",
            "Content-Type": "application/json",
            "User-Agent": "gitpush",
        },
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())


def create_repo(tok: str, name: str, private: bool, meta: dict = None) -> dict:
    meta = meta or {}
    payload = {"name": name, "private": private}
    payload.update({k: v for k, v in meta.items() if not k.startswith("_")})
    try:
        data = gh_api("POST", "/user/repos", tok, payload)
        if meta.get("_topics"):
            try:
                gh_api("PUT", f"/repos/{data['full_name']}/topics", tok, {"names": meta["_topics"]})
                print(f"  Topics set: {', '.join(meta['_topics'])}")
            except urllib.error.HTTPError:
                print("  (couldn't set topics - continuing anyway)")
        return data
    except urllib.error.HTTPError as e:
        body = e.read().decode(errors="replace")
        if e.code == 422:
            sys.exit(
                f"GitHub says a repo named '{name}' already exists, but your token "
                f"can't see it. If it's a fine-grained token, add that repo under "
                f"'Repository access'.\n{body}"
            )
        if e.code in (401, 403, 404):
            sys.exit(
                f"GitHub refused to create the repo ({e.code}). Fine-grained tokens "
                f"need 'Administration: write' permission - or use a classic token "
                f"with 'repo' scope.\n{body}"
            )
        sys.exit(f"GitHub error {e.code}: {body}")


def resolve_repo(args, repo_path: str, tok: str) -> str:
    """Return 'user/name'. Derive from folder if --repo missing; auto-create if absent on GitHub."""
    if args.repo:
        full = args.repo if "/" in args.repo else f"{username_from(tok)}/{args.repo}"
    else:
        full = f"{username_from(tok)}/{Path(repo_path).resolve().name}"
        print(f"No --repo given, using: {full}")
    try:
        gh_api("GET", f"/repos/{full}", tok)
    except urllib.error.HTTPError as e:
        if e.code == 404:
            print(f"Repo {full} doesn't exist on GitHub - creating it...")
            meta = prompt_repo_metadata()
            data = create_repo(tok, full.split("/", 1)[1], getattr(args, "private", False), meta)
            print(f"Created: {data['html_url']}  ({'private' if data['private'] else 'public'})")
        else:
            raise
    return full


def auth_url(full_repo: str, tok: str) -> str:
    # username:token format works for classic (ghp_) and fine-grained (github_pat_) tokens
    return f"https://x-access-token:{tok}@github.com/{full_repo}.git"


def open_or_init(folder: str) -> Repo:
    p = Path(folder)
    p.mkdir(parents=True, exist_ok=True)
    if (p / ".git").exists():
        return Repo(str(p))
    print(f"Initializing new git repo in {p}")
    return porcelain.init(str(p))


def expand_entry(repo_path: str, rel: str):
    """Dulwich status can report bare directories; expand them to files."""
    root = Path(repo_path)
    p = root / rel
    if p.is_dir():
        out = []
        for f in sorted(p.rglob("*")):
            if f.is_file():
                r = str(f.relative_to(root))
                if not is_junk(r):
                    out.append(r)
        return out
    return [] if is_junk(rel) else [rel]


def collect_changes(repo_path: str):
    """Return (new, modified, deleted) relative paths."""
    patterns = load_ignore_patterns(repo_path)
    st = porcelain.status(repo_path)
    new = []
    for f in st.untracked:
        f = f.decode() if isinstance(f, bytes) else f
        new.extend(x for x in expand_entry(repo_path, f) if not matches_patterns(x, patterns))
    modified = [f.decode() for f in st.unstaged]
    modified = [f for f in modified if not is_junk(f) and not matches_patterns(f, patterns)]
    deleted = []
    # staged dict: {'add': [...], 'modify': [...], 'delete': [...]}
    for k, files in st.staged.items():
        for f in files:
            f = f.decode() if isinstance(f, bytes) else f
            if k == "delete":
                deleted.append(f)
            elif k == "add" and f not in new:
                new.append(f)
            elif f not in modified:
                modified.append(f)
    return new, modified, deleted


# ---------- commands ----------

def cmd_push(args):
    repo_path = args.folder
    check_root(repo_path)
    open_or_init(repo_path)
    ensure_ignore_file(repo_path)

    has_readme = any(
        f.is_file() and f.name.lower().startswith("readme")
        for f in Path(repo_path).iterdir()
    )
    if not has_readme:
        print("Note: no README in this folder. Repos look better (and rank better) with one.")

    new, modified, deleted = collect_changes(repo_path)
    all_changes = new + modified + deleted

    if all_changes:
        print(f"New: {len(new)}  Modified: {len(modified)}  Deleted: {len(deleted)}")
        for f in all_changes:
            print(f"  {f}")
    else:
        print("Working tree clean - will push any unpushed commits.")

    if args.dry_run:
        print("(dry run - nothing committed or pushed)")
        return

    if all_changes:
        to_add = [str((Path(repo_path) / f).resolve()) for f in new + modified]
        if to_add:
            added, ignored = None, None
            result = porcelain.add(repo_path, paths=to_add)
            if isinstance(result, tuple):
                added, ignored = result
                if not added:
                    sys.exit("Staging failed - dulwich added 0 files. Nothing committed.")
                print(f"Staged {len(added)} file(s)")
        for f in deleted:
            porcelain.remove(repo_path, paths=[str((Path(repo_path) / f).resolve())])

        msg = args.message or "made some changes"
        committer = author_identity(token_from(args))
        try:
            porcelain.commit(repo_path, message=msg, no_verify=True,
                             committer=committer, author=committer)
        except TypeError:  # older dulwich without no_verify
            porcelain.commit(repo_path, message=msg,
                             committer=committer, author=committer)
        print(f"Committed: {msg}")

    tok = token_from(args)
    full = resolve_repo(args, repo_path, tok)
    # local branch may be 'master' (dulwich default) while remote should be 'main'
    try:
        local = porcelain.active_branch(repo_path)
        local = local.decode() if isinstance(local, bytes) else local
    except Exception:
        local = args.branch
    refspec = f"refs/heads/{local}:refs/heads/{args.branch}"
    if args.force:
        refspec = "+" + refspec
    try:
        porcelain.push(repo_path, auth_url(full, tok), refspec, force=args.force)
    except TypeError:  # older dulwich without force kwarg - refspec '+' still forces
        porcelain.push(repo_path, auth_url(full, tok), refspec)
    print(f"{'Force-pushed' if args.force else 'Pushed'} to github.com/{full} ({args.branch})")


def looks_like_repo_spec(arg: str) -> bool:
    """'user/name' that isn't an existing local path."""
    return arg.count("/") == 1 and not arg.startswith((".", "/", "~")) and not Path(arg).is_dir()


def cmd_clone(args):
    tok = token_from(args)
    spec = args.repo
    if not spec and args.folder != "." and looks_like_repo_spec(args.folder):
        spec = args.folder  # allow: gitpush clone user/name
    if not spec and args.folder != ".":
        spec = args.folder  # allow: gitpush clone name  (your own repo)
    if not spec:
        sys.exit("clone needs a repo:  gitpush clone user/name")
    if "/" not in spec:
        spec = f"{username_from(tok)}/{spec}"
    target = spec.split("/")[-1]
    porcelain.clone(auth_url(spec, tok), target)
    print(f"Cloned github.com/{spec} into ./{target}")


def cmd_pull(args):
    tok = token_from(args)
    arg = args.folder
    if not args.repo and looks_like_repo_spec(arg):
        # allow: gitpush pull user/name  -> pulls into ./name (clones if absent)
        full = arg
        folder = arg.split("/")[1]
        if not (Path(folder) / ".git").exists():
            porcelain.clone(auth_url(full, tok), folder)
            print(f"No local copy - cloned github.com/{full} into ./{folder}")
            return
    else:
        folder = arg
        full = resolve_repo(args, folder, tok)
    porcelain.pull(folder, auth_url(full, tok), f"refs/heads/{args.branch}")
    print(f"Pulled {args.branch} from github.com/{full} into {folder}")


def cmd_status(args):
    new, modified, deleted = collect_changes(args.folder)
    if not (new or modified or deleted):
        print("Clean - nothing new to push.")
        return
    for label, files in (("New", new), ("Modified", modified), ("Deleted", deleted)):
        for f in files:
            print(f"{label:>9}: {f}")


def cmd_log(args):
    repo = Repo(args.folder)
    for entry in repo.get_walker(max_entries=args.count):
        c = entry.commit
        print(f"{c.id.decode()[:8]}  {c.message.decode(errors='replace').strip().splitlines()[0]}")


def cmd_init(args):
    open_or_init(args.folder)
    print("Ready. Now:  gitpush push", args.folder)


def cmd_create(args):
    tok = token_from(args)
    data = create_repo(tok, args.name, args.private, prompt_repo_metadata())
    print(f"Created: {data['html_url']}  ({'private' if data['private'] else 'public'})")
    print(f"Now:  gitpush push YOUR_FOLDER -r {data['full_name']}")


def cmd_pushall(args):
    root = Path(args.folder)
    check_root(root)
    if not root.is_dir():
        sys.exit(f"Not a folder: {root}")
    ensure_ignore_file(root)
    patterns = load_ignore_patterns(root)
    subs, skipped = [], []
    for d in sorted(root.iterdir()):
        if not d.is_dir() or d.name.startswith(".") or d.name in SKIP_PARTS:
            continue
        if matches_patterns(d.name, patterns):
            skipped.append(d.name)
        else:
            subs.append(d)
    if skipped:
        print(f"Ignored ({IGNORE_FILENAMES[0]}): {', '.join(skipped)}")
    if not subs:
        sys.exit(f"No subfolders to push in {root}")

    print(f"These {len(subs)} folder(s) will each be pushed to their own GitHub repo:")
    for d in subs:
        print(f"  {d.name}  ->  {os.environ.get('GITHUB_USER', 'you')}/{d.name}")
    if args.dry_run:
        print("(dry run - showing changes per folder, pushing nothing)\n")
    elif not args.yes:
        ans = input(f"Continue? [y/N] ").strip().lower()
        if ans not in ("y", "yes"):
            sys.exit("Aborted - nothing pushed.")

    ok, failed = [], []
    for d in subs:
        print(f"\n=== {d.name} ===")
        sub = argparse.Namespace(**vars(args))
        sub.folder = str(d)
        sub.repo = None  # each folder becomes its own repo
        try:
            cmd_push(sub)
            ok.append(d.name)
        except SystemExit as e:
            failed.append((d.name, str(e)))
            print(f"  stopped: {e}")
        except Exception as e:
            failed.append((d.name, f"{type(e).__name__}: {e}"))
            print(f"  FAILED: {type(e).__name__}: {e}")

    print(f"\n{'=' * 30}\nPushed {len(ok)}/{len(subs)} folder(s).")
    for name, err in failed:
        print(f"  {name}: {err}")


# ---------- CLI ----------

def main():
    p = argparse.ArgumentParser(
        prog="gitpush",
        description="Minimal Dulwich-based GitHub pusher. Folder name = repo name, auto-created.",
    )
    sub = p.add_subparsers(dest="command", required=True)

    def common(sp, needs_repo=True):
        sp.add_argument("folder", nargs="?", default=".", help="local folder (default: current dir)")
        if needs_repo:
            sp.add_argument("--repo", "-r", help="GitHub repo as user/name (default: you/foldername)")
            sp.add_argument("--token", "-t", help="GitHub token (or set GITHUB_TOKEN)")
        sp.add_argument("--branch", "-b", default=DEFAULT_BRANCH, help=f"branch (default: {DEFAULT_BRANCH})")

    sp = sub.add_parser("push", help="add all changes, commit, push; creates the GitHub repo if needed")
    common(sp)
    sp.add_argument("--message", "-m", help="commit message (default: 'made some changes')")
    sp.add_argument("--dry-run", "-n", action="store_true", help="show what would be pushed, do nothing")
    sp.add_argument("--private", "-p", action="store_true", help="if auto-creating the repo, make it private")
    sp.add_argument("--force", "-f", action="store_true", help="force push (overwrites remote history)")
    sp.set_defaults(func=cmd_push)

    sp = sub.add_parser("pushall", help="push every subfolder of a folder to its own repo")
    sp.add_argument("folder", nargs="?", default=".", help="parent folder (default: current dir)")
    sp.add_argument("--token", "-t", help="GitHub token (or set GITHUB_TOKEN)")
    sp.add_argument("--branch", "-b", default=DEFAULT_BRANCH, help=f"branch (default: {DEFAULT_BRANCH})")
    sp.add_argument("--message", "-m", help="commit message for all folders")
    sp.add_argument("--dry-run", "-n", action="store_true", help="preview all folders, push nothing")
    sp.add_argument("--private", "-p", action="store_true", help="auto-created repos are private")
    sp.add_argument("--force", "-f", action="store_true", help="force push each folder")
    sp.add_argument("--yes", "-y", action="store_true", help="skip the confirmation prompt")
    sp.set_defaults(func=cmd_pushall)

    sp = sub.add_parser("clone", help="clone a GitHub repo")
    common(sp)
    sp.set_defaults(func=cmd_clone)

    sp = sub.add_parser("pull", help="pull latest from GitHub")
    common(sp)
    sp.set_defaults(func=cmd_pull)

    sp = sub.add_parser("status", help="show unpushed changes")
    common(sp, needs_repo=False)
    sp.set_defaults(func=cmd_status)

    sp = sub.add_parser("log", help="show recent commits")
    common(sp, needs_repo=False)
    sp.add_argument("--count", "-c", type=int, default=10, help="number of commits (default: 10)")
    sp.set_defaults(func=cmd_log)

    sp = sub.add_parser("init", help="initialize a folder as a repo")
    common(sp, needs_repo=False)
    sp.set_defaults(func=cmd_init)

    sp = sub.add_parser("create", help="create a new repo on GitHub (push does this automatically)")
    sp.add_argument("name", help="repo name (no username, just the name)")
    sp.add_argument("--private", "-p", action="store_true", help="make it private (default: public)")
    sp.add_argument("--token", "-t", help="GitHub token (or set GITHUB_TOKEN)")
    sp.set_defaults(func=cmd_create)

    args = p.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()