Developer
113 lines
3.3 KiB
Python
113 lines
3.3 KiB
Python
"""Clone service for repository cloning and dirty state checking."""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _slugify_directory_name(name: str) -> str:
|
|
"""Return a filesystem-safe, lowercase directory slug."""
|
|
slug = name.lower().strip()
|
|
slug = re.sub(r"[^a-z0-9_-]+", "-", slug)
|
|
slug = re.sub(r"-+", "-", slug).strip("-")
|
|
return slug or "project"
|
|
|
|
|
|
def clone_repository(
|
|
remote_url: str,
|
|
ssh_key_path: str | None,
|
|
instance_dir: str,
|
|
branch: str = "main",
|
|
project_name: str | None = None,
|
|
) -> str:
|
|
"""Clone a git repository into the instance directory.
|
|
|
|
Args:
|
|
remote_url: Git remote URL (SSH or HTTPS)
|
|
ssh_key_path: Path to SSH private key for authentication (optional)
|
|
instance_dir: Path to instance directory
|
|
branch: Branch to clone (default: main)
|
|
project_name: Optional project name used as the clone directory name
|
|
instead of the generic ``repo-clone``.
|
|
|
|
Returns:
|
|
Path to the cloned repository
|
|
"""
|
|
clone_name = _slugify_directory_name(project_name) if project_name else "repo-clone"
|
|
clone_path = Path(instance_dir) / clone_name
|
|
clone_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
env = os.environ.copy()
|
|
if ssh_key_path:
|
|
# Use SSH key for cloning
|
|
env["GIT_SSH_COMMAND"] = f"ssh -i {ssh_key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
|
|
|
|
cmd = [
|
|
"git",
|
|
"clone",
|
|
"--branch", branch,
|
|
"--single-branch",
|
|
remote_url,
|
|
str(clone_path),
|
|
]
|
|
|
|
logger.debug("Cloning repository %s (branch: %s) into %s", remote_url, branch, clone_path)
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
timeout=300,
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
logger.error("Git clone failed: %s", result.stderr)
|
|
raise RuntimeError(f"Failed to clone repository: {result.stderr}")
|
|
|
|
logger.debug("Successfully cloned repository into %s", clone_path)
|
|
return str(clone_path)
|
|
|
|
|
|
def check_dirty_state(clone_path: str) -> tuple[bool, list[str]]:
|
|
"""Check for uncommitted changes in a cloned repository.
|
|
|
|
Args:
|
|
clone_path: Path to the cloned repository
|
|
|
|
Returns:
|
|
Tuple of (is_dirty, list_of_changed_files)
|
|
"""
|
|
result = subprocess.run(
|
|
["git", "-C", clone_path, "status", "--short"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
logger.warning("Failed to check git status: %s", result.stderr)
|
|
return False, []
|
|
|
|
changed_files = [line.strip() for line in result.stdout.split("\n") if line.strip()]
|
|
is_dirty = len(changed_files) > 0
|
|
|
|
return is_dirty, changed_files
|
|
|
|
|
|
def remove_clone_directory(instance_dir: str, project_name: str | None = None) -> None:
|
|
"""Remove the cloned repository from the instance directory.
|
|
|
|
Args:
|
|
instance_dir: Path to instance directory
|
|
project_name: Optional project name used as the clone directory name.
|
|
"""
|
|
clone_name = _slugify_directory_name(project_name) if project_name else "repo-clone"
|
|
clone_path = Path(instance_dir) / clone_name
|
|
if clone_path.exists():
|
|
import shutil
|
|
shutil.rmtree(clone_path)
|
|
logger.debug("Removed clone directory: %s", clone_path)
|