234 lines
8.9 KiB
Python
234 lines
8.9 KiB
Python
import git
|
|
import os
|
|
from services.odoo.connection import OdooConnection
|
|
import lib.color_log as color_log
|
|
import subprocess
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
class GitHandler:
|
|
def __init__(self, config_path="config/settings.yaml"):
|
|
self.config = OdooConnection(config_path)
|
|
self.local_path = None # Will be set based on instance configuration
|
|
|
|
def _execute_command(self, cmd, instance_name):
|
|
"""Execute a shell command and handle errors."""
|
|
try:
|
|
color_log.Show("INFO", f"Executing command: {cmd}")
|
|
result = subprocess.run(
|
|
cmd, shell=True, check=True, capture_output=True, text=True
|
|
)
|
|
|
|
# Print the output if there is any
|
|
if result.stdout:
|
|
print(result.stdout.strip())
|
|
if result.stderr:
|
|
print(result.stderr.strip())
|
|
|
|
color_log.Show("OK", f"Command executed successfully for {instance_name}")
|
|
return result.stdout
|
|
except subprocess.CalledProcessError as e:
|
|
# Print the error output
|
|
if e.stdout:
|
|
print(e.stdout.strip())
|
|
if e.stderr:
|
|
print(e.stderr.strip())
|
|
|
|
color_log.Show(
|
|
"FAILED",
|
|
f"Error performing git operation for {instance_name}: {e}",
|
|
)
|
|
raise
|
|
|
|
def _get_auth_url(self, repo_url, instance):
|
|
"""Add authentication to repository URL if credentials are provided."""
|
|
if not repo_url:
|
|
return repo_url
|
|
|
|
git_config = instance.get("git", {})
|
|
username = git_config.get("username")
|
|
password = git_config.get("password")
|
|
|
|
if username and password:
|
|
parsed_url = urlparse(repo_url)
|
|
# Replace the URL with authenticated version
|
|
auth_url = f"{parsed_url.scheme}://{username}:{password}@{parsed_url.netloc}{parsed_url.path}"
|
|
return auth_url
|
|
return repo_url
|
|
|
|
def _get_remote_command(self, instance, cmd):
|
|
"""Generate SSH command for remote execution."""
|
|
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"]
|
|
|
|
if not local_host:
|
|
if not ssh_key_path:
|
|
return f"ssh -t {ssh_user}@{host} 'sudo -s bash -c \"{cmd}\"'"
|
|
else:
|
|
return f"ssh -i {ssh_key_path} {ssh_user}@{host} 'sudo -s bash -c \"{cmd}\"'"
|
|
return cmd
|
|
|
|
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")
|
|
|
|
# Set local_path from instance configuration
|
|
self.local_path = instance.get("git", {}).get("local_path")
|
|
if not self.local_path:
|
|
raise ValueError(
|
|
f"No local_path configured for instance {instance_name}"
|
|
)
|
|
|
|
# Get repository URL from instance configuration if not provided
|
|
if not repo_url:
|
|
repo_url = instance.get("git", {}).get("repo_url")
|
|
if not repo_url:
|
|
raise ValueError(
|
|
f"No repository URL configured for instance {instance_name}"
|
|
)
|
|
|
|
# Add authentication to repository URL
|
|
auth_url = self._get_auth_url(repo_url, instance)
|
|
|
|
# Check if repo exists
|
|
check_cmd = (
|
|
f"test -d {self.local_path}/.git && echo 'exists' || echo 'not exists'"
|
|
)
|
|
remote_check_cmd = self._get_remote_command(instance, check_cmd)
|
|
result = self._execute_command(remote_check_cmd, instance_name)
|
|
|
|
if "not exists" in result:
|
|
# Clone repository with authentication
|
|
clone_cmd = (
|
|
f"git clone -b {branch or 'main'} {auth_url} {self.local_path}"
|
|
)
|
|
remote_clone_cmd = self._get_remote_command(instance, clone_cmd)
|
|
self._execute_command(remote_clone_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, force=False):
|
|
"""Pull updates with SSH support
|
|
|
|
Args:
|
|
instance_name (str, optional): Name of the instance
|
|
branch (str, optional): Branch to pull from
|
|
force (bool, optional): If True, will perform hard reset before pulling
|
|
"""
|
|
try:
|
|
if not instance_name:
|
|
# Local operation
|
|
if not self.local_path:
|
|
raise ValueError("local_path not set for 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)
|
|
branch = instance.get("git", {}).get("branch") or branch
|
|
if not instance:
|
|
raise ValueError(f"Instance {instance_name} not found")
|
|
|
|
# Set local_path from instance configuration
|
|
self.local_path = instance.get("git", {}).get("local_path")
|
|
if not self.local_path:
|
|
raise ValueError(
|
|
f"No local_path configured for instance {instance_name}"
|
|
)
|
|
|
|
# Get repository URL and add authentication
|
|
repo_url = instance.get("git", {}).get("repo_url")
|
|
auth_url = self._get_auth_url(repo_url, instance)
|
|
|
|
# Configure git to use credentials and set remote URL
|
|
git_commands = [
|
|
f"cd {self.local_path}",
|
|
f"git config --local credential.helper store",
|
|
f"git remote set-url origin {auth_url}",
|
|
]
|
|
|
|
# Add force reset commands if force is True
|
|
if force:
|
|
git_commands.extend(
|
|
[
|
|
"git fetch origin",
|
|
f"git reset --hard origin/{branch or 'main'}",
|
|
"git clean -fd", # Remove untracked files and directories
|
|
]
|
|
)
|
|
else:
|
|
git_commands.append(f"git pull origin {branch or 'main'}")
|
|
|
|
# Combine commands and execute remotely
|
|
combined_cmd = " && ".join(git_commands)
|
|
remote_cmd = self._get_remote_command(instance, combined_cmd)
|
|
self._execute_command(remote_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
|