log-center/repair_agent/agent/git_manager.py

"""
Git Manager - Git 操作管理
"""
import os
from typing import Optional
from git import Repo, InvalidGitRepositoryError
from loguru import logger

from ..config import settings


class GitManager:
    """负责 Git 操作"""

    def __init__(self, project_path: str, github_repo: str = ""):
        self.project_path = project_path
        self.github_repo = github_repo.strip()
        self.repo: Optional[Repo] = None
        self._init_repo()

    def _init_repo(self):
        """初始化 Git 仓库"""
        try:
            self.repo = Repo(self.project_path)
            # 配置 Git 用户
            with self.repo.config_writer() as config:
                config.set_value("user", "name", settings.git_user_name)
                config.set_value("user", "email", settings.git_user_email)

            # 如果指定了 GitHub 仓库地址，确保 origin 指向正确的仓库
            if self.github_repo:
                self._ensure_remote(self.github_repo)

            logger.info(f"Git 仓库初始化成功: {self.project_path}")
        except InvalidGitRepositoryError:
            logger.error(f"无效的 Git 仓库: {self.project_path}")
            self.repo = None

    def _ensure_remote(self, repo_url: str):
        """确保 origin remote 指向指定的仓库地址"""
        if not self.repo:
            return
        try:
            if "origin" in [r.name for r in self.repo.remotes]:
                current_url = self.repo.remotes.origin.url
                if current_url != repo_url:
                    self.repo.remotes.origin.set_url(repo_url)
                    logger.info(f"更新 origin 地址: {repo_url}")
            else:
                self.repo.create_remote("origin", repo_url)
                logger.info(f"添加 origin 地址: {repo_url}")
        except Exception as e:
            logger.error(f"配置 remote 失败: {e}")
    
    def pull(self) -> bool:
        """拉取最新代码"""
        if not self.repo:
            return False
        
        try:
            origin = self.repo.remotes.origin
            origin.pull()
            logger.info("代码拉取成功")
            return True
        except Exception as e:
            logger.error(f"拉取代码失败: {e}")
            return False
    
    def create_branch(self, branch_name: str) -> bool:
        """
        创建并切换到新分支
        
        Args:
            branch_name: 分支名称
        
        Returns:
            是否成功
        """
        if not self.repo:
            return False
        
        try:
            # 先切换到 main 分支
            if "main" in self.repo.heads:
                self.repo.heads.main.checkout()
            elif "master" in self.repo.heads:
                self.repo.heads.master.checkout()
            
            # 创建新分支
            new_branch = self.repo.create_head(branch_name)
            new_branch.checkout()
            logger.info(f"创建并切换到分支: {branch_name}")
            return True
            
        except Exception as e:
            logger.error(f"创建分支失败: {e}")
            return False
    
    def checkout(self, branch_name: str) -> bool:
        """切换分支"""
        if not self.repo:
            return False
        
        try:
            if branch_name in self.repo.heads:
                self.repo.heads[branch_name].checkout()
                logger.info(f"切换到分支: {branch_name}")
                return True
            else:
                logger.error(f"分支不存在: {branch_name}")
                return False
        except Exception as e:
            logger.error(f"切换分支失败: {e}")
            return False
    
    def get_diff(self) -> str:
        """获取当前修改的 diff"""
        if not self.repo:
            return ""
        
        try:
            return self.repo.git.diff()
        except Exception as e:
            logger.error(f"获取 diff 失败: {e}")
            return ""
    
    def get_modified_files(self) -> list[str]:
        """获取已修改的文件列表"""
        if not self.repo:
            return []
        
        try:
            return [item.a_path for item in self.repo.index.diff(None)]
        except Exception as e:
            logger.error(f"获取修改文件失败: {e}")
            return []
    
    def commit(self, message: str) -> bool:
        """提交更改"""
        if not self.repo:
            return False
        
        try:
            # 添加所有修改
            self.repo.git.add(A=True)
            
            # 检查是否有更改
            if not self.repo.is_dirty():
                logger.warning("没有需要提交的更改")
                return False
            
            self.repo.index.commit(message)
            logger.info(f"提交成功: {message}")
            return True
            
        except Exception as e:
            logger.error(f"提交失败: {e}")
            return False
    
    def push(self, branch_name: Optional[str] = None) -> bool:
        """推送到远程"""
        if not self.repo:
            return False
        
        try:
            origin = self.repo.remotes.origin
            if branch_name:
                origin.push(branch_name)
            else:
                origin.push()
            logger.info("推送成功")
            return True
        except Exception as e:
            logger.error(f"推送失败: {e}")
            return False
    
    def reset_hard(self) -> bool:
        """重置所有更改"""
        if not self.repo:
            return False
        
        try:
            self.repo.git.reset("--hard")
            self.repo.git.clean("-fd")
            logger.info("重置成功")
            return True
        except Exception as e:
            logger.error(f"重置失败: {e}")
            return False
    
    def get_current_branch(self) -> str:
        """获取当前分支名"""
        if not self.repo:
            return ""
        
        try:
            return self.repo.active_branch.name
        except Exception:
            return ""