Deskaid

#!/usr/bin/env python3 import logging import os import re import subprocess from .shell import run_command __all__ = [ "get_head_commit_message", "get_head_commit_hash", "get_head_commit_chat_id", "get_repository_root", "is_git_repository", "get_ref_commit_chat_id", ] log = logging.getLogger(__name__) async def get_head_commit_message(directory: str) -> str | None: """Get the full commit message from HEAD. Args: directory: The directory to check Returns: The commit message if available, None otherwise """ try: # Check if HEAD exists result = await run_command( ["git", "rev-parse", "--verify", "HEAD"], cwd=directory, check=False, capture_output=True, text=True, ) if result.returncode != 0: # No commits yet return None # Get the commit message result = await run_command( ["git", "log", "-1", "--pretty=%B"], cwd=directory, check=True, capture_output=True, text=True, ) return result.stdout.strip() except Exception as e: logging.warning( f"Exception when getting HEAD commit message: {e!s}", exc_info=True ) return None async def get_head_commit_hash(directory: str, short: bool = True) -> str | None: """Get the commit hash from HEAD. Args: directory: The directory to check short: Whether to get short hash (default) or full hash Returns: The commit hash if available, None otherwise """ try: # Check if HEAD exists result = await run_command( ["git", "rev-parse", "--verify", "HEAD"], cwd=directory, check=False, capture_output=True, text=True, ) if result.returncode != 0: # No commits yet return None # Get the commit hash (short or full) cmd = ["git", "rev-parse"] if short: cmd.append("--short") cmd.append("HEAD") result = await run_command( cmd, cwd=directory, check=True, capture_output=True, text=True, ) return result.stdout.strip() except Exception as e: logging.warning( f"Exception when getting HEAD commit hash: {e!s}", exc_info=True ) return None async def get_head_commit_chat_id(directory: str) -> str | None: """Get the chat ID from the HEAD commit's message. Args: directory: The directory to check Returns: The chat ID if found, None otherwise """ try: commit_message = await get_head_commit_message(directory) if not commit_message: return None # Use regex to find the last occurrence of codemcp-id: XXX # The pattern looks for "codemcp-id: " followed by any characters up to a newline or end of string matches = re.findall(r"codemcp-id:\s*([a-zA-Z0-9-]+)", commit_message) # Return the last match if any matches found if matches: return matches[-1].strip() return None except Exception as e: logging.warning( f"Exception when getting HEAD commit chat ID: {e!s}", exc_info=True ) return None async def get_repository_root(path: str) -> str: """Get the root directory of the Git repository containing the path. Args: path: The file path to get the repository root for Returns: The absolute path to the repository root Raises: ValueError: If the path is not in a Git repository """ try: # Get the directory containing the file or use the path itself if it's a directory directory = os.path.dirname(path) if os.path.isfile(path) else path # Get the absolute path to ensure consistency directory = os.path.abspath(directory) # Get the repository root result = await run_command( ["git", "rev-parse", "--show-toplevel"], cwd=directory, check=True, capture_output=True, text=True, ) return result.stdout.strip() except (subprocess.SubprocessError, OSError) as e: raise ValueError(f"Path is not in a git repository: {str(e)}") async def is_git_repository(path: str) -> bool: """Check if the path is within a Git repository. Args: path: The file path to check Returns: True if path is in a Git repository, False otherwise """ try: # Get the directory containing the file or use the path itself if it's a directory directory = os.path.dirname(path) if os.path.isfile(path) else path # Get the absolute path to ensure consistency directory = os.path.abspath(directory) # Run git command to verify this is a git repository await run_command( ["git", "rev-parse", "--is-inside-work-tree"], cwd=directory, check=True, capture_output=True, text=True, ) # Also get the repository root to use for all git operations try: await run_command( ["git", "rev-parse", "--show-toplevel"], cwd=directory, check=True, capture_output=True, text=True, ) # Store the repository root in a global or class variable if needed # This could be used to ensure all git operations use the same root return True except (subprocess.SubprocessError, OSError): # If we can't get the repo root, it's not a proper git repository return False except (subprocess.SubprocessError, OSError): return False async def get_ref_commit_chat_id(directory: str, ref_name: str) -> str | None: """Get the chat ID from a specific reference's commit message. Args: directory: The directory to check ref_name: The reference name to check Returns: The chat ID if found, None otherwise """ try: # Check if the reference exists result = await run_command( ["git", "show-ref", "--verify", ref_name], cwd=directory, check=False, capture_output=True, text=True, ) if result.returncode != 0: # Reference doesn't exist return None # Get the commit message from the reference message_result = await run_command( ["git", "log", "-1", "--pretty=%B", ref_name], cwd=directory, check=True, capture_output=True, text=True, ) commit_message = message_result.stdout.strip() # Use regex to find the last occurrence of codemcp-id: XXX # The pattern looks for "codemcp-id: " followed by any characters up to a newline or end of string matches = re.findall(r"codemcp-id:\s*([^\n]*)", commit_message) # Return the last match if any matches found if matches: return matches[-1].strip() return None except Exception as e: logging.warning( f"Exception when getting reference commit chat ID: {e!s}", exc_info=True ) return None