448 lines
23 KiB
Python
448 lines
23 KiB
Python
# git_commands.py
|
|
import os
|
|
import subprocess
|
|
import logging
|
|
|
|
class GitCommandError(Exception):
|
|
"""
|
|
Custom exception for handling Git command errors.
|
|
Includes the original command and error details if available.
|
|
"""
|
|
def __init__(self, message, command=None, stderr=None):
|
|
super().__init__(message)
|
|
self.command = command
|
|
self.stderr = stderr
|
|
|
|
def __str__(self):
|
|
# Provide a more informative string representation
|
|
base_message = super().__str__()
|
|
details = []
|
|
if self.command:
|
|
details.append(f"Command: '{' '.join(self.command)}'")
|
|
if self.stderr:
|
|
details.append(f"Stderr: {self.stderr.strip()}")
|
|
if details:
|
|
return f"{base_message} ({'; '.join(details)})"
|
|
else:
|
|
return base_message
|
|
|
|
|
|
class GitCommands:
|
|
"""
|
|
A class to manage Git commands execution, logging, and error handling.
|
|
This class is decoupled from the GUI and operates on provided directory paths.
|
|
"""
|
|
|
|
def __init__(self, logger):
|
|
"""
|
|
Initializes the GitCommands with a logger.
|
|
Args:
|
|
logger (logging.Logger): Logger instance for logging messages.
|
|
"""
|
|
if not isinstance(logger, logging.Logger):
|
|
raise ValueError("A valid logging.Logger instance is required.")
|
|
self.logger = logger
|
|
|
|
def log_and_execute(self, command, working_directory, check=True):
|
|
"""
|
|
Executes a command within a specific working directory, logs it, and handles errors.
|
|
|
|
Args:
|
|
command (list): List representing the command and its arguments.
|
|
working_directory (str): The path to the directory where the command should run.
|
|
check (bool, optional): If True, raises CalledProcessError wrapped in
|
|
GitCommandError if the command returns a non-zero exit code.
|
|
Defaults to True.
|
|
|
|
Returns:
|
|
subprocess.CompletedProcess: The result object from subprocess.run.
|
|
|
|
Raises:
|
|
GitCommandError: If working_directory is invalid, the command is not found,
|
|
fails (and check=True), or other execution errors occur.
|
|
ValueError: If working_directory is None or empty.
|
|
"""
|
|
log_message = f"Executing: {' '.join(command)}"
|
|
self.logger.debug(log_message) # Log command execution at debug level
|
|
|
|
# Validate working directory
|
|
if not working_directory:
|
|
msg = "Working directory cannot be None or empty."
|
|
self.logger.error(msg)
|
|
raise ValueError(msg)
|
|
|
|
abs_path = os.path.abspath(working_directory)
|
|
if not os.path.isdir(abs_path):
|
|
msg = f"Working directory does not exist or is not a directory: {abs_path}"
|
|
self.logger.error(msg)
|
|
raise GitCommandError(msg, command=command) # Include command context
|
|
|
|
cwd = abs_path
|
|
self.logger.debug(f"Working directory: {cwd}")
|
|
|
|
try:
|
|
# Platform-specific setup to hide console window on Windows
|
|
startupinfo = None
|
|
creationflags = 0
|
|
if os.name == 'nt':
|
|
startupinfo = subprocess.STARTUPINFO()
|
|
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
|
startupinfo.wShowWindow = subprocess.SW_HIDE
|
|
# Optional: Use CREATE_NO_WINDOW if you don't need stdin/stdout/stderr pipes
|
|
# creationflags = subprocess.CREATE_NO_WINDOW
|
|
|
|
# Execute the command
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=cwd,
|
|
capture_output=True, # Capture stdout and stderr
|
|
text=True, # Decode stdout/stderr as text
|
|
check=check, # Raise CalledProcessError if return code != 0 and check is True
|
|
encoding='utf-8', # Specify encoding
|
|
errors='replace', # Handle potential decoding errors
|
|
startupinfo=startupinfo, # Hide window on Windows
|
|
creationflags=creationflags # Additional flags if needed
|
|
)
|
|
|
|
# Log stdout and stderr clearly
|
|
stdout_log = result.stdout.strip() if result.stdout else "<no stdout>"
|
|
stderr_log = result.stderr.strip() if result.stderr else "<no stderr>"
|
|
|
|
# Log success with output
|
|
self.logger.info(f"Command successful. Output:\n--- stdout ---\n{stdout_log}\n--- stderr ---\n{stderr_log}\n---")
|
|
return result
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
# Log detailed error information from the failed process
|
|
stderr_err = e.stderr.strip() if e.stderr else "<no stderr>"
|
|
stdout_err = e.stdout.strip() if e.stdout else "<no stdout>"
|
|
error_msg = (
|
|
f"Command failed with return code {e.returncode} in '{cwd}'.\n"
|
|
f"--- command ---\n{' '.join(command)}\n"
|
|
f"--- stderr ---\n{stderr_err}\n"
|
|
f"--- stdout ---\n{stdout_err}\n"
|
|
f"---"
|
|
)
|
|
self.logger.error(error_msg)
|
|
# Wrap the original exception in GitCommandError for consistent handling
|
|
raise GitCommandError(f"Git command failed in '{cwd}'.", command=command, stderr=e.stderr) from e
|
|
|
|
except FileNotFoundError as e:
|
|
# Log error if the command itself (e.g., 'git') is not found
|
|
error_msg = f"Command not found: '{command[0]}'. Is Git installed and in system PATH? (Working directory: '{cwd}')"
|
|
self.logger.error(error_msg, exc_info=False) # Log basic message, no need for full stack trace here
|
|
self.logger.debug(f"FileNotFoundError details: {e}") # Log exception details at debug level
|
|
raise GitCommandError(error_msg, command=command) from e
|
|
|
|
except PermissionError as e:
|
|
# Handle errors related to execution permissions
|
|
error_msg = f"Permission denied executing command in '{cwd}'."
|
|
self.logger.error(error_msg, exc_info=False)
|
|
self.logger.debug(f"PermissionError details: {e}")
|
|
raise GitCommandError(error_msg, command=command, stderr=str(e)) from e
|
|
|
|
except Exception as e:
|
|
# Catch any other unexpected errors during execution
|
|
self.logger.exception(f"Unexpected error executing command in '{cwd}': {e}") # Log full trace
|
|
raise GitCommandError(f"Unexpected error during command execution in '{cwd}': {e}", command=command) from e
|
|
|
|
|
|
def create_git_bundle(self, working_directory, bundle_path):
|
|
"""
|
|
Creates a Git bundle file from the repository in working_directory.
|
|
|
|
Args:
|
|
working_directory (str): Path to the local Git repository.
|
|
bundle_path (str): Full path where the bundle file should be saved.
|
|
|
|
Raises:
|
|
GitCommandError: If the git bundle command fails or working_directory is invalid.
|
|
ValueError: If working_directory is None or empty.
|
|
"""
|
|
# Normalize bundle path for consistency across OS
|
|
normalized_bundle_path = os.path.normpath(bundle_path).replace("\\", "/")
|
|
command = ["git", "bundle", "create", normalized_bundle_path, "--all"]
|
|
self.logger.info(f"Attempting to create Git bundle: {normalized_bundle_path}")
|
|
|
|
try:
|
|
# Execute command, allow non-zero exit for specific warning
|
|
result = self.log_and_execute(command, working_directory, check=False)
|
|
|
|
# Check for non-fatal warnings vs actual errors
|
|
if result.returncode != 0:
|
|
if "Refusing to create empty bundle" in result.stderr:
|
|
# Log as a warning, not an error - bundle wasn't created, but it's not a failure state
|
|
self.logger.warning(f"Bundle creation skipped: Repository at '{working_directory}' has no commits to bundle.")
|
|
# Do NOT raise an error here, let the caller know via logs
|
|
else:
|
|
# An actual error occurred during bundle creation
|
|
error_msg = f"Git bundle command failed with return code {result.returncode}."
|
|
# log_and_execute already logged details, raise specific error here
|
|
raise GitCommandError(error_msg, command=command, stderr=result.stderr)
|
|
else:
|
|
self.logger.info(f"Git bundle created successfully at '{normalized_bundle_path}'.")
|
|
|
|
except (GitCommandError, ValueError) as e:
|
|
# Log context and re-raise errors from log_and_execute or validation
|
|
self.logger.error(f"Failed to create Git bundle for repo '{working_directory}'. Reason: {e}")
|
|
raise
|
|
except Exception as e:
|
|
self.logger.exception(f"Unexpected error during Git bundle creation for '{working_directory}': {e}")
|
|
raise GitCommandError(f"Unexpected error creating bundle: {e}", command=command) from e
|
|
|
|
|
|
def fetch_from_git_bundle(self, working_directory, bundle_path):
|
|
"""
|
|
Fetches changes from a Git bundle file into the specified local repository and merges them.
|
|
|
|
Args:
|
|
working_directory (str): Path to the local Git repository.
|
|
bundle_path (str): Path to the Git bundle file to fetch from.
|
|
|
|
Raises:
|
|
GitCommandError: If fetch or merge fails, or paths are invalid.
|
|
ValueError: If working_directory is None or empty.
|
|
"""
|
|
# Normalize bundle path
|
|
normalized_bundle_path = os.path.normpath(bundle_path).replace("\\", "/")
|
|
self.logger.info(f"Attempting to fetch from bundle '{normalized_bundle_path}' into '{working_directory}'")
|
|
|
|
# Define commands
|
|
fetch_command = ["git", "fetch", normalized_bundle_path]
|
|
# Merge the fetched head. Consider adding --no-edit or specifying a merge message.
|
|
merge_command = ["git", "merge", "FETCH_HEAD", "--no-commit", "--no-ff"] # Example: No auto commit, no fast-forward
|
|
|
|
try:
|
|
# 1. Fetch changes from the bundle
|
|
self.logger.debug(f"Executing fetch command...")
|
|
self.log_and_execute(fetch_command, working_directory, check=True) # Check=True ensures fetch succeeds
|
|
self.logger.info("Successfully fetched from Git bundle.")
|
|
|
|
# 2. Merge the fetched changes
|
|
self.logger.debug(f"Executing merge command...")
|
|
# Use check=False for merge, as conflicts are possible and return non-zero code
|
|
merge_result = self.log_and_execute(merge_command, working_directory, check=False)
|
|
|
|
# Analyze merge result
|
|
if merge_result.returncode == 0:
|
|
self.logger.info("Successfully merged fetched changes (or already up-to-date).")
|
|
# Optional: If using --no-commit, you might need a final commit step here or inform the user.
|
|
# If merge resulted in changes, perform the commit
|
|
if "Already up to date" not in merge_result.stdout:
|
|
self.logger.info("Merge successful, committing changes.")
|
|
commit_merge_command = ["git", "commit", "-m", f"Merge changes from bundle {os.path.basename(bundle_path)}"]
|
|
self.log_and_execute(commit_merge_command, working_directory, check=True)
|
|
else:
|
|
self.logger.info("Repository was already up-to-date with the bundle.")
|
|
|
|
else:
|
|
# Merge likely failed due to conflicts
|
|
stderr_log = merge_result.stderr.strip() if merge_result.stderr else ""
|
|
stdout_log = merge_result.stdout.strip() if merge_result.stdout else ""
|
|
if "conflict" in stderr_log.lower() or "conflict" in stdout_log.lower():
|
|
conflict_msg = f"Merge conflict occurred after fetching from bundle '{bundle_path}'. Please resolve conflicts manually in '{working_directory}' and commit."
|
|
self.logger.error(conflict_msg)
|
|
# Raise a specific error indicating conflict
|
|
raise GitCommandError(conflict_msg, command=merge_command, stderr=merge_result.stderr)
|
|
else:
|
|
# Other merge error
|
|
error_msg = f"Merge command failed unexpectedly after fetch (return code {merge_result.returncode})."
|
|
self.logger.error(error_msg)
|
|
# log_and_execute logged details, raise specific error
|
|
raise GitCommandError(error_msg, command=merge_command, stderr=merge_result.stderr)
|
|
|
|
except (GitCommandError, ValueError) as e:
|
|
# Log context and re-raise errors from log_and_execute, validation or merge conflict
|
|
self.logger.error(f"Error during fetch/merge process for repo '{working_directory}' from bundle '{bundle_path}'. Reason: {e}")
|
|
raise
|
|
except Exception as e:
|
|
self.logger.exception(f"Unexpected error during fetch/merge for '{working_directory}': {e}")
|
|
# Determine if fetch or merge command was being run if possible
|
|
raise GitCommandError(f"Unexpected error during fetch/merge: {e}") from e
|
|
|
|
|
|
def prepare_svn_for_git(self, working_directory):
|
|
"""
|
|
Prepares a directory for use with Git by initializing a repository (if needed)
|
|
and ensuring a .gitignore file exists ignoring '.svn'.
|
|
|
|
Args:
|
|
working_directory (str): The path to the directory to prepare.
|
|
|
|
Raises:
|
|
GitCommandError: If Git commands fail, file operations fail, or path is invalid.
|
|
ValueError: If working_directory is None or empty.
|
|
"""
|
|
self.logger.info(f"Preparing directory for Git: '{working_directory}'")
|
|
|
|
# Basic path validation (log_and_execute will do more thorough check)
|
|
if not working_directory:
|
|
raise ValueError("Working directory cannot be None or empty.")
|
|
if not os.path.isdir(working_directory):
|
|
msg = f"Directory does not exist or is not a directory: {working_directory}"
|
|
self.logger.error(msg)
|
|
raise GitCommandError(msg) # Raise Git error for consistency
|
|
|
|
# Define paths
|
|
gitignore_path = os.path.join(working_directory, ".gitignore")
|
|
git_dir_path = os.path.join(working_directory, ".git")
|
|
|
|
# 1. Initialize Git repository if it doesn't exist
|
|
if not os.path.exists(git_dir_path):
|
|
self.logger.info(f"No existing Git repository found in '{working_directory}'. Initializing...")
|
|
try:
|
|
init_command = ["git", "init"]
|
|
self.log_and_execute(init_command, working_directory, check=True)
|
|
self.logger.info(f"Git repository initialized successfully in '{working_directory}'.")
|
|
except (GitCommandError, ValueError) as e:
|
|
self.logger.error(f"Failed to initialize Git repository in '{working_directory}': {e}")
|
|
raise # Re-raise to signal preparation failure
|
|
except Exception as e:
|
|
self.logger.exception(f"Unexpected error initializing Git repository in '{working_directory}': {e}")
|
|
raise GitCommandError(f"Unexpected error during git init: {e}") from e
|
|
else:
|
|
self.logger.info(f"Git repository already exists in '{working_directory}'. Skipping initialization.")
|
|
|
|
# 2. Ensure .gitignore exists and ignores .svn
|
|
self.logger.debug(f"Checking/updating .gitignore file: {gitignore_path}")
|
|
try:
|
|
svn_ignore_entry = ".svn"
|
|
needs_update = False
|
|
if not os.path.exists(gitignore_path):
|
|
self.logger.info("'.gitignore' file not found. Creating...")
|
|
with open(gitignore_path, "w", encoding='utf-8') as f:
|
|
f.write(f"{svn_ignore_entry}\n")
|
|
self.logger.info(f"Created '.gitignore' and added '{svn_ignore_entry}' entry.")
|
|
else:
|
|
# Check if .svn is already ignored
|
|
try:
|
|
with open(gitignore_path, "r", encoding='utf-8') as f:
|
|
# Check if a line exactly matches ".svn" or starts with ".svn/"
|
|
is_ignored = any(line.strip() == svn_ignore_entry or line.strip().startswith(svn_ignore_entry + '/') for line in f)
|
|
if not is_ignored:
|
|
self.logger.info(f"'{svn_ignore_entry}' not found in existing '.gitignore'. Appending...")
|
|
needs_update = True
|
|
else:
|
|
self.logger.info(f"'{svn_ignore_entry}' entry already present in '.gitignore'.")
|
|
except IOError as e:
|
|
self.logger.warning(f"Could not read existing '.gitignore' to check for '{svn_ignore_entry}': {e}. Assuming update is needed.")
|
|
needs_update = True # Be safe and try to append
|
|
|
|
if needs_update:
|
|
try:
|
|
# Append '.svn' ensuring it's on a new line
|
|
with open(gitignore_path, "a", encoding='utf-8') as f:
|
|
# Check if file ends with newline before appending
|
|
f.seek(0, os.SEEK_END) # Go to end of file
|
|
if f.tell() > 0: # If file is not empty
|
|
f.seek(f.tell() - 1, os.SEEK_SET) # Go to last character
|
|
if f.read(1) != '\n':
|
|
f.write('\n') # Add newline if missing
|
|
f.write(f"{svn_ignore_entry}\n")
|
|
self.logger.info(f"Appended '{svn_ignore_entry}' entry to '.gitignore'.")
|
|
except IOError as e:
|
|
self.logger.error(f"Error appending '{svn_ignore_entry}' to '.gitignore' in '{working_directory}': {e}")
|
|
raise GitCommandError(f"Failed to update .gitignore: {e}") from e
|
|
|
|
except IOError as e:
|
|
self.logger.error(f"Error accessing or writing '.gitignore' file in '{working_directory}': {e}")
|
|
raise GitCommandError(f"File I/O error for .gitignore: {e}") from e
|
|
except Exception as e:
|
|
self.logger.exception(f"Unexpected error managing '.gitignore' in '{working_directory}': {e}")
|
|
raise GitCommandError(f"Unexpected error with .gitignore: {e}") from e
|
|
|
|
self.logger.info(f"Directory preparation complete for '{working_directory}'.")
|
|
|
|
|
|
def git_commit(self, working_directory, message="Autocommit"):
|
|
"""
|
|
Stages all changes (git add .) and commits them in the specified repository.
|
|
|
|
Args:
|
|
working_directory (str): Path to the local Git repository.
|
|
message (str): The commit message. Defaults to "Autocommit".
|
|
|
|
Returns:
|
|
bool: True if a commit was successfully made, False if there were no changes to commit.
|
|
|
|
Raises:
|
|
GitCommandError: If staging or committing fails unexpectedly.
|
|
ValueError: If working_directory is None or empty.
|
|
"""
|
|
self.logger.info(f"Attempting to stage and commit changes in '{working_directory}' with message: '{message}'")
|
|
|
|
try:
|
|
# 1. Stage all changes
|
|
add_command = ["git", "add", "."]
|
|
self.logger.debug("Staging all changes (git add .)...")
|
|
self.log_and_execute(add_command, working_directory, check=True) # Check=True ensures staging works
|
|
self.logger.debug("Staging successful.")
|
|
|
|
# 2. Commit staged changes
|
|
commit_command = ["git", "commit", "-m", message]
|
|
self.logger.debug("Attempting commit...")
|
|
# Use check=False because 'nothing to commit' returns non-zero exit code
|
|
result = self.log_and_execute(commit_command, working_directory, check=False)
|
|
|
|
# Analyze commit result
|
|
stdout_lower = result.stdout.lower() if result.stdout else ""
|
|
stderr_lower = result.stderr.lower() if result.stderr else ""
|
|
|
|
if result.returncode == 0:
|
|
# Commit was successful and created a new commit
|
|
self.logger.info(f"Commit successful in '{working_directory}'.")
|
|
return True
|
|
elif "nothing to commit" in stdout_lower or \
|
|
"no changes added to commit" in stdout_lower or \
|
|
"nothing added to commit" in stdout_lower or \
|
|
(result.returncode == 1 and not stderr_lower): # Some git versions return 1 with no stderr for no changes
|
|
# No changes were detected to commit
|
|
self.logger.info("No changes to commit.")
|
|
return False
|
|
else:
|
|
# An unexpected error occurred during the commit attempt
|
|
error_msg = f"Commit command failed unexpectedly (return code {result.returncode})."
|
|
# log_and_execute already logged details, raise specific error
|
|
raise GitCommandError(error_msg, command=commit_command, stderr=result.stderr)
|
|
|
|
except (GitCommandError, ValueError) as e:
|
|
self.logger.error(f"Error during staging or commit process in '{working_directory}': {e}")
|
|
raise # Re-raise the caught specific error
|
|
except Exception as e:
|
|
self.logger.exception(f"Unexpected error during staging/commit in '{working_directory}': {e}")
|
|
raise GitCommandError(f"Unexpected error during commit: {e}") from e
|
|
|
|
|
|
def git_status_has_changes(self, working_directory):
|
|
"""
|
|
Checks if the Git repository in working_directory has uncommitted changes
|
|
(unstaged or staged).
|
|
|
|
Args:
|
|
working_directory (str): Path to the local Git repository.
|
|
|
|
Returns:
|
|
bool: True if there are any changes, False otherwise.
|
|
|
|
Raises:
|
|
GitCommandError: If the git status command fails.
|
|
ValueError: If working_directory is None or empty.
|
|
"""
|
|
self.logger.debug(f"Checking Git status for changes in '{working_directory}'...")
|
|
|
|
try:
|
|
# Use '--porcelain' for script-friendly output. Empty output means no changes.
|
|
status_command = ["git", "status", "--porcelain"]
|
|
result = self.log_and_execute(status_command, working_directory, check=True) # Check=True ensures status works
|
|
|
|
# Check if the porcelain output is empty or not
|
|
has_changes = bool(result.stdout.strip())
|
|
self.logger.debug(f"Status check complete. Has changes: {has_changes}")
|
|
return has_changes
|
|
|
|
except (GitCommandError, ValueError) as e:
|
|
self.logger.error(f"Error checking Git status in '{working_directory}': {e}")
|
|
raise # Re-raise the specific error
|
|
except Exception as e:
|
|
self.logger.exception(f"Unexpected error checking Git status in '{working_directory}': {e}")
|
|
raise GitCommandError(f"Unexpected error checking status: {e}") from e |