remote_branches.py

"""Read remote branch names without creating or modifying a repository."""
import os
from pathlib import Path
import re
import shutil
import signal
import subprocess
import threading
from urllib.parse import urlsplit
from network_settings import git_environment
from github_auth import github_login, needs_github_login

_slots = threading.BoundedSemaphore(3)


def validate_url(value):
    url = str(value or "").strip()
    if not url or any(ord(c) < 32 for c in url) or url.startswith("-"):
        raise ValueError("请输入有效的 Git 仓库地址。")
    if re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://", url):
        parsed = urlsplit(url)
        if parsed.scheme not in {"https", "http", "ssh", "git", "file"}:
            raise ValueError("支持 HTTPS、SSH、git://、file:// 或本地仓库路径。")
        if parsed.password:
            raise ValueError("地址中请勿包含密码，请使用 Git 凭据管理器或 SSH 密钥。")
        if parsed.scheme in {"http", "https"} and (parsed.username or parsed.query):
            raise ValueError("地址中请勿包含账号、密码或令牌，请使用 Git 凭据管理器。")
    elif "::" in url:
        raise ValueError("不支持自定义 Git remote helper 地址。")
    return url


def parse_refs(output):
    branches = set()
    default = None
    for line in output.splitlines():
        left, sep, ref = line.partition("\t")
        if not sep:
            continue
        if ref == "HEAD" and left.startswith("ref: refs/heads/"):
            default = left[len("ref: refs/heads/"):]
        elif ref.startswith("refs/heads/") and re.fullmatch(r"[0-9a-fA-F]{40,64}", left):
            branches.add(ref[len("refs/heads/"):])
    if default not in branches:
        default = None
    return {"branches": sorted(branches, key=lambda name: (name != default, name.casefold(), name)),
            "default_branch": default}


def list_branches(url, timeout=45, settings=None):
    url = validate_url(url)
    if not _slots.acquire(blocking=False):
        raise ValueError("分支查询较多，请稍后点击刷新重试。")
    try:
        version = github_login.version()
        try:
            return _query_branches(url, timeout, settings)
        except ValueError as exc:
            if not needs_github_login(url, str(exc)):
                raise
        github_login.ensure_authenticated(settings, version)
        try:
            return _query_branches(url, timeout, settings)
        except ValueError as exc:
            if needs_github_login(url, str(exc)):
                raise ValueError("浏览器登录后仍无法访问仓库，请确认登录账号拥有该仓库权限。\n" + str(exc)) from None
            raise
    finally:
        _slots.release()


def _query_branches(url, timeout, settings):
    # A browser login is handled by the caller only after a credential failure.
    # This subprocess must never wait for terminal input.
    env = git_environment(settings, url)
    env.update(GIT_TERMINAL_PROMPT="0", GCM_INTERACTIVE="Never", LC_ALL="C")
    env.setdefault("GIT_SSH_COMMAND", "ssh -o BatchMode=yes -o ConnectTimeout=15 -o ServerAliveInterval=10 -o ServerAliveCountMax=2")
    command = [shutil.which("git") or "git", "-c", "http.version=HTTP/1.1",
               "-c", "http.lowSpeedLimit=1024", "-c", "http.lowSpeedTime=20",
               "ls-remote", "--symref", "--", url, "HEAD", "refs/heads/*"]
    proc = subprocess.Popen(command, cwd=Path(__file__).resolve().parent, env=env,
                            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                            creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
                            start_new_session=os.name != "nt")
    try:
        output, error = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        if os.name == "nt":
            subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
                           capture_output=True, creationflags=subprocess.CREATE_NO_WINDOW)
        else:
            try:
                os.killpg(proc.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        proc.kill() if proc.poll() is None else None
        proc.communicate()
        raise ValueError("获取分支超时，请检查网络或代理后点击刷新。") from None
    if proc.returncode:
        detail = error.decode("utf-8", errors="replace").strip()[-1800:]
        raise ValueError("获取分支失败，请检查仓库地址、网络和本机 Git 访问权限。\n" + detail)
    return parse_refs(output.decode("utf-8", errors="replace"))