#!/usr/bin/env python3 """Preflight and execute git push with optional Gitea token fallback.""" from __future__ import annotations import argparse import json import shlex import subprocess import sys from common import ( build_authenticated_push_url, current_branch, ensure_git_repo, get_current_user_login, get_upstream, is_auth_error, load_token, mask_url, remote_branch_sha, resolve_repo, worktree_changes, ahead_behind, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Preflight and optionally push the current branch to Gitea." ) parser.add_argument("--repo-url", help="Explicit target repo URL.") parser.add_argument("--repo", help="Shorthand owner/repo. Requires GITEA_BASE_URL.") parser.add_argument("--remote", default="origin", help="Git remote name. Default: origin") parser.add_argument("--branch", help="Branch to push. Default: current branch") parser.add_argument( "--execute", action="store_true", help="Run git push after preflight instead of only printing the plan.", ) parser.add_argument( "--force", action="store_true", help="Allow force push. Uses --force-with-lease under the hood.", ) return parser.parse_args() def build_push_command(target: str, branch: str, force: bool) -> list[str]: command = ["git", "push"] if force: command.append("--force-with-lease") command.extend([target, f"{branch}:{branch}"]) return command def command_text(command: list[str]) -> str: return " ".join(shlex.quote(part) for part in command) def compute_push_plan( repo_url: str | None = None, repo: str | None = None, remote: str = "origin", branch: str | None = None, force: bool = False, ) -> dict: ensure_git_repo() context = resolve_repo(repo_url=repo_url, repo=repo, remote=remote) branch_name = branch or current_branch() dirty_lines = worktree_changes() upstream = get_upstream(branch_name) compare_ref = upstream remote_exists = True ahead = 0 behind = 0 if upstream: behind, ahead = ahead_behind(upstream) else: remote_sha = remote_branch_sha(remote, branch_name) if remote_sha: compare_ref = f"{remote}/{branch_name}" behind, ahead = ahead_behind(remote_sha) else: remote_exists = False compare_ref = None ahead = None behind = 0 diverged = bool(remote_exists and behind > 0 and (ahead or 0) > 0) behind_only = bool(remote_exists and behind > 0 and (ahead or 0) == 0) needs_push = (not remote_exists) or ((ahead or 0) > 0) status = "ready" if diverged and not force: status = "blocked" elif not needs_push: status = "noop" messages: list[str] = [] if dirty_lines: messages.append("Working tree is dirty; push will include only committed changes.") if not remote_exists: messages.append("Remote branch does not exist yet; push will create it.") if behind_only: messages.append("Local branch is behind the remote branch; there are no local commits to push.") if diverged and not force: messages.append("Branch has diverged from remote; rerun with explicit force only if that is intended.") push_command = build_push_command(remote, branch_name, force) fallback_command = None token_available = bool(load_token(required=False)) if token_available: try: token = load_token(required=False) login = get_current_user_login(context.origin, token) auth_url = build_authenticated_push_url(context.repo_url, login, token) fallback_command = build_push_command(auth_url, branch_name, force) except SystemExit as exc: messages.append(f"Failed to prepare HTTPS token fallback: {exc}") return { "repo_url": context.repo_url, "origin": context.origin, "owner": context.owner, "repo": context.repo, "remote": remote, "branch": branch_name, "upstream": upstream, "compare_ref": compare_ref, "remote_branch_exists": remote_exists, "dirty": bool(dirty_lines), "dirty_paths": dirty_lines, "ahead": ahead, "behind": behind, "diverged": diverged, "behind_only": behind_only, "needs_push": needs_push, "status": status, "messages": messages, "push_command": command_text(push_command), "fallback_push_command": command_text( [ *(fallback_command[:-2] if fallback_command else []), mask_url(fallback_command[-2]) if fallback_command else "", *(fallback_command[-1:] if fallback_command else []), ] ) if fallback_command else None, } def run_command(command: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(command, capture_output=True, text=True) def execute_push(plan: dict, force: bool = False) -> dict: if plan["status"] == "noop": return { "status": "noop", "executed": False, "reason": "nothing_to_push", "messages": plan["messages"], } if plan["diverged"] and not force: raise SystemExit( "Branch has diverged from remote. Refusing to push without explicit --force." ) primary_command = build_push_command(plan["remote"], plan["branch"], force) primary_result = run_command(primary_command) if primary_result.returncode == 0: return { "status": "pushed", "executed": True, "method": "remote", "command": command_text(primary_command), "stdout": primary_result.stdout.strip(), "stderr": primary_result.stderr.strip(), } primary_stderr = primary_result.stderr.strip() or primary_result.stdout.strip() if not is_auth_error(primary_stderr): raise SystemExit(f"git push failed: {primary_stderr}") token = load_token(required=True) login = get_current_user_login(plan["origin"], token) auth_url = build_authenticated_push_url(plan["repo_url"], login, token) fallback_command = build_push_command(auth_url, plan["branch"], force) fallback_result = run_command(fallback_command) if fallback_result.returncode == 0: return { "status": "pushed", "executed": True, "method": "https-token", "command": command_text( [ *(fallback_command[:-2]), mask_url(fallback_command[-2]), fallback_command[-1], ] ), "stdout": fallback_result.stdout.strip(), "stderr": fallback_result.stderr.strip(), } fallback_stderr = fallback_result.stderr.strip() or fallback_result.stdout.strip() raise SystemExit( "git push failed with remote auth and HTTPS token fallback: " f"{fallback_stderr}" ) def main() -> int: args = parse_args() plan = compute_push_plan( repo_url=args.repo_url, repo=args.repo, remote=args.remote, branch=args.branch, force=args.force, ) if not args.execute: print(json.dumps(plan, ensure_ascii=False, indent=2)) return 0 result = execute_push(plan, force=args.force) payload = {"plan": plan, "result": result} print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())