From ab4de663341b9a2ab77edb49916a90ce7bae4772 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 20:07:52 +0000 Subject: [PATCH] Add utility modules: encryption, file_utils, git_utils, path_utils --- confsync/utils/git_utils.py | 217 ++++++++++++++++++++++++++++++++++++ 1 file changed, 217 insertions(+) create mode 100644 confsync/utils/git_utils.py diff --git a/confsync/utils/git_utils.py b/confsync/utils/git_utils.py new file mode 100644 index 0000000..f53d0b4 --- /dev/null +++ b/confsync/utils/git_utils.py @@ -0,0 +1,217 @@ +"""Git utilities for ConfSync.""" + +import os +from typing import Dict, List, Optional + + +class GitManager: + """Manages Git operations for sync and history tracking.""" + + def __init__(self, repo_path: Optional[str] = None): + self.repo_path = repo_path + self._repo = None + + @property + def repo(self): + """Get the GitPython repo object.""" + if self._repo is None and self.repo_path: + try: + import git + self._repo = git.Repo(self.repo_path) + except ImportError: + pass + return self._repo + + def init_repo(self, path: str, bare: bool = False) -> bool: + """Initialize a new Git repository.""" + try: + import git + if bare: + git.Repo.init(path, bare=True) + else: + git.Repo.init(path) + self.repo_path = path + return True + except ImportError: + print("Error: GitPython not installed") + return False + except Exception as e: + print(f"Error initializing repository: {e}") + return False + + def clone_repo(self, url: str, path: str, branch: Optional[str] = None) -> bool: + """Clone a remote repository.""" + try: + import git + if branch: + git.Repo.clone_from(url, to_path=path, branch=branch) + else: + git.Repo.clone_from(url, to_path=path) + self.repo_path = path + return True + except ImportError: + print("Error: GitPython not installed") + return False + except Exception as e: + print(f"Error cloning repository: {e}") + return False + + def add_files(self, files: List[str], repo_path: Optional[str] = None) -> bool: + """Add files to staging.""" + try: + repo = self._get_repo(repo_path) + repo.index.add(files) + return True + except Exception as e: + print(f"Error adding files: {e}") + return False + + def commit(self, message: str, repo_path: Optional[str] = None) -> bool: + """Commit staged changes.""" + try: + repo = self._get_repo(repo_path) + if repo.is_dirty() or repo.untracked_files: + repo.index.commit(message) + return True + return False + except Exception as e: + print(f"Error committing: {e}") + return False + + def push(self, remote: str = "origin", branch: Optional[str] = None, repo_path: Optional[str] = None) -> bool: + """Push changes to remote.""" + try: + repo = self._get_repo(repo_path) + push_kwargs = {} + if branch: + push_kwargs["refspec"] = branch + repo.remotes[remote].push(**push_kwargs) + return True + except Exception as e: + print(f"Error pushing: {e}") + return False + + def pull(self, remote: str = "origin", branch: Optional[str] = None, repo_path: Optional[str] = None) -> bool: + """Pull changes from remote.""" + try: + repo = self._get_repo(repo_path) + pull_kwargs = {} + if branch: + pull_kwargs["branch"] = branch + repo.remotes[remote].pull(**pull_kwargs) + return True + except Exception as e: + print(f"Error pulling: {e}") + return False + + def get_status(self, repo_path: Optional[str] = None) -> Dict: + """Get repository status.""" + try: + repo = self._get_repo(repo_path) + return { + "is_dirty": repo.is_dirty(), + "untracked_files": repo.untracked_files, + "modified_files": [item.a_path for item in repo.index.diff(None)], + "staged_files": [item.a_path for item in repo.index.diff("HEAD")], + "ahead": 0, + "behind": 0, + } + except Exception as e: + print(f"Error getting status: {e}") + return {} + + def get_file_diff(self, file_path: str, repo_path: Optional[str] = None) -> str: + """Get diff for a specific file.""" + try: + repo = self._get_repo(repo_path) + if os.path.exists(os.path.join(repo.working_dir, file_path)): + diff = repo.git.diff(file_path) + return diff + return "" + except Exception as e: + print(f"Error getting diff: {e}") + return "" + + def get_commit_history(self, path: Optional[str] = None, max_count: int = 20, repo_path: Optional[str] = None) -> List[Dict]: + """Get commit history.""" + try: + repo = self._get_repo(repo_path) + commits = [] + if path: + for commit in repo.iter_commits("HEAD", paths=path, max_count=max_count): + commits.append({ + "hexsha": commit.hexsha, + "message": commit.message, + "author": str(commit.author), + "authored_datetime": commit.authored_datetime.isoformat(), + }) + else: + for commit in repo.iter_commits("HEAD", max_count=max_count): + commits.append({ + "hexsha": commit.hexsha, + "message": commit.message, + "author": str(commit.author), + "authored_datetime": commit.authored_datetime.isoformat(), + }) + return commits + except Exception as e: + print(f"Error getting history: {e}") + return [] + + def checkout(self, commit_or_branch: str, repo_path: Optional[str] = None) -> bool: + """Checkout a specific commit or branch.""" + try: + repo = self._get_repo(repo_path) + repo.git.checkout(commit_or_branch) + return True + except Exception as e: + print(f"Error checking out: {e}") + return False + + def create_branch(self, branch_name: str, repo_path: Optional[str] = None) -> bool: + """Create a new branch.""" + try: + repo = self._get_repo(repo_path) + repo.create_head(branch_name) + return True + except Exception as e: + print(f"Error creating branch: {e}") + return False + + def get_branches(self, repo_path: Optional[str] = None) -> List[str]: + """Get list of branches.""" + try: + repo = self._get_repo(repo_path) + return [head.name for head in repo.heads] + except Exception as e: + print(f"Error getting branches: {e}") + return [] + + def add_remote(self, name: str, url: str, repo_path: Optional[str] = None) -> bool: + """Add a remote repository.""" + try: + repo = self._get_repo(repo_path) + repo.create_remote(name, url) + return True + except Exception as e: + print(f"Error adding remote: {e}") + return False + + def get_remotes(self, repo_path: Optional[str] = None) -> Dict[str, str]: + """Get list of remotes and their URLs.""" + try: + repo = self._get_repo(repo_path) + return {remote.name: remote.url for remote in repo.remotes} + except Exception as e: + print(f"Error getting remotes: {e}") + return {} + + def _get_repo(self, repo_path: Optional[str] = None): + """Get or initialize the repository.""" + import git + path = repo_path or self.repo_path + if path is None: + raise ValueError("Repository path not specified") + if self._repo is None or self._repo.working_dir != os.path.abspath(path): + self._repo = git.Repo(path) + return self._repo