677 lines
27 KiB
Python
677 lines
27 KiB
Python
# projectutility/core/git_manager.py
|
|
|
|
import os
|
|
import sys # Needed for path calculation
|
|
import logging
|
|
import shutil
|
|
from typing import Optional, Dict, Any, Tuple, Callable
|
|
|
|
# --- Constants ---
|
|
# Git Status Strings (used for reporting status consistently)
|
|
GIT_STATUS_UP_TO_DATE = "Up-to-date"
|
|
GIT_STATUS_BEHIND = "Behind"
|
|
GIT_STATUS_AHEAD = "Ahead"
|
|
GIT_STATUS_DIVERGED = "Diverged"
|
|
GIT_STATUS_NOT_CLONED = "Not Cloned"
|
|
GIT_STATUS_ERROR = "Error" # Generic Git or manager error
|
|
GIT_STATUS_CHECKING = "Checking..." # UI transient state
|
|
GIT_STATUS_UPDATING = "Updating..." # UI transient state
|
|
GIT_STATUS_GITPYTHON_MISSING = "GitPython Missing" # Specific error if library unavailable
|
|
|
|
# --- Optional GitPython Import --- (unchanged)
|
|
try:
|
|
import git
|
|
from git.exc import GitCommandError, InvalidGitRepositoryError, NoSuchPathError
|
|
from git import RemoteProgress
|
|
|
|
GITPYTHON_AVAILABLE = True
|
|
logger = logging.getLogger(__name__)
|
|
logger.info("GitPython library loaded successfully.")
|
|
except ImportError:
|
|
GITPYTHON_AVAILABLE = False
|
|
logger = logging.getLogger(__name__)
|
|
logger.warning(
|
|
"GitPython library not found. Git-related features will be disabled. "
|
|
"Install it using: pip install GitPython"
|
|
)
|
|
|
|
class GitCommandError(Exception):
|
|
pass
|
|
|
|
class InvalidGitRepositoryError(Exception):
|
|
pass
|
|
|
|
class NoSuchPathError(Exception):
|
|
pass
|
|
|
|
class RemoteProgress:
|
|
pass
|
|
|
|
class Repo:
|
|
pass # Dummy
|
|
|
|
# ... (altri dummy se necessario)
|
|
|
|
# --- Internal Model Import --- (unchanged)
|
|
try:
|
|
from projectutility.core.registry_models import ToolRegistryEntry
|
|
except ImportError:
|
|
logger.critical(
|
|
"Failed to import .registry_models. GitManager functions may fail!",
|
|
exc_info=True,
|
|
)
|
|
from dataclasses import dataclass
|
|
|
|
@dataclass
|
|
class ToolRegistryEntry:
|
|
id: str = "dummy_import_failed"
|
|
type: str = "unknown"
|
|
git_url: Optional[str] = None
|
|
git_ref: str = "main"
|
|
local_dir_name: Optional[str] = None # Aggiunto per coerenza con l'uso
|
|
display_name: str = "Dummy" # Aggiunto per coerenza
|
|
|
|
|
|
# --- Constants --- (unchanged)
|
|
GIT_STATUS_UP_TO_DATE = "Up-to-date"
|
|
GIT_STATUS_BEHIND = "Behind"
|
|
# ... (altri status)
|
|
GIT_STATUS_GITPYTHON_MISSING = "GitPython Missing"
|
|
|
|
# --- Determine if Running as Frozen Executable ---
|
|
IS_FROZEN = getattr(sys, "frozen", False)
|
|
|
|
# --- Calculate Base Paths ---
|
|
# APP_ROOT_DIR is the base for MANAGED_TOOLS_DIR.
|
|
# In frozen mode, it's the directory containing the executable.
|
|
# In script mode, it's the repository root.
|
|
_app_root_dir_calculated: str
|
|
try:
|
|
if IS_FROZEN:
|
|
_app_root_dir_calculated = os.path.dirname(sys.executable)
|
|
logger.debug(
|
|
f"Git Manager (FROZEN): Executable directory (used as App Root): {_app_root_dir_calculated}"
|
|
)
|
|
else:
|
|
_current_file_path = os.path.abspath(__file__) # .../core/git_manager.py
|
|
_core_dir = os.path.dirname(_current_file_path) # .../core
|
|
_app_source_root = os.path.dirname(_core_dir) # .../projectutility (package)
|
|
_app_root_dir_calculated = os.path.dirname(_app_source_root) # Repository Root
|
|
logger.debug(
|
|
f"Git Manager (SCRIPT): Repository root: {_app_root_dir_calculated}"
|
|
)
|
|
|
|
APP_ROOT_DIR = _app_root_dir_calculated
|
|
MANAGED_TOOLS_DIR = os.path.join(APP_ROOT_DIR, "managed_tools")
|
|
|
|
logger.debug(
|
|
f"Git Manager: Final APP_ROOT_DIR (Repo Root or Exe Dir): {APP_ROOT_DIR}"
|
|
)
|
|
logger.debug(
|
|
f"Git Manager: Final Managed tools directory (MANAGED_TOOLS_DIR): {MANAGED_TOOLS_DIR}"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.critical(
|
|
f"Failed to calculate base paths in git_manager: {e}. "
|
|
f"Managed tools path will be incorrect.",
|
|
exc_info=True,
|
|
)
|
|
APP_ROOT_DIR = os.getcwd() # Fallback
|
|
MANAGED_TOOLS_DIR = os.path.join(APP_ROOT_DIR, "managed_tools") # Fallback
|
|
logger.warning(f"Using fallback managed tools path: {MANAGED_TOOLS_DIR}")
|
|
|
|
|
|
# --- Progress Reporting Class --- (unchanged)
|
|
class CloneProgressReporter(RemoteProgress):
|
|
def __init__(
|
|
self,
|
|
tool_id: str,
|
|
progress_callback: Optional[Callable[[str, str], None]] = None,
|
|
):
|
|
super().__init__()
|
|
self.tool_id = tool_id
|
|
self.progress_callback = progress_callback
|
|
self._last_message = ""
|
|
|
|
def update(self, op_code, cur_count, max_count=None, message=""):
|
|
if self.progress_callback and message:
|
|
progress_text = f"[{self.tool_id}] Git: {message.strip()}"
|
|
try:
|
|
self.progress_callback("git_progress", progress_text)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error occurred within the progress callback "
|
|
f"function for tool '{self.tool_id}': {e}",
|
|
exc_info=True,
|
|
)
|
|
|
|
|
|
# --- Helper Functions --- (get_local_repo_path usa MANAGED_TOOLS_DIR che ora è frozen-aware)
|
|
# (Le funzioni _get_repo_object e _fetch_origin rimangono invariate nella logica interna)
|
|
|
|
|
|
def get_local_repo_path(tool_id: str, local_dir_name: Optional[str] = None) -> str:
|
|
dir_name = local_dir_name if local_dir_name else tool_id
|
|
absolute_path = os.path.abspath(
|
|
os.path.join(MANAGED_TOOLS_DIR, dir_name)
|
|
) # MANAGED_TOOLS_DIR è ora corretto
|
|
# logger.debug(f"Resolved local repo path for '{tool_id}': {absolute_path}") # Già loggato
|
|
return absolute_path
|
|
|
|
|
|
def _get_repo_object(
|
|
local_path: str,
|
|
) -> Optional["Repo"]: # "Repo" tra virgolette per forward declaration
|
|
if not GITPYTHON_AVAILABLE:
|
|
# logger.error("Cannot get Repo object: GitPython library is not available.") # Log ripetitivo
|
|
return None
|
|
if not os.path.isdir(local_path):
|
|
# logger.debug(f"Directory does not exist, cannot be a repo: {local_path}")
|
|
return None
|
|
try:
|
|
repo = git.Repo(local_path)
|
|
# logger.debug(f"Successfully accessed Git repository object at: {local_path}")
|
|
return repo
|
|
except InvalidGitRepositoryError:
|
|
logger.warning(
|
|
f"Directory exists but is not a valid Git repository: {local_path}"
|
|
)
|
|
return None
|
|
except NoSuchPathError:
|
|
logger.error(
|
|
f"GitPython reported 'NoSuchPathError' for existing (?) directory: {local_path}"
|
|
)
|
|
return None
|
|
except Exception as e:
|
|
logger.exception(f"Unexpected error accessing repository at {local_path}: {e}")
|
|
return None
|
|
|
|
|
|
def _fetch_origin(
|
|
repo: "Repo", # "Repo" tra virgolette
|
|
tool_id: str,
|
|
progress_callback: Optional[Callable[[str, str], None]] = None,
|
|
) -> bool:
|
|
if not GITPYTHON_AVAILABLE:
|
|
# logger.error(f"[{tool_id}] Cannot fetch: GitPython is not available.") # Log ripetitivo
|
|
return False
|
|
logger.info(f"[{tool_id}] Attempting to fetch updates from 'origin'...")
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_status", f"[{tool_id}] Fetching from remote...")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (pre-fetch): {cb_err}", exc_info=True
|
|
)
|
|
try:
|
|
if not repo.remotes or "origin" not in [r.name for r in repo.remotes]:
|
|
logger.error(
|
|
f"[{tool_id}] Cannot fetch: Remote 'origin' not found or no remotes configured."
|
|
)
|
|
return False
|
|
origin = repo.remotes.origin
|
|
progress_reporter = (
|
|
CloneProgressReporter(tool_id, progress_callback)
|
|
if progress_callback
|
|
else None
|
|
)
|
|
fetch_info_list = origin.fetch(prune=True, progress=progress_reporter)
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
for info in fetch_info_list:
|
|
logger.debug(
|
|
f"[{tool_id}] Fetch info: {info.name}, {info.flags}, {info.note}"
|
|
)
|
|
logger.info(f"[{tool_id}] Fetch operation completed successfully.")
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_status", f"[{tool_id}] Fetch complete.")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (post-fetch): {cb_err}", exc_info=True
|
|
)
|
|
return True
|
|
except GitCommandError as e:
|
|
error_output = e.stderr.strip() if e.stderr else str(e)
|
|
logger.error(
|
|
f"[{tool_id}] Git fetch command failed. Error: {error_output}",
|
|
exc_info=True,
|
|
)
|
|
if progress_callback:
|
|
try:
|
|
progress_callback(
|
|
"git_error", f"[{tool_id}] Fetch failed: {error_output}"
|
|
)
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (fetch error): {cb_err}", exc_info=True
|
|
)
|
|
return False
|
|
except AttributeError: # E.g., repo.remotes.origin non esiste
|
|
logger.error(f"[{tool_id}] Failed to access remote 'origin' during fetch.")
|
|
return False
|
|
except Exception as e:
|
|
logger.exception(f"[{tool_id}] An unexpected error occurred during fetch: {e}")
|
|
if progress_callback:
|
|
try:
|
|
progress_callback(
|
|
"git_error", f"[{tool_id}] Unexpected fetch error: {e}"
|
|
)
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (fetch unexpected error): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
return False
|
|
|
|
|
|
# --- Core Git Operations ---
|
|
# (ensure_repository_cloned, get_repository_status, _get_local_ref_name, update_repository)
|
|
# La logica interna di queste funzioni rimane invariata, ma beneficiano dei percorsi corretti.
|
|
|
|
|
|
def ensure_repository_cloned(
|
|
tool_entry: ToolRegistryEntry,
|
|
progress_callback: Optional[Callable[[str, str], None]] = None,
|
|
) -> Tuple[Optional["Repo"], str]: # "Repo"
|
|
if not GITPYTHON_AVAILABLE:
|
|
return None, GIT_STATUS_GITPYTHON_MISSING
|
|
if tool_entry.type != "git" or not tool_entry.git_url:
|
|
return None, "Tool not configured for Git or missing URL."
|
|
|
|
tool_id = tool_entry.id
|
|
local_path = get_local_repo_path(tool_id, tool_entry.local_dir_name)
|
|
repo = _get_repo_object(local_path)
|
|
if repo:
|
|
return repo, "Repository exists locally."
|
|
|
|
logger.info(
|
|
f"[{tool_id}] Repository not found. Cloning from '{tool_entry.git_url}' into '{local_path}'..."
|
|
)
|
|
if progress_callback:
|
|
try:
|
|
# Usa tool_entry.display_name per un messaggio più user-friendly
|
|
progress_callback(
|
|
"git_status",
|
|
f"[{tool_id}] Cloning {tool_entry.display_name or tool_id}...",
|
|
)
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (pre-clone): {cb_err}", exc_info=True
|
|
)
|
|
try:
|
|
os.makedirs(
|
|
MANAGED_TOOLS_DIR, exist_ok=True
|
|
) # MANAGED_TOOLS_DIR è ora corretto
|
|
except OSError as e:
|
|
logger.error(
|
|
f"[{tool_id}] Failed to create parent directory '{MANAGED_TOOLS_DIR}': {e}.",
|
|
exc_info=True,
|
|
)
|
|
return None, f"Failed to create base directory: {e}"
|
|
|
|
progress_reporter = (
|
|
CloneProgressReporter(tool_id, progress_callback) if progress_callback else None
|
|
)
|
|
cloned_repo: Optional["Repo"] = None # "Repo"
|
|
clone_failed_msg = "Clone operation failed." # Default
|
|
try:
|
|
cloned_repo = git.Repo.clone_from(
|
|
url=tool_entry.git_url, to_path=local_path, progress=progress_reporter
|
|
)
|
|
logger.info(f"[{tool_id}] Repository cloned successfully to {local_path}.")
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_status", f"[{tool_id}] Clone successful.")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (post-clone success): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
return cloned_repo, "Repository cloned successfully."
|
|
except GitCommandError as e:
|
|
error_output = e.stderr.strip() if e.stderr else str(e)
|
|
logger.error(
|
|
f"[{tool_id}] Failed to clone from '{tool_entry.git_url}'. Error: {error_output}",
|
|
exc_info=True,
|
|
)
|
|
clone_failed_msg = f"Failed to clone: {error_output}"
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_error", f"[{tool_id}] {clone_failed_msg}")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (clone GitCommandError): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
except Exception as e:
|
|
logger.exception(f"[{tool_id}] An unexpected error occurred during clone: {e}")
|
|
clone_failed_msg = f"Unexpected clone error: {e}"
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_error", f"[{tool_id}] {clone_failed_msg}")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (clone unexpected error): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
|
|
if os.path.exists(local_path): # Cleanup on failure
|
|
logger.warning(
|
|
f"[{tool_id}] Cleaning up incomplete clone directory: {local_path}"
|
|
)
|
|
try:
|
|
shutil.rmtree(local_path)
|
|
logger.info(f"[{tool_id}] Successfully removed incomplete clone directory.")
|
|
except OSError as rm_e:
|
|
logger.error(
|
|
f"[{tool_id}] Failed to remove incomplete clone directory '{local_path}': {rm_e}",
|
|
exc_info=True,
|
|
)
|
|
except Exception as rm_e: # Catch broader exceptions
|
|
logger.exception(
|
|
f"[{tool_id}] Unexpected error removing directory '{local_path}': {rm_e}"
|
|
)
|
|
return None, clone_failed_msg
|
|
|
|
|
|
def get_repository_status(tool_entry: ToolRegistryEntry) -> Dict[str, Any]:
|
|
status_result: Dict[str, Any] = {
|
|
"status": GIT_STATUS_ERROR,
|
|
"local_hash": None,
|
|
"remote_hash": None,
|
|
"local_ref": None,
|
|
"remote_ref": None,
|
|
"message": "Init error.",
|
|
}
|
|
if not GITPYTHON_AVAILABLE:
|
|
status_result.update(
|
|
{
|
|
"status": GIT_STATUS_GITPYTHON_MISSING,
|
|
"message": "GitPython not installed.",
|
|
}
|
|
)
|
|
return status_result
|
|
if tool_entry.type != "git" or not tool_entry.git_url:
|
|
status_result.update(
|
|
{"status": GIT_STATUS_ERROR, "message": "Not Git tool or no URL."}
|
|
)
|
|
return status_result
|
|
|
|
tool_id = tool_entry.id
|
|
local_path = get_local_repo_path(tool_id, tool_entry.local_dir_name)
|
|
repo = _get_repo_object(local_path)
|
|
if not repo:
|
|
status_result.update(
|
|
{"status": GIT_STATUS_NOT_CLONED, "message": "Not cloned."}
|
|
)
|
|
return status_result
|
|
|
|
logger.info(f"[{tool_id}] Checking repo status: {local_path}...")
|
|
if not _fetch_origin(repo, tool_id, progress_callback=None):
|
|
status_result.update({"status": GIT_STATUS_ERROR, "message": "Fetch failed."})
|
|
try: # Try to get local info even if fetch fails
|
|
status_result["local_hash"] = repo.head.commit.hexsha
|
|
status_result["local_ref"] = _get_local_ref_name(repo)
|
|
except Exception:
|
|
pass
|
|
return status_result
|
|
|
|
try:
|
|
local_commit: "Commit" = repo.head.commit # "Commit"
|
|
status_result["local_hash"] = local_commit.hexsha
|
|
status_result["local_ref"] = _get_local_ref_name(repo)
|
|
remote_ref_name = f"origin/{tool_entry.git_ref}"
|
|
status_result["remote_ref"] = remote_ref_name
|
|
remote_commit: Optional["Commit"] = None # "Commit"
|
|
try:
|
|
remote_commit = repo.commit(remote_ref_name)
|
|
status_result["remote_hash"] = remote_commit.hexsha
|
|
except (git.BadName, GitCommandError) as e:
|
|
logger.warning(f"[{tool_id}] Remote ref '{remote_ref_name}' not found: {e}")
|
|
status_result.update(
|
|
{
|
|
"status": GIT_STATUS_ERROR,
|
|
"remote_hash": None,
|
|
"message": f"Remote ref '{remote_ref_name}' not found.",
|
|
}
|
|
)
|
|
return status_result
|
|
|
|
if local_commit == remote_commit:
|
|
status_result.update(
|
|
{"status": GIT_STATUS_UP_TO_DATE, "message": "Up-to-date."}
|
|
)
|
|
elif repo.is_ancestor(local_commit.hexsha, remote_commit.hexsha):
|
|
status_result.update(
|
|
{"status": GIT_STATUS_BEHIND, "message": "Behind remote."}
|
|
)
|
|
elif repo.is_ancestor(remote_commit.hexsha, local_commit.hexsha):
|
|
status_result.update(
|
|
{"status": GIT_STATUS_AHEAD, "message": "Ahead of remote."}
|
|
)
|
|
else:
|
|
status_result.update(
|
|
{"status": GIT_STATUS_DIVERGED, "message": "Diverged from remote."}
|
|
)
|
|
logger.info(f"[{tool_id}] Status: {status_result['status']}")
|
|
except GitCommandError as e:
|
|
error_output = e.stderr.strip() if e.stderr else str(e)
|
|
logger.error(
|
|
f"[{tool_id}] Git error during status check: {error_output}", exc_info=True
|
|
)
|
|
status_result.update(
|
|
{"status": GIT_STATUS_ERROR, "message": f"Git error: {error_output}"}
|
|
)
|
|
except Exception as e:
|
|
logger.exception(f"[{tool_id}] Unexpected error during status check: {e}")
|
|
status_result.update(
|
|
{"status": GIT_STATUS_ERROR, "message": f"Unexpected error: {e}"}
|
|
)
|
|
return status_result
|
|
|
|
|
|
def _get_local_ref_name(repo: "Repo") -> str: # "Repo"
|
|
if not GITPYTHON_AVAILABLE:
|
|
return "N/A"
|
|
try:
|
|
head = repo.head
|
|
if head.is_detached:
|
|
return f"Detached@{head.commit.hexsha[:7]}"
|
|
else:
|
|
return head.ref.name
|
|
except Exception as e:
|
|
logger.warning(f"Could not determine local ref name: {e}")
|
|
return "Unknown"
|
|
|
|
|
|
def update_repository(
|
|
tool_entry: ToolRegistryEntry,
|
|
progress_callback: Optional[Callable[[str, str], None]] = None,
|
|
) -> Tuple[bool, str]:
|
|
if not GITPYTHON_AVAILABLE:
|
|
return False, GIT_STATUS_GITPYTHON_MISSING
|
|
if tool_entry.type != "git" or not tool_entry.git_url:
|
|
return False, "Not Git tool or no URL."
|
|
|
|
tool_id = tool_entry.id
|
|
logger.info(f"[{tool_id}] Starting repo update...")
|
|
repo, clone_msg = ensure_repository_cloned(tool_entry, progress_callback)
|
|
if not repo:
|
|
logger.error(f"[{tool_id}] Update failed (clone): {clone_msg}")
|
|
return False, f"Update failed (clone): {clone_msg}"
|
|
if not _fetch_origin(repo, tool_id, progress_callback):
|
|
logger.error(f"[{tool_id}] Update failed (fetch).")
|
|
return False, "Update failed (fetch from remote)."
|
|
|
|
target_ref = tool_entry.git_ref
|
|
remote_target_ref_name = f"origin/{target_ref}"
|
|
logger.info(f"[{tool_id}] Checking out target ref: '{target_ref}'")
|
|
if progress_callback:
|
|
try:
|
|
progress_callback(
|
|
"git_status", f"[{tool_id}] Checking out '{target_ref}'..."
|
|
)
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (pre-checkout): {cb_err}", exc_info=True
|
|
)
|
|
try:
|
|
target_commit_on_remote: Optional["Commit"] = None # "Commit"
|
|
try:
|
|
target_commit_on_remote = repo.commit(remote_target_ref_name)
|
|
logger.debug(
|
|
f"[{tool_id}] Verified remote ref '{remote_target_ref_name}' (Commit: {target_commit_on_remote.hexsha[:7]})"
|
|
)
|
|
except (git.BadName, GitCommandError):
|
|
logger.error(
|
|
f"[{tool_id}] Update failed: Ref '{target_ref}' (as '{remote_target_ref_name}') not on remote 'origin'."
|
|
)
|
|
return False, f"Update failed: Ref '{target_ref}' not on remote 'origin'."
|
|
|
|
if repo.head.commit.hexsha == target_commit_on_remote.hexsha:
|
|
logger.info(
|
|
f"[{tool_id}] Already at target commit for '{target_ref}'. Checking out to ensure correct state."
|
|
)
|
|
repo.git.checkout(target_ref) # Checkout the ref (branch, tag, commit)
|
|
logger.info(f"[{tool_id}] Checked out '{target_ref}'.")
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_status", f"[{tool_id}] Checkout complete.")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (post-checkout): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
|
|
if not repo.head.is_detached:
|
|
current_branch_name = repo.head.ref.name
|
|
logger.info(
|
|
f"[{tool_id}] On branch '{current_branch_name}'. Attempting ff-only pull..."
|
|
)
|
|
if progress_callback:
|
|
try:
|
|
progress_callback(
|
|
"git_status",
|
|
f"[{tool_id}] Pulling '{current_branch_name}' (ff-only)...",
|
|
)
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (pre-pull): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
try:
|
|
repo.git.pull("origin", current_branch_name, "--ff-only")
|
|
logger.info(
|
|
f"[{tool_id}] FF-only pull successful for '{current_branch_name}'."
|
|
)
|
|
if progress_callback:
|
|
try:
|
|
progress_callback("git_status", f"[{tool_id}] Pull successful.")
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (post-pull success): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
|
|
# --- Submodule handling (if configured) ---
|
|
try:
|
|
if getattr(tool_entry, "auto_update_submodules", False) or getattr(
|
|
tool_entry, "submodules", None
|
|
):
|
|
local_path = get_local_repo_path(tool_id, tool_entry.local_dir_name)
|
|
try:
|
|
from projectutility.core.submodule_manager import ensure_submodules
|
|
|
|
ok = ensure_submodules(local_path, tool_entry.submodules)
|
|
if not ok:
|
|
logger.warning(
|
|
f"[{tool_id}] Submodule update returned non-ok status."
|
|
)
|
|
except Exception as e:
|
|
logger.exception(
|
|
f"[{tool_id}] Failed to run submodule update: {e}"
|
|
)
|
|
except Exception as e:
|
|
logger.exception(f"[{tool_id}] Error checking submodule config: {e}")
|
|
|
|
return (
|
|
True,
|
|
f"Repo updated. On branch '{current_branch_name}' tracking '{target_ref}'.",
|
|
)
|
|
except GitCommandError as pull_err:
|
|
error_output = (
|
|
pull_err.stderr.strip() if pull_err.stderr else str(pull_err)
|
|
)
|
|
logger.warning(
|
|
f"[{tool_id}] FF-only pull failed for '{current_branch_name}'. Git: {error_output}"
|
|
)
|
|
status_now = get_repository_status(tool_entry) # Re-check status
|
|
if status_now["status"] == GIT_STATUS_UP_TO_DATE:
|
|
logger.info(
|
|
f"[{tool_id}] Repo up-to-date despite ff-pull fail. Update OK."
|
|
)
|
|
if progress_callback:
|
|
try:
|
|
progress_callback(
|
|
"git_status",
|
|
f"[{tool_id}] Pull failed (ff-only), but repo is Up-to-date.",
|
|
)
|
|
except Exception as cb_err:
|
|
logger.error(
|
|
f"Error in progress callback (pull fail but ok): {cb_err}",
|
|
exc_info=True,
|
|
)
|
|
return (
|
|
True,
|
|
f"Repo up-to-date at '{target_ref}' (ff-pull failed/unnecessary).",
|
|
)
|
|
else:
|
|
logger.error(
|
|
f"[{tool_id}] Update failed: FF-pull failed for '{current_branch_name}', not up-to-date (Status: {status_now['status']}). Error: {error_output}"
|
|
)
|
|
return (
|
|
False,
|
|
f"Update failed: Pull error on '{current_branch_name}' ({status_now['status']}). Check logs.",
|
|
)
|
|
else: # Detached HEAD
|
|
detached_commit_sha = repo.head.commit.hexsha[:7]
|
|
logger.info(
|
|
f"[{tool_id}] Checkout to detached HEAD at {detached_commit_sha} ('{target_ref}'). Update complete."
|
|
)
|
|
# Try submodule update for detached HEAD too
|
|
try:
|
|
if getattr(tool_entry, "auto_update_submodules", False) or getattr(
|
|
tool_entry, "submodules", None
|
|
):
|
|
local_path = get_local_repo_path(tool_id, tool_entry.local_dir_name)
|
|
try:
|
|
from projectutility.core.submodule_manager import ensure_submodules
|
|
|
|
ok = ensure_submodules(local_path, tool_entry.submodules)
|
|
if not ok:
|
|
logger.warning(
|
|
f"[{tool_id}] Submodule update returned non-ok status (detached HEAD)."
|
|
)
|
|
except Exception as e:
|
|
logger.exception(
|
|
f"[{tool_id}] Failed to run submodule update (detached HEAD): {e}"
|
|
)
|
|
except Exception:
|
|
logger.exception(f"[{tool_id}] Error checking submodule config (detached HEAD)")
|
|
|
|
return (
|
|
True,
|
|
f"Repo at '{target_ref}' (Detached HEAD {detached_commit_sha}).",
|
|
)
|
|
except GitCommandError as e:
|
|
error_output = e.stderr.strip() if e.stderr else str(e)
|
|
logger.error(
|
|
f"[{tool_id}] Git error updating to '{target_ref}'. Code: {e.status}. Output: {error_output}",
|
|
exc_info=True,
|
|
)
|
|
return False, f"Update failed: Git error (Code: {e.status}). Check logs."
|
|
except Exception as e:
|
|
logger.exception(f"[{tool_id}] Unexpected error during repo update: {e}")
|
|
return False, f"Update failed: Unexpected error - {e}"
|