117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
import git
|
|
import os
|
|
from services.odoo.connection import OdooConnection
|
|
import lib.color_log as color_log
|
|
import subprocess
|
|
|
|
|
|
class GitHandler:
|
|
def __init__(self, config_path="config/settings.yaml"):
|
|
self.config = OdooConnection(config_path)
|
|
|
|
def _execute_command(self, cmd, instance_name):
|
|
"""Execute a shell command and handle errors."""
|
|
try:
|
|
color_log.Show("INFO", f"Executing command: {cmd}")
|
|
subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
|
|
color_log.Show("OK", f"Command executed successfully for {instance_name}")
|
|
except subprocess.CalledProcessError as e:
|
|
color_log.Show(
|
|
"FAILED",
|
|
f"Error performing git operation for {instance_name}: {e}",
|
|
)
|
|
|
|
raise
|
|
def clone_or_open_repo(self, instance_name=None, repo_url=None, branch=None):
|
|
"""Clone or open repository with SSH support"""
|
|
try:
|
|
if not instance_name:
|
|
# Local operation
|
|
if not os.path.exists(self.local_path):
|
|
cmd = f"git clone -b {branch or 'main'} {repo_url} {self.local_path}"
|
|
self._execute_command(cmd, "local")
|
|
return True
|
|
|
|
# Remote operation
|
|
instance = self.config.get_instance(instance_name)
|
|
if not instance:
|
|
raise ValueError(f"Instance {instance_name} not found")
|
|
|
|
# Check if repo exists
|
|
check_cmd = f"test -d {self.local_path}/.git && echo 'exists' || echo 'not exists'"
|
|
result = self._execute_command(check_cmd, instance_name)
|
|
|
|
if "not exists" in result:
|
|
# Clone repository
|
|
cmd = self._get_command(instance, "clone", repo_url, branch)
|
|
self._execute_command(cmd, instance_name)
|
|
return True
|
|
|
|
except Exception as e:
|
|
color_log.Show("FAILED", f"Error in clone_or_open_repo: {e}")
|
|
return False
|
|
|
|
def pull_updates(self, instance_name=None, branch=None):
|
|
"""Pull updates with SSH support"""
|
|
try:
|
|
if not instance_name:
|
|
# Local operation
|
|
cmd = f"git --git-dir={self.local_path}/.git --work-tree={self.local_path} pull origin {branch or 'main'}"
|
|
self._execute_command(cmd, "local")
|
|
return True
|
|
|
|
# Remote operation
|
|
instance = self.config.get_instance(instance_name)
|
|
if not instance:
|
|
raise ValueError(f"Instance {instance_name} not found")
|
|
|
|
cmd = self._get_command(instance, "pull", branch=branch)
|
|
self._execute_command(cmd, instance_name)
|
|
return True
|
|
|
|
except Exception as e:
|
|
color_log.Show("FAILED", f"Error pulling updates: {e}")
|
|
return False
|
|
|
|
def get_current_commit(self):
|
|
return self.repo.head.commit.hexsha if self.repo else None
|
|
def _get_command(self, instance, action, repo_url=None, branch=None):
|
|
"""
|
|
Generate the appropriate git command based on instance type and action.
|
|
|
|
Args:
|
|
instance (dict): Instance configuration
|
|
action (str): Git action (clone/pull)
|
|
repo_url (str, optional): Repository URL for clone operation
|
|
branch (str, optional): Branch name
|
|
|
|
Returns:
|
|
str: Generated git command
|
|
"""
|
|
host = instance["host"]
|
|
ssh_settings = instance.get("ssh", {})
|
|
ssh_user = ssh_settings.get("user", "root")
|
|
ssh_key_path = ssh_settings.get("key_path")
|
|
|
|
local_host = host in ["localhost", "127.0.0.1"]
|
|
|
|
# Base git command
|
|
if action == "clone":
|
|
if not repo_url:
|
|
raise ValueError("Repository URL is required for clone operation")
|
|
cmd = f"git clone -b {branch or 'main'} {repo_url} {self.local_path}"
|
|
elif action == "pull":
|
|
cmd = f"git --git-dir={self.local_path}/.git --work-tree={self.local_path} pull origin {branch or 'main'}"
|
|
else:
|
|
raise ValueError(f"Unsupported git action: {action}")
|
|
|
|
# Wrap with SSH if remote host
|
|
if not local_host:
|
|
if not ssh_key_path:
|
|
cmd = f"ssh -t {ssh_user}@{host} 'sudo {cmd}'"
|
|
else:
|
|
cmd = f"ssh -i {ssh_key_path} {ssh_user}@{host} 'sudo {cmd}'"
|
|
|
|
return cmd
|
|
|
|
|