import base64 import asyncio import os import shutil import subprocess import tempfile from pathlib import Path from typing import Tuple from urllib.parse import urlsplit, urlunsplit class GitService: """Run Git synchronization without persisting repository credentials.""" def _clean_repo_url(self, repo_url: str) -> str: """Remove any user info already embedded in an HTTP(S) URL.""" parsed = urlsplit(repo_url) if parsed.scheme not in ("http", "https"): return repo_url host = parsed.hostname or "" if parsed.port: host = f"{host}:{parsed.port}" return urlunsplit((parsed.scheme, host, parsed.path, parsed.query, parsed.fragment)) def _auth_env(self, username: str = None, token: str = None) -> dict: """Pass HTTPS Basic credentials to Git for this process only. 同时禁止交互式提示，避免凭证/主机密钥校验卡住进程： - GIT_TERMINAL_PROMPT=0：禁止 Git 弹出用户名/密码输入 - GIT_ASKPASS=echo：凭证缺失时立即返回空串而不是挂起 - GIT_SSH_COMMAND：SSH 仓库走 BatchMode + 连接超时，不等待交互 """ env = { "GIT_TERMINAL_PROMPT": "0", "GIT_ASKPASS": "echo", "GIT_SSH_COMMAND": "ssh -o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=accept-new", } if username and token: credential = base64.b64encode(f"{username}:{token}".encode()).decode() env.update({ "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "http.extraHeader", "GIT_CONFIG_VALUE_0": f"Authorization: Basic {credential}", }) return env def _run_command(self, cmd: list, cwd: Path, env: dict = None, timeout: int = 60) -> Tuple[bool, str]: """Run a command and return its output without exposing process environment.""" try: run_env = os.environ.copy() run_env.update({"LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"}) if env: run_env.update(env) result = subprocess.run( cmd, cwd=str(cwd), capture_output=True, text=True, encoding="utf-8", env=run_env, timeout=timeout, check=False, ) if result.returncode == 0: return True, result.stdout return False, result.stderr.strip() or result.stdout.strip() or "Git command failed" except subprocess.TimeoutExpired: return False, f"Git 命令超时（{timeout} 秒）：{' '.join(cmd)}" except Exception as exc: # noqa: BLE001 return False, str(exc) def _ensure_git_initialized(self, cwd: Path, repo_url: str) -> Tuple[bool, str]: """Ensure the project has a Git repository pointing to the credential-free URL.""" if not (cwd / ".git").exists(): ok, message = self._run_command(["git", "init"], cwd) if not ok: return False, message ok, message = self._run_command(["git", "branch", "-M", "main"], cwd) if not ok: return False, message for key, value in ( ("core.quotepath", "false"), ("i18n.logOutputEncoding", "utf-8"), ("core.precomposeunicode", "true"), ): ok, message = self._run_command(["git", "config", key, value], cwd) if not ok: return False, message clean_url = self._clean_repo_url(repo_url) ok, _ = self._run_command(["git", "remote", "get-url", "origin"], cwd) if ok: return self._run_command(["git", "remote", "set-url", "origin", clean_url], cwd) return self._run_command(["git", "remote", "add", "origin", clean_url], cwd) def _normalize_sub_path(self, sub_path: str = None) -> str: if not sub_path: return "" normalized = str(sub_path).strip().strip("/") if not normalized: return "" parts = normalized.split("/") if any(part in ("", ".", "..") or part.startswith(".") for part in parts): raise ValueError(f"非法的同步目录: {normalized}") return normalized def _ensure_commit_identity(self, cwd: Path) -> Tuple[bool, str]: ok, name = self._run_command(["git", "config", "user.name"], cwd) if not ok or not name.strip(): ok, message = self._run_command(["git", "config", "user.name", "NEX Docus"], cwd) if not ok: return False, message ok, email = self._run_command(["git", "config", "user.email"], cwd) if not ok or not email.strip(): return self._run_command( ["git", "config", "user.email", "noreply@nex-docus.local"], cwd ) return True, "" def _commit_if_changed(self, cwd: Path, message: str) -> Tuple[bool, str, bool]: ok, status = self._run_command(["git", "status", "--porcelain"], cwd) if not ok: return False, status, False if not status.strip(): return True, "", False ok, identity_message = self._ensure_commit_identity(cwd) if not ok: return False, identity_message, False ok, commit_message = self._run_command(["git", "commit", "-m", message], cwd) return ok, commit_message, ok def _remote_worktree(self, repo_url: str, branch: str, auth_env: dict): """Create an isolated worktree at the latest remote branch revision.""" temp_dir = tempfile.TemporaryDirectory(prefix="nex-docus-git-") root = Path(temp_dir.name) ok, message = self._run_command(["git", "init"], root) if not ok: temp_dir.cleanup() return None, None, message ok, message = self._run_command( ["git", "remote", "add", "origin", self._clean_repo_url(repo_url)], root ) if not ok: temp_dir.cleanup() return None, None, message ok, message = self._run_command(["git", "fetch", "origin"], root, auth_env) if not ok: temp_dir.cleanup() return None, None, message remote_ref = f"origin/{branch}" ok, _ = self._run_command(["git", "rev-parse", "--verify", remote_ref], root) if ok: ok, message = self._run_command(["git", "checkout", "-B", branch, remote_ref], root) else: ok, message = self._run_command(["git", "checkout", "-B", branch], root) if not ok: temp_dir.cleanup() return None, None, message return temp_dir, root, "" def _replace_path(self, source_root: Path, target_root: Path, sub_path: str) -> None: source = source_root / sub_path target = target_root / sub_path if target.exists() or target.is_symlink(): if target.is_dir() and not target.is_symlink(): shutil.rmtree(target) else: target.unlink() if source.exists(): target.parent.mkdir(parents=True, exist_ok=True) if source.is_dir(): shutil.copytree(source, target) else: shutil.copy2(source, target) def _pull( self, project_path: Path, repo_url: str, branch: str = "main", username: str = None, token: str = None, force: bool = False, sub_path: str = None, ) -> Tuple[bool, str]: if not project_path.exists(): return False, "项目目录不存在" sub_path = self._normalize_sub_path(sub_path) auth_env = self._auth_env(username, token) if sub_path: if force: return False, "目录级同步不支持强制拉取" temp_dir, remote_root, message = self._remote_worktree(repo_url, branch, auth_env) if not temp_dir: return False, f"获取远端仓库失败: {message}" try: self._replace_path(remote_root, project_path, sub_path) finally: temp_dir.cleanup() return True, f"已从远端同步目录 [{sub_path}]" ok, message = self._ensure_git_initialized(project_path, repo_url) if not ok: return False, message ok, message = self._run_command(["git", "fetch", "origin"], project_path, auth_env) if not ok: return False, f"获取远端更新失败: {message}" if force: return self._run_command(["git", "reset", "--hard", f"origin/{branch}"], project_path) return self._run_command(["git", "pull", "--ff-only", "origin", branch], project_path, auth_env) async def pull(self, *args, **kwargs) -> Tuple[bool, str]: """Run pull outside the async event loop.""" return await asyncio.to_thread(self._pull, *args, **kwargs) def _push( self, project_path: Path, repo_url: str, branch: str = "main", username: str = None, token: str = None, force: bool = False, sub_path: str = None, ) -> Tuple[bool, str]: if not project_path.exists(): return False, "项目目录不存在" sub_path = self._normalize_sub_path(sub_path) auth_env = self._auth_env(username, token) if sub_path: if force: return False, "目录级推送不支持强制推送" temp_dir, remote_root, message = self._remote_worktree(repo_url, branch, auth_env) if not temp_dir: return False, f"获取远端仓库失败: {message}" try: self._replace_path(project_path, remote_root, sub_path) ok, message = self._run_command(["git", "add", "-A", "--", sub_path], remote_root) if not ok: return False, f"暂存目录失败: {message}" ok, message, changed = self._commit_if_changed( remote_root, f"Update {sub_path} from NEX Docus" ) if not ok: return False, f"提交目录失败: {message}" if not changed: return True, f"目录 [{sub_path}] 没有需要推送的变更" ok, message = self._run_command( ["git", "push", "-u", "origin", branch], remote_root, auth_env ) if not ok: return False, f"推送被远端拒绝，请重新拉取该目录后再试: {message}" return True, f"已推送目录 [{sub_path}]" finally: temp_dir.cleanup() ok, message = self._ensure_git_initialized(project_path, repo_url) if not ok: return False, message ok, message = self._run_command(["git", "fetch", "origin"], project_path, auth_env) if not ok: return False, f"获取远端更新失败: {message}" if not force: ok, _ = self._run_command( ["git", "merge-base", "--is-ancestor", f"origin/{branch}", "HEAD"], project_path ) if not ok: return False, "远端仓库已有未拉取的更新，请先执行整库拉取后再推送" ok, message = self._run_command(["git", "add", "."], project_path) if not ok: return False, f"暂存项目失败: {message}" ok, message, _ = self._commit_if_changed(project_path, "Update from NEX Docus") if not ok: return False, f"提交项目失败: {message}" command = ["git", "push", "-u", "origin", branch] if force: command.append("--force-with-lease") return self._run_command(command, project_path, auth_env) async def push(self, *args, **kwargs) -> Tuple[bool, str]: """Run push outside the async event loop.""" return await asyncio.to_thread(self._push, *args, **kwargs) def _test_connection( self, repo_url: str, branch: str = "main", username: str = None, token: str = None ) -> Tuple[bool, str]: """Verify remote access without changing the project working tree. 通过 Git 配置让连接快速失败，避免错误配置（地址错误、认证失败、 网络不可达）长时间挂起后仅返回"命令超时"： - http.connectTimeout：TCP/TLS 连接阶段超时（含 DNS） - http.lowSpeedLimit/lowSpeedTime：连接后传输过慢视为失败 """ if not repo_url or not str(repo_url).strip(): return False, "仓库地址不能为空" with tempfile.TemporaryDirectory(prefix="nex-docus-git-test-") as temp_path: cwd = Path(temp_path) ok, message = self._run_command( [ "git", "-c", "http.connectTimeout=10", "-c", "http.lowSpeedLimit=1000", "-c", "http.lowSpeedTime=15", "ls-remote", "--heads", self._clean_repo_url(repo_url), f"refs/heads/{branch or 'main'}", ], cwd, self._auth_env(username, token), timeout=30, ) if ok: return True, "" return False, self._humanize_git_error(message) def _humanize_git_error(self, message: str) -> str: """把底层 Git/libcurl 报错翻译成对用户友好的提示。""" if not message: return "无法连接远端仓库" lowered = message.lower() if "超时" in message or "timed out" in lowered or "operation timed out" in lowered: return "连接远端仓库超时，请检查仓库地址、网络或代理设置" if "could not resolve host" in lowered or "name or service not known" in lowered: return "无法解析仓库地址，请检查仓库 URL 是否正确" if ( "connection refused" in lowered or "connection reset" in lowered or "couldn't connect" in lowered or "could not connect" in lowered or "failed to connect" in lowered or "unable to access" in lowered or "connection timed out" in lowered ): return "无法连接到远端仓库，请检查仓库地址、网络或代理设置" if "authentication failed" in lowered or "401" in lowered or "403" in lowered: return "认证失败，请检查用户名与访问令牌是否正确" if "could not read username" in lowered or "terminal prompts disabled" in lowered: return "未提供有效的用户名与访问令牌" if "host key verification failed" in lowered: return "SSH 主机密钥校验失败，请确认该主机已受信任" if "permission denied" in lowered or "publickey" in lowered: return "SSH 认证失败，请检查用户名与访问令牌" if "repository not found" in lowered or "not found" in lowered or "does not exist" in lowered: return "远端仓库不存在或无权访问" return message async def test_connection( self, repo_url: str, branch: str = "main", username: str = None, token: str = None ) -> Tuple[bool, str]: """Run remote authentication test outside the async event loop.""" return await asyncio.to_thread(self._test_connection, repo_url, branch, username, token) def list_remote_tree( self, project_path: Path, repo_url: str, branch: str = "main", username: str = None, token: str = None, ) -> Tuple[bool, str]: if not project_path.exists(): return False, "项目目录不存在" temp_dir, remote_root, message = self._remote_worktree( repo_url, branch, self._auth_env(username, token) ) if not temp_dir: return False, f"获取远端仓库失败: {message}" try: return self._run_command( ["git", "-c", "core.quotepath=false", "ls-tree", "-r", "--name-only", "HEAD"], remote_root, ) finally: temp_dir.cleanup() git_service = GitService()