"""Git utilities including proxy management.""" import shutil import signal import subprocess import sys import click from bin.config import config def _get_proxies() -> dict[str, str]: """Get git proxies from config.""" proxies = {} for key, value in config.get_all().items(): if key.startswith("proxy_"): name = key[6:] # Remove 'proxy_' prefix proxies[name] = value return proxies def _get_git_proxy(key: str) -> str: """Get a global git proxy config value.""" result = subprocess.run( ["git", "config", "--global", key], capture_output=True, text=True, check=False, ) return result.stdout.strip() if result.returncode == 0 else "(not set)" def _unset_git_proxy(key: str) -> None: """Unset a global git proxy config value (no-op if not set).""" subprocess.run( ["git", "config", "--global", "--unset", key], capture_output=True, check=False, ) @click.group(name="git") def git_cmd() -> None: """Git utilities.""" pass @git_cmd.group(name="proxy") def proxy_cmd() -> None: """Manage git proxy settings.""" pass def _complete_proxy_names(_ctx, _param, incomplete): """Shell completion for proxy names.""" return [ name for name in _get_proxies() if name.startswith(incomplete) ] @proxy_cmd.command("set") @click.argument("name", shell_complete=_complete_proxy_names) def proxy_set(name: str) -> None: """Set git proxy by name.""" proxies = _get_proxies() if name not in proxies: click.echo(f"Unknown proxy: {name}", err=True) click.echo(f"Available: {', '.join(proxies.keys())}", err=True) sys.exit(1) proxy_url = proxies[name] subprocess.run( ["git", "config", "--global", "http.proxy", proxy_url], check=True ) subprocess.run( ["git", "config", "--global", "https.proxy", proxy_url], check=True ) click.secho(f"Git proxy set to {name}: {proxy_url}", fg="green") @proxy_cmd.command("unset") def proxy_unset() -> None: """Unset git proxy.""" _unset_git_proxy("http.proxy") _unset_git_proxy("https.proxy") click.secho("Git proxy unset", fg="green") @proxy_cmd.command("status") def proxy_status() -> None: """Show current git proxy status.""" http_proxy = _get_git_proxy("http.proxy") https_proxy = _get_git_proxy("https.proxy") proxies = _get_proxies() current_name = next( (p for p, url in proxies.items() if url == http_proxy), None ) click.secho("Git proxy status:", fg="cyan") if current_name: click.echo(f" Active: {current_name}") click.echo(f" http.proxy: {http_proxy}") click.echo(f" https.proxy: {https_proxy}") click.secho("\nAvailable proxies:", fg="cyan") for proxy_name, url in proxies.items(): marker = "*" if proxy_name == current_name else " " click.echo(f" [{marker}] {proxy_name:12} {url}") @proxy_cmd.command("list") def proxy_list_cmd() -> None: """List available git proxies.""" proxies = _get_proxies() click.secho("Available proxies:", fg="cyan") for name, url in proxies.items(): click.echo(f" {name:12} {url}") @proxy_cmd.command("run") @click.argument("name", shell_complete=_complete_proxy_names) @click.argument("command", nargs=-1, required=False) def proxy_run(name: str, command: tuple[str, ...]) -> None: """Set proxy, run a command, auto-unset on exit. If no COMMAND is given, tries to run a program with the same name as the proxy. If that program is not found, sets the proxy and waits for Ctrl-C. """ proxies = _get_proxies() if name not in proxies: click.echo(f"Unknown proxy: {name}", err=True) click.echo(f"Available: {', '.join(proxies.keys())}", err=True) sys.exit(1) # Save current state old_http = _get_git_proxy("http.proxy") old_https = _get_git_proxy("https.proxy") proxy_url = proxies[name] subprocess.run( ["git", "config", "--global", "http.proxy", proxy_url], check=True ) subprocess.run( ["git", "config", "--global", "https.proxy", proxy_url], check=True ) click.secho(f"Git proxy set to {name}: {proxy_url}", fg="green") # Determine what to run cmd_to_run = list(command) if command else None if cmd_to_run is None: if shutil.which(name): cmd_to_run = [name] def cleanup(): if old_http == "(not set)": _unset_git_proxy("http.proxy") else: subprocess.run( ["git", "config", "--global", "http.proxy", old_http], capture_output=True, check=False, ) if old_https == "(not set)": _unset_git_proxy("https.proxy") else: subprocess.run( ["git", "config", "--global", "https.proxy", old_https], capture_output=True, check=False, ) click.secho("Git proxy restored", fg="green") if cmd_to_run: click.secho(f"Running: {' '.join(cmd_to_run)}", fg="cyan") click.secho("Press Ctrl-C to stop and auto-unset proxy\n", fg="cyan") proc = None try: proc = subprocess.Popen(cmd_to_run) proc.wait() except KeyboardInterrupt: if proc is not None: proc.send_signal(signal.SIGINT) try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.terminate() proc.wait() finally: cleanup() if proc is not None: sys.exit(proc.returncode) sys.exit(0) else: click.secho( f"Proxy active. Press Ctrl-C to stop and auto-unset proxy", fg="cyan" ) try: signal.pause() except KeyboardInterrupt: pass finally: cleanup()