129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
import subprocess
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple
|
|
from urllib.parse import quote_plus
|
|
|
|
class GitService:
|
|
|
|
def _get_auth_url(self, repo_url: str, username: str = None, token: str = None) -> str:
|
|
"""
|
|
Constructs a URL with authentication credentials.
|
|
Note: This is sensitive, so be careful not to log this URL.
|
|
"""
|
|
if not username or not token:
|
|
return repo_url
|
|
|
|
# Encode credentials to handle special characters (e.g. @, :, /)
|
|
safe_username = quote_plus(username)
|
|
safe_token = quote_plus(token)
|
|
|
|
# Remove scheme if present to insert auth
|
|
if repo_url.startswith("https://"):
|
|
url_body = repo_url[8:]
|
|
return f"https://{safe_username}:{safe_token}@{url_body}"
|
|
elif repo_url.startswith("http://"):
|
|
url_body = repo_url[7:]
|
|
return f"http://{safe_username}:{safe_token}@{url_body}"
|
|
|
|
return repo_url
|
|
|
|
def _run_command(self, cmd: list, cwd: Path) -> Tuple[bool, str]:
|
|
"""
|
|
Runs a shell command in the given directory.
|
|
Returns (success, message).
|
|
"""
|
|
try:
|
|
# git operations might take time
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=str(cwd),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False # We handle return code manually
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return True, result.stdout
|
|
else:
|
|
return False, f"Command failed: {' '.join(cmd)}\nError: {result.stderr}"
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
def _ensure_git_initialized(self, cwd: Path, auth_url: str):
|
|
"""
|
|
Ensures the directory is a git repository and has the correct remote.
|
|
"""
|
|
git_dir = cwd / ".git"
|
|
if not git_dir.exists():
|
|
self._run_command(["git", "init"], cwd)
|
|
self._run_command(["git", "branch", "-M", "main"], cwd) # Default to main
|
|
|
|
# Check remote
|
|
success, output = self._run_command(["git", "remote", "get-url", "origin"], cwd)
|
|
if not success:
|
|
# Remote doesn't exist, add it
|
|
self._run_command(["git", "remote", "add", "origin", auth_url], cwd)
|
|
else:
|
|
# Remote exists, update it (in case credentials or URL changed)
|
|
current_url = output.strip()
|
|
if current_url != auth_url:
|
|
self._run_command(["git", "remote", "set-url", "origin", auth_url], cwd)
|
|
|
|
async def pull(self, project_path: Path, repo_url: str, branch: str = "main", username: str = None, token: str = None, force: bool = False) -> Tuple[bool, str]:
|
|
"""
|
|
Executes git pull.
|
|
"""
|
|
if not project_path.exists():
|
|
return False, "Project path does not exist"
|
|
|
|
auth_url = self._get_auth_url(repo_url, username, token)
|
|
|
|
# Ensure git init and remote
|
|
self._ensure_git_initialized(project_path, auth_url)
|
|
|
|
# Fetch first
|
|
success, msg = self._run_command(["git", "fetch", "origin"], project_path)
|
|
if not success:
|
|
return False, f"Fetch failed: {msg}"
|
|
|
|
if force:
|
|
# Force Reset to remote
|
|
cmd = ["git", "reset", "--hard", f"origin/{branch}"]
|
|
else:
|
|
# Simple pull
|
|
cmd = ["git", "pull", "origin", branch]
|
|
|
|
return self._run_command(cmd, project_path)
|
|
|
|
async def push(self, project_path: Path, repo_url: str, branch: str = "main", username: str = None, token: str = None, force: bool = False) -> Tuple[bool, str]:
|
|
"""
|
|
Executes git push.
|
|
"""
|
|
if not project_path.exists():
|
|
return False, "Project path does not exist"
|
|
|
|
auth_url = self._get_auth_url(repo_url, username, token)
|
|
|
|
# Ensure git init and remote
|
|
self._ensure_git_initialized(project_path, auth_url)
|
|
|
|
# Add all changes
|
|
self._run_command(["git", "add", "."], project_path)
|
|
|
|
# Commit if changes exist
|
|
# Check if there are changes to commit
|
|
status_success, status_output = self._run_command(["git", "status", "--porcelain"], project_path)
|
|
if status_success and status_output.strip():
|
|
# Create a commit
|
|
self._run_command(["git", "commit", "-m", "Update from Nex Docus"], project_path)
|
|
|
|
# Push
|
|
cmd = ["git", "push", "-u", "origin", branch]
|
|
if force:
|
|
cmd.append("--force")
|
|
|
|
return self._run_command(cmd, project_path)
|
|
|
|
git_service = GitService()
|