366 lines
16 KiB
Python
366 lines
16 KiB
Python
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()
|