NextERP/NextERP-scripts
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update utility

Browse Source
This commit is contained in:
KaySar12 2025-04-08 09:49:25 +07:00
parent 4938a55ad7
commit f12b605180
6 changed files with 310 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
cli/git.py
View File

@ -1,11 +1,71 @@
# cli/module.py
import tqdm
import argparse
from services.git.handler import GitHandler
import lib.color_log as color_log
def setup_cli(subparsers):
pass
git_parser = subparsers.add_parser('git', help='Git operations for Odoo instances')
git_subparsers = git_parser.add_subparsers(dest='git_command', help='Git commands')
# Clone command
clone_parser = git_subparsers.add_parser('clone', help='Clone repository for an instance')
clone_parser.add_argument('instance_name', help='Name of the instance to clone repository for')
clone_parser.add_argument('--branch', help='Branch to clone (optional)')
# Pull command
pull_parser = git_subparsers.add_parser('pull', help='Pull updates for an instance')
pull_parser.add_argument('instance_name', help='Name of the instance to pull updates for')
pull_parser.add_argument('--branch', help='Branch to pull from (optional)')
pull_parser.add_argument('--force', action='store_true', help='Force pull with hard reset (discards local changes)')
# Get commit command
commit_parser = git_subparsers.add_parser('commit', help='Get current commit hash')
commit_parser.add_argument('instance_name', help='Name of the instance to get commit for')
git_parser.set_defaults(func=git)
def git(args):
git_handler = GitHandler(args.repo_url, args.local_path, args.branch)
git_handler.pull_updates()
git_handler = GitHandler(config_path="utility/config/settings.yaml")
if args.git_command == 'clone':
try:
success = git_handler.clone_or_open_repo(
instance_name=args.instance_name,
branch=args.branch
)
if success:
color_log.Show("OK", f"Successfully cloned repository for {args.instance_name}")
else:
color_log.Show("FAILED", f"Failed to clone repository for {args.instance_name}")
except Exception as e:
color_log.Show("FAILED", f"Error cloning repository: {str(e)}")
elif args.git_command == 'pull':
try:
success = git_handler.pull_updates(
instance_name=args.instance_name,
branch=args.branch,
force=args.force
)
if success:
if args.force:
color_log.Show("OK", f"Successfully force pulled updates for {args.instance_name}")
else:
color_log.Show("OK", f"Successfully pulled updates for {args.instance_name}")
else:
color_log.Show("FAILED", f"Failed to pull updates for {args.instance_name}")
except Exception as e:
color_log.Show("FAILED", f"Error pulling updates: {str(e)}")
elif args.git_command == 'commit':
try:
commit_hash = git_handler.get_current_commit()
if commit_hash:
color_log.Show("INFO", f"Current commit for {args.instance_name}: {commit_hash}")
else:
color_log.Show("WARNING", f"No commit found for {args.instance_name}")
except Exception as e:
color_log.Show("FAILED", f"Error getting commit: {str(e)}")
else:
color_log.Show("ERROR", "Please specify a valid git command (clone/pull/commit)")

2
main.py
View File

@ -2,6 +2,7 @@
import argparse
from cli.service import setup_cli as setup_service_cli
from cli.module import setup_cli as setup_module_cli
from cli.git import setup_cli as setup_git_cli
def setup_cli():
@ -12,6 +13,7 @@ def setup_cli():
subparsers = parser.add_subparsers(dest="command", required=True)
setup_service_cli(subparsers)
setup_module_cli(subparsers)
setup_git_cli(subparsers)
return parser

110
scripts/update_modules.py
View File

@ -4,30 +4,96 @@ import os
import subprocess
import sys
from ..services import config as Config
from ..lib import color_log
# Parse arguments properly
parser = argparse.ArgumentParser(description="Uninstall module for each instance")
parser.add_argument(
"action", help="Action to perform", choices=["uninstall", "install", "upgrade"]
)
parser.add_argument("config_path", help="Path to the config file")
args = parser.parse_args()
def run_command(cmd, description):
"""Run a command and handle its output."""
try:
color_log.Show("INFO", f"Executing: {description}")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.stdout:
print(result.stdout.strip())
if result.stderr:
print(result.stderr.strip())
return result.returncode == 0
except subprocess.CalledProcessError as e:
color_log.Show("FAILED", f"Error executing {description}: {str(e)}")
return False
# Load the configuration
print(f"Config path: {args.config_path}")
config = Config.Config(config_path=args.config_path)
def update_instance(instance_name, action, force_pull=False):
"""Update a single instance with git pull, module update, and service restart."""
# 1. Pull latest code
pull_cmd = ["python", "utility/main.py", "git", "pull", instance_name]
if force_pull:
pull_cmd.append("--force")
if not run_command(pull_cmd, f"Pulling latest code for {instance_name}"):
color_log.Show("WARNING", f"Skipping module update for {instance_name} due to pull failure")
return False
instances = config.get_instances()
# 2. Update modules
module_cmd = ["python", "utility/main.py", "module", action, instance_name]
if not run_command(module_cmd, f"Updating modules for {instance_name}"):
color_log.Show("WARNING", f"Module update failed for {instance_name}")
return False
# Ensure instances is a valid list
if not isinstance(instances, list):
print("Error: instances is not a valid list.")
sys.exit(1)
# 3. Restart service
restart_cmd = ["python", "utility/main.py", "service", "restart", instance_name]
if not run_command(restart_cmd, f"Restarting service for {instance_name}"):
color_log.Show("WARNING", f"Service restart failed for {instance_name}")
return False
# Loop over each instance and run the uninstall command
for instance in instances:
if "name" in instance:
cmd = ["python", "utility/main.py", "module", args.action, instance["name"]]
subprocess.run(cmd)
else:
print(f"Warning: Instance missing 'name' field. Skipping.")
return True
def main():
# Parse arguments
parser = argparse.ArgumentParser(description="Update modules for all instances")
parser.add_argument(
"action",
help="Action to perform",
choices=["uninstall", "install", "upgrade"]
)
parser.add_argument(
"config_path",
help="Path to the config file"
)
parser.add_argument(
"--force-pull",
action="store_true",
help="Force pull with hard reset (discards local changes)"
)
args = parser.parse_args()
# Load the configuration
color_log.Show("INFO", f"Loading configuration from {args.config_path}")
config = Config.Config(config_path=args.config_path)
# Get instances
instances = config.get_instances()
if not isinstance(instances, list):
color_log.Show("FAILED", "Error: instances is not a valid list.")
sys.exit(1)
# Process each instance
success_count = 0
for instance in instances:
if "name" not in instance:
color_log.Show("WARNING", f"Instance missing 'name' field. Skipping.")
continue
instance_name = instance["name"]
color_log.Show("INFO", f"\nProcessing instance: {instance_name}")
if update_instance(instance_name, args.action, args.force_pull):
success_count += 1
color_log.Show("OK", f"Successfully updated {instance_name}")
else:
color_log.Show("FAILED", f"Failed to update {instance_name}")
# Summary
color_log.Show("INFO", f"\nUpdate Summary:")
color_log.Show("INFO", f"Total instances: {len(instances)}")
color_log.Show("INFO", f"Successful updates: {success_count}")
color_log.Show("INFO", f"Failed updates: {len(instances) - success_count}")
if __name__ == "__main__":
main()

18
services/config.py
View File

@ -1,4 +1,3 @@
import os
import yaml
class Config:
@ -18,4 +17,21 @@ class Config:
def get_instances(self):
"""Return the list of Odoo instances."""
return self.settings.get("odoo_instances", [])
def get_instance(self, name):
"""
Get a single instance configuration by name.
Args:
name (str): Name of the instance to retrieve
Returns:
dict: Instance configuration if found, None otherwise
"""
instances = self.get_instances()
for instance in instances:
if instance.get("name") == name:
return instance
return None

156
services/git/handler.py
View File

@ -3,32 +3,84 @@ 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}")
subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
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}"
cmd = (
f"git clone -b {branch or 'main'} {repo_url} {self.local_path}"
)
self._execute_command(cmd, "local")
return True
@ -37,36 +89,101 @@ class GitHandler:
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'"
result = self._execute_command(check_cmd, instance_name)
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
cmd = self._get_command(instance, "clone", repo_url, branch)
self._execute_command(cmd, instance_name)
# 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):
"""Pull updates with SSH support"""
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")
cmd = self._get_command(instance, "pull", branch=branch)
self._execute_command(cmd, instance_name)
# 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:
@ -75,16 +192,17 @@ class GitHandler:
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
"""
@ -92,9 +210,9 @@ class GitHandler:
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:
@ -104,14 +222,12 @@ class GitHandler:
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

4
services/odoo/connection.py
View File

@ -79,7 +79,9 @@ class OdooConnection:
def get_instances(self):
"""Return the list of configured instances."""
return self.instances
def get_instance(self,instance_name):
"""Return the instance configuration."""
return self.config.get_instance(instance_name)
def execute(self, instance_name, model, method, *args, **kwargs):
"""Execute a method on a model for a specific instance."""
odoo = self.get_connection(instance_name)