Files
pkgmgr/main.py

782 lines
37 KiB
Python
Raw Normal View History

2025-03-04 15:41:39 +01:00
import re
import hashlib
2025-03-04 13:17:57 +01:00
import os
import subprocess
import shutil
import sys
import yaml
2025-03-05 09:11:45 +01:00
import argparse
2025-03-05 11:20:59 +01:00
import json
2025-03-04 13:17:57 +01:00
2025-03-04 14:51:31 +01:00
# Define configuration file paths.
DEFAULT_CONFIG_PATH = os.path.join("config", "defaults.yaml")
USER_CONFIG_PATH = os.path.join("config", "config.yaml")
2025-03-04 13:17:57 +01:00
BIN_DIR = os.path.expanduser("~/.local/bin")
2025-03-05 09:11:45 +01:00
# Commands proxied by package-manager
GIT_DEFAULT_COMMANDS = [
"pull",
"push",
"diff",
"add",
"show",
"checkout",
"clone",
2025-03-05 11:48:32 +01:00
"reset",
2025-03-05 14:19:08 +01:00
"revert",
"commit"
2025-03-05 09:11:45 +01:00
]
2025-03-04 14:51:31 +01:00
def load_config():
2025-03-04 15:41:39 +01:00
"""Load configuration from defaults and merge in user config if present."""
2025-03-04 14:51:31 +01:00
if not os.path.exists(DEFAULT_CONFIG_PATH):
print(f"Default configuration file '{DEFAULT_CONFIG_PATH}' not found.")
2025-03-04 13:17:57 +01:00
sys.exit(1)
2025-03-04 14:51:31 +01:00
with open(DEFAULT_CONFIG_PATH, 'r') as f:
2025-03-04 13:17:57 +01:00
config = yaml.safe_load(f)
2025-03-05 11:20:59 +01:00
if "directories" not in config or "repositories" not in config:
print("Default config file must contain 'directories' and 'repositories' keys.")
2025-03-04 13:17:57 +01:00
sys.exit(1)
2025-03-04 14:51:31 +01:00
if os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, 'r') as f:
user_config = yaml.safe_load(f)
if user_config:
2025-03-05 11:20:59 +01:00
if "directories" in user_config:
config["directories"] = user_config["directories"]
if "repositories" in user_config:
config["repositories"].extend(user_config["repositories"])
2025-03-04 13:17:57 +01:00
return config
2025-03-04 14:51:31 +01:00
def save_user_config(user_config):
"""Save the user configuration to USER_CONFIG_PATH."""
os.makedirs(os.path.dirname(USER_CONFIG_PATH), exist_ok=True)
with open(USER_CONFIG_PATH, 'w') as f:
yaml.dump(user_config, f)
print(f"User configuration updated in {USER_CONFIG_PATH}.")
2025-03-04 13:43:23 +01:00
2025-03-04 13:17:57 +01:00
def run_command(command, cwd=None, preview=False):
"""Run a shell command in a given directory, or print it in preview mode."""
if preview:
print(f"[Preview] In '{cwd or os.getcwd()}': {command}")
else:
print(f"Running in '{cwd or os.getcwd()}': {command}")
subprocess.run(command, cwd=cwd, shell=True, check=False)
def get_repo_identifier(repo, all_repos):
"""
Return a unique identifier for the repository.
2025-03-04 15:41:39 +01:00
If the repository name is unique among all_repos, return repository name;
otherwise, return 'provider/account/repository'.
2025-03-04 13:17:57 +01:00
"""
repo_name = repo.get("repository")
count = sum(1 for r in all_repos if r.get("repository") == repo_name)
if count == 1:
return repo_name
else:
return f'{repo.get("provider")}/{repo.get("account")}/{repo.get("repository")}'
def resolve_repos(identifiers, all_repos):
"""
2025-03-04 13:43:23 +01:00
Given a list of identifier strings, return a list of repository configs.
2025-03-04 15:58:37 +01:00
The identifier can be:
- the full identifier "provider/account/repository"
- the repository name (if unique among all_repos)
- the alias (if defined)
2025-03-04 13:17:57 +01:00
"""
selected = []
for ident in identifiers:
matches = []
for repo in all_repos:
full_id = f'{repo.get("provider")}/{repo.get("account")}/{repo.get("repository")}'
if ident == full_id:
matches.append(repo)
2025-03-04 15:58:37 +01:00
elif ident == repo.get("alias"):
matches.append(repo)
2025-03-04 13:17:57 +01:00
elif ident == repo.get("repository"):
2025-03-04 15:58:37 +01:00
# Only match if repository name is unique among all_repos.
2025-03-04 13:17:57 +01:00
if sum(1 for r in all_repos if r.get("repository") == ident) == 1:
matches.append(repo)
if not matches:
print(f"Identifier '{ident}' did not match any repository in config.")
else:
selected.extend(matches)
return selected
2025-03-04 15:41:39 +01:00
def filter_ignored(repos):
"""Filter out repositories that have 'ignore' set to True."""
return [r for r in repos if not r.get("ignore", False)]
def generate_alias(repo, bin_dir, existing_aliases):
"""
Generate an alias for a repository based on its repository name.
Steps:
1. Keep only consonants from the repository name (letters from BCDFGHJKLMNPQRSTVWXYZ).
2. Collapse consecutive identical consonants.
3. Truncate to at most 12 characters.
4. If that alias conflicts (already in existing_aliases or a file exists in bin_dir),
then prefix with the first letter of provider and account.
5. If still conflicting, append a three-character hash until the alias is unique.
"""
repo_name = repo.get("repository")
# Keep only consonants.
consonants = re.sub(r"[^bcdfghjklmnpqrstvwxyzBCDFGHJKLMNPQRSTVWXYZ]", "", repo_name)
# Collapse consecutive identical consonants.
collapsed = re.sub(r"(.)\1+", r"\1", consonants)
base_alias = collapsed[:12] if len(collapsed) > 12 else collapsed
candidate = base_alias.lower()
def conflict(alias):
alias_path = os.path.join(bin_dir, alias)
return alias in existing_aliases or os.path.exists(alias_path)
if not conflict(candidate):
return candidate
prefix = (repo.get("provider", "")[0] + repo.get("account", "")[0]).lower()
candidate2 = (prefix + candidate)[:12]
if not conflict(candidate2):
return candidate2
h = hashlib.md5(repo_name.encode("utf-8")).hexdigest()[:3]
candidate3 = (candidate2 + h)[:12]
while conflict(candidate3):
candidate3 += "x"
candidate3 = candidate3[:12]
return candidate3
2025-03-05 11:20:59 +01:00
def create_executable(repo, repositories_base_dir, bin_dir, all_repos, quiet=False, preview=False, no_verification=False):
"""
Create an executable bash wrapper for the repository.
2025-03-04 14:02:21 +01:00
If 'verified' is set, the wrapper will checkout that commit and warn (unless quiet is True).
If no verified commit is set, a warning is printed unless quiet is True.
2025-03-04 15:41:39 +01:00
If an 'alias' field is provided, a symlink is created in bin_dir with that alias.
2025-03-04 14:02:21 +01:00
"""
2025-03-04 13:17:57 +01:00
repo_identifier = get_repo_identifier(repo, all_repos)
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir,repo)
2025-03-04 13:17:57 +01:00
command = repo.get("command")
if not command:
main_sh = os.path.join(repo_dir, "main.sh")
main_py = os.path.join(repo_dir, "main.py")
if os.path.exists(main_sh):
command = "bash main.sh"
elif os.path.exists(main_py):
command = "python3 main.py"
else:
if not quiet:
print(f"No command defined and no main.sh/main.py found in {repo_dir}. Skipping alias creation.")
2025-03-04 13:17:57 +01:00
return
2025-03-04 14:02:21 +01:00
2025-03-04 14:04:54 +01:00
ORANGE = r"\033[38;5;208m"
RESET = r"\033[0m"
2025-03-04 18:58:11 +01:00
if no_verification:
preamble = ""
else:
if verified := repo.get("verified"):
if not quiet:
preamble = f"""\
2025-03-04 14:04:54 +01:00
git checkout {verified} || echo -e "{ORANGE}Warning: Failed to checkout commit {verified}.{RESET}"
2025-03-04 14:02:21 +01:00
CURRENT=$(git rev-parse HEAD)
if [ "$CURRENT" != "{verified}" ]; then
2025-03-04 14:04:54 +01:00
echo -e "{ORANGE}Warning: Current commit ($CURRENT) does not match verified commit ({verified}).{RESET}"
2025-03-04 14:02:21 +01:00
fi
"""
2025-03-04 18:58:11 +01:00
else:
preamble = ""
else:
2025-03-04 18:58:11 +01:00
preamble = "" if quiet else f'echo -e "{ORANGE}Warning: No verified commit set for this repository.{RESET}"'
2025-03-04 13:17:57 +01:00
script_content = f"""#!/bin/bash
cd "{repo_dir}"
2025-03-04 14:02:21 +01:00
{preamble}
2025-03-04 13:17:57 +01:00
{command} "$@"
"""
alias_path = os.path.join(bin_dir, repo_identifier)
if preview:
print(f"[Preview] Would create executable '{alias_path}' with content:\n{script_content}")
else:
os.makedirs(bin_dir, exist_ok=True)
with open(alias_path, "w") as f:
f.write(script_content)
os.chmod(alias_path, 0o755)
if not quiet:
print(f"Installed executable for {repo_identifier} at {alias_path}")
2025-03-04 13:17:57 +01:00
2025-03-04 14:38:33 +01:00
alias_name = repo.get("alias")
if alias_name:
alias_link_path = os.path.join(bin_dir, alias_name)
try:
if os.path.exists(alias_link_path) or os.path.islink(alias_link_path):
os.remove(alias_link_path)
os.symlink(alias_path, alias_link_path)
if not quiet:
print(f"Created alias '{alias_name}' pointing to {repo_identifier}")
2025-03-04 14:38:33 +01:00
except Exception as e:
if not quiet:
print(f"Error creating alias '{alias_name}': {e}")
2025-03-05 11:20:59 +01:00
def install_repos(selected_repos, repositories_base_dir, bin_dir, all_repos:[], no_verification:bool, preview=False, quiet=False):
2025-03-04 14:11:56 +01:00
"""Install repositories by creating executable wrappers and running setup."""
2025-03-04 13:17:57 +01:00
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir,repo)
2025-03-04 13:17:57 +01:00
if not os.path.exists(repo_dir):
print(f"Repository directory '{repo_dir}' does not exist. Clone it first.")
continue
2025-03-05 11:20:59 +01:00
create_executable(repo, repositories_base_dir, bin_dir, all_repos, quiet=quiet, preview=preview, no_verification=no_verification)
2025-03-04 13:17:57 +01:00
setup_cmd = repo.get("setup")
if setup_cmd:
run_command(setup_cmd, cwd=repo_dir, preview=preview)
2025-03-05 11:20:59 +01:00
def exec_git_command(selected_repos, repositories_base_dir, all_repos, git_cmd, extra_args, preview=False):
2025-03-04 14:32:05 +01:00
"""Execute a given git command with extra arguments for each repository."""
2025-03-04 13:17:57 +01:00
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir,repo)
2025-03-04 13:17:57 +01:00
if os.path.exists(repo_dir):
2025-03-04 14:32:05 +01:00
full_cmd = f"git {git_cmd} {' '.join(extra_args)}"
run_command(full_cmd, cwd=repo_dir, preview=preview)
2025-03-04 13:17:57 +01:00
else:
print(f"Repository directory '{repo_dir}' not found for {repo_identifier}.")
2025-03-04 14:32:05 +01:00
2025-03-05 11:20:59 +01:00
def status_repos(selected_repos, repositories_base_dir, all_repos, extra_args, list_only=False, system_status=False, preview=False):
2025-03-04 14:32:05 +01:00
if system_status:
print("System status:")
run_command("yay -Qu", preview=preview)
if list_only:
for repo in selected_repos:
print(get_repo_identifier(repo, all_repos))
else:
2025-03-05 11:20:59 +01:00
exec_git_command(selected_repos, repositories_base_dir, all_repos, "status", extra_args, preview)
2025-03-04 14:32:05 +01:00
2025-03-05 11:20:59 +01:00
def get_repo_dir(repositories_base_dir:str,repo:{})->str:
2025-03-05 09:11:45 +01:00
try:
2025-03-05 11:20:59 +01:00
return os.path.join(repositories_base_dir, repo.get("provider"), repo.get("account"), repo.get("repository"))
2025-03-05 09:11:45 +01:00
except TypeError as e:
2025-03-05 11:20:59 +01:00
if repositories_base_dir:
2025-03-05 09:11:45 +01:00
print(f"Error: {e} \nThe repository {repo} seems not correct configured.\nPlease configure it correct.")
for key in ["provider","account","repository"]:
if not repo.get(key,False):
print(f"Key '{key}' is missing.")
else:
print(f"Error: {e} \nThe base {base} seems not correct configured.\nPlease configure it correct.")
sys.exit(1)
2025-03-05 17:35:11 +01:00
def clone_repos(selected_repos, repositories_base_dir: str, all_repos, preview=False):
2025-03-04 13:17:57 +01:00
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
2025-03-05 17:35:11 +01:00
repo_dir = get_repo_dir(repositories_base_dir, repo)
2025-03-04 13:17:57 +01:00
if os.path.exists(repo_dir):
2025-03-05 17:35:11 +01:00
print(f"[INFO] Repository '{repo_identifier}' already exists at '{repo_dir}'. Skipping clone.")
2025-03-04 13:17:57 +01:00
continue
2025-03-05 17:35:11 +01:00
2025-03-04 13:17:57 +01:00
parent_dir = os.path.dirname(repo_dir)
os.makedirs(parent_dir, exist_ok=True)
2025-03-05 17:35:11 +01:00
try:
target = repo.get("replacement") if repo.get("replacement") else f"{repo.get('provider')}:{repo.get('account')}/{repo.get('repository')}"
clone_url = f"git@{target}.git"
print(f"[INFO] Attempting to clone '{repo_identifier}' using SSH from {clone_url} into '{repo_dir}'.")
run_command(f"git clone {clone_url} {repo_dir}", cwd=parent_dir, preview=preview)
except Exception as exception1:
print(f"[WARNING] SSH clone failed for '{repo_identifier}' (error: {exception1}). Trying HTTPS...")
target = repo.get("replacement") if repo.get("replacement") else f"{repo.get('provider')}/{repo.get('account')}/{repo.get('repository')}"
clone_url = f"https://{target}.git"
print(f"[INFO] Attempting to clone '{repo_identifier}' using HTTPS from {clone_url} into '{repo_dir}'.")
run_command(f"git clone {clone_url} {repo_dir}", cwd=parent_dir, preview=preview)
2025-03-04 13:17:57 +01:00
2025-03-05 11:20:59 +01:00
def deinstall_repos(selected_repos, repositories_base_dir, bin_dir, all_repos, preview=False):
2025-03-04 13:17:57 +01:00
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
alias_path = os.path.join(bin_dir, repo_identifier)
if os.path.exists(alias_path):
if preview:
print(f"[Preview] Would remove executable '{alias_path}'.")
else:
os.remove(alias_path)
print(f"Removed executable for {repo_identifier}.")
else:
print(f"No executable found for {repo_identifier} in {bin_dir}.")
teardown_cmd = repo.get("teardown")
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir,repo)
2025-03-04 13:17:57 +01:00
if teardown_cmd and os.path.exists(repo_dir):
run_command(teardown_cmd, cwd=repo_dir, preview=preview)
2025-03-05 11:20:59 +01:00
def delete_repos(selected_repos, repositories_base_dir, all_repos, preview=False):
2025-03-04 13:17:57 +01:00
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir,repo)
2025-03-04 13:17:57 +01:00
if os.path.exists(repo_dir):
if preview:
print(f"[Preview] Would delete directory '{repo_dir}' for {repo_identifier}.")
else:
shutil.rmtree(repo_dir)
print(f"Deleted repository directory '{repo_dir}' for {repo_identifier}.")
else:
print(f"Repository directory '{repo_dir}' not found for {repo_identifier}.")
2025-03-05 11:20:59 +01:00
def update_repos(selected_repos, repositories_base_dir, bin_dir, all_repos:[], no_verification:bool, system_update=False, preview=False, quiet=False):
git_default_exec(selected_repos, repositories_base_dir, all_repos, extra_args=[],command="pull", preview=preview)
install_repos(selected_repos, repositories_base_dir, bin_dir, all_repos, no_verification, preview=preview, quiet=quiet)
2025-03-04 13:17:57 +01:00
if system_update:
2025-03-04 18:06:10 +01:00
run_command("yay -Syu", preview=preview)
2025-03-04 13:17:57 +01:00
run_command("sudo pacman -Syyu", preview=preview)
2025-03-05 09:11:45 +01:00
2025-03-05 11:20:59 +01:00
def git_default_exec(selected_repos, repositories_base_dir, all_repos, extra_args, command:str, preview=False):
exec_git_command(selected_repos, repositories_base_dir, all_repos, command, extra_args, preview)
2025-03-04 14:22:21 +01:00
2025-03-04 13:43:23 +01:00
def show_config(selected_repos, full_config=False):
2025-03-04 15:41:39 +01:00
"""Display configuration for one or more repositories, or the entire merged config."""
2025-03-04 13:43:23 +01:00
if full_config:
2025-03-04 14:51:31 +01:00
merged = load_config()
print(yaml.dump(merged, default_flow_style=False))
2025-03-04 13:43:23 +01:00
else:
for repo in selected_repos:
identifier = f'{repo.get("provider")}/{repo.get("account")}/{repo.get("repository")}'
print(f"Repository: {identifier}")
for key, value in repo.items():
print(f" {key}: {value}")
print("-" * 40)
def interactive_add(config):
2025-03-04 15:41:39 +01:00
"""Interactively prompt the user to add a new repository entry to the user config."""
2025-03-04 13:43:23 +01:00
print("Adding a new repository configuration entry.")
new_entry = {}
new_entry["provider"] = input("Provider (e.g., github.com): ").strip()
new_entry["account"] = input("Account (e.g., yourusername): ").strip()
new_entry["repository"] = input("Repository name (e.g., mytool): ").strip()
new_entry["verified"] = input("Verified commit id: ").strip()
new_entry["command"] = input("Command (optional, leave blank to auto-detect): ").strip()
new_entry["description"] = input("Description (optional): ").strip()
new_entry["replacement"] = input("Replacement (optional): ").strip()
new_entry["setup"] = input("Setup command (optional): ").strip()
new_entry["teardown"] = input("Teardown command (optional): ").strip()
2025-03-04 14:51:31 +01:00
new_entry["alias"] = input("Alias (optional): ").strip()
2025-03-04 15:41:39 +01:00
# Allow the user to mark this entry as ignored.
ignore_val = input("Ignore this entry? (y/N): ").strip().lower()
if ignore_val == "y":
new_entry["ignore"] = True
2025-03-04 13:43:23 +01:00
print("\nNew entry:")
for key, value in new_entry.items():
if value:
print(f"{key}: {value}")
2025-03-04 14:51:31 +01:00
confirm = input("Add this entry to user config? (y/N): ").strip().lower()
2025-03-04 13:43:23 +01:00
if confirm == "y":
2025-03-04 14:51:31 +01:00
if os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, 'r') as f:
user_config = yaml.safe_load(f) or {}
else:
2025-03-05 11:20:59 +01:00
user_config = {"repositories": []}
user_config.setdefault("repositories", [])
user_config["repositories"].append(new_entry)
2025-03-04 14:51:31 +01:00
save_user_config(user_config)
2025-03-04 13:43:23 +01:00
else:
print("Entry not added.")
2025-03-04 15:41:39 +01:00
def edit_config():
"""Open the user configuration file in nano."""
run_command(f"nano {USER_CONFIG_PATH}")
2025-03-04 15:08:25 +01:00
def config_init(user_config, defaults_config, bin_dir):
"""
2025-03-04 15:18:22 +01:00
Scan the base directory (defaults_config["base"]) for repositories.
The folder structure is assumed to be:
{base}/{provider}/{account}/{repository}
2025-03-04 15:08:25 +01:00
For each repository found, automatically determine:
2025-03-04 15:41:39 +01:00
- provider, account, repository from folder names.
- verified: the latest commit (via 'git log -1 --format=%H').
- alias: generated from the repository name using generate_alias().
2025-03-05 11:20:59 +01:00
Repositories already defined in defaults_config["repositories"] or user_config["repositories"] are skipped.
2025-03-04 15:08:25 +01:00
"""
2025-03-05 11:20:59 +01:00
repositories_base_dir = os.path.expanduser(defaults_config["directories"]["repositories"])
if not os.path.isdir(repositories_base_dir):
print(f"Base directory '{repositories_base_dir}' does not exist.")
2025-03-04 15:08:25 +01:00
return
2025-03-04 15:41:39 +01:00
default_keys = {(entry.get("provider"), entry.get("account"), entry.get("repository"))
2025-03-05 11:20:59 +01:00
for entry in defaults_config.get("repositories", [])}
2025-03-04 15:41:39 +01:00
existing_keys = {(entry.get("provider"), entry.get("account"), entry.get("repository"))
2025-03-05 11:20:59 +01:00
for entry in user_config.get("repositories", [])}
existing_aliases = {entry.get("alias") for entry in user_config.get("repositories", []) if entry.get("alias")}
2025-03-04 15:08:25 +01:00
new_entries = []
2025-03-05 11:20:59 +01:00
for provider in os.listdir(repositories_base_dir):
provider_path = os.path.join(repositories_base_dir, provider)
2025-03-04 15:08:25 +01:00
if not os.path.isdir(provider_path):
continue
for account in os.listdir(provider_path):
account_path = os.path.join(provider_path, account)
if not os.path.isdir(account_path):
continue
for repo_name in os.listdir(account_path):
repo_path = os.path.join(account_path, repo_name)
if not os.path.isdir(repo_path):
continue
key = (provider, account, repo_name)
2025-03-04 15:18:22 +01:00
if key in default_keys or key in existing_keys:
2025-03-04 15:41:39 +01:00
continue
2025-03-04 15:08:25 +01:00
try:
result = subprocess.run(
["git", "log", "-1", "--format=%H"],
cwd=repo_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True,
)
verified = result.stdout.strip()
except Exception as e:
verified = ""
print(f"Could not determine latest commit for {repo_name} ({provider}/{account}): {e}")
entry = {
"provider": provider,
"account": account,
"repository": repo_name,
"verified": verified,
2025-03-04 15:41:39 +01:00
"ignore": True
2025-03-04 15:08:25 +01:00
}
alias = generate_alias({"repository": repo_name, "provider": provider, "account": account}, bin_dir, existing_aliases)
entry["alias"] = alias
existing_aliases.add(alias)
new_entries.append(entry)
print(f"Adding new repo entry: {entry}")
if new_entries:
2025-03-05 11:20:59 +01:00
user_config.setdefault("repositories", []).extend(new_entries)
2025-03-04 15:08:25 +01:00
save_user_config(user_config)
else:
print("No new repositories found.")
2025-03-05 09:11:45 +01:00
2025-03-05 11:48:32 +01:00
def get_selected_repos(show_all:bool,all_repos_list,identifiers=None):
selected = all_repos_list if show_all or (not identifiers) else resolve_repos(identifiers, all_repos_list)
2025-03-05 09:11:45 +01:00
return filter_ignored(selected)
2025-03-04 14:11:56 +01:00
2025-03-05 11:20:59 +01:00
def list_repositories(all_repos, repositories_base_dir, bin_dir, search_filter="", status_filter=""):
2025-03-05 10:03:20 +01:00
"""
List all repositories with their attributes and status information.
2025-03-05 10:21:25 +01:00
2025-03-05 10:03:20 +01:00
Parameters:
all_repos (list): List of repository configurations.
2025-03-05 11:20:59 +01:00
repositories_base_dir (str): The base directory where repositories are located.
2025-03-05 10:03:20 +01:00
bin_dir (str): The directory where executable wrappers are stored.
search_filter (str): Filter for repository attributes (case insensitive).
status_filter (str): Filter for computed status info (case insensitive).
2025-03-05 10:21:25 +01:00
For each repository, the identifier is printed in bold, the description (if available)
in italic, then all other attributes and computed status are printed.
If the repository is installed, a hint is displayed under the attributes.
2025-03-05 10:03:20 +01:00
Repositories are filtered out if either the search_filter is not found in any attribute or
if the status_filter is not found in the computed status string.
"""
search_filter = search_filter.lower() if search_filter else ""
status_filter = status_filter.lower() if status_filter else ""
2025-03-05 10:21:25 +01:00
# Define status colors using colors not used for other attributes:
# Avoid red (for ignore), blue (for homepage) and yellow (for verified).
status_colors = {
"Installed": "\033[1;32m", # Green
"Not Installed": "\033[1;35m", # Magenta
"Cloned": "\033[1;36m", # Cyan
"Clonable": "\033[1;37m", # White
"Ignored": "\033[38;5;208m", # Orange (extended)
"Active": "\033[38;5;129m", # Light Purple (extended)
2025-03-05 11:20:59 +01:00
"Installable": "\033[38;5;82m" # Light Green (extended)
2025-03-05 10:21:25 +01:00
}
2025-03-05 10:03:20 +01:00
for repo in all_repos:
# Combine all attribute values into one string for filtering.
repo_text = " ".join(str(v) for v in repo.values()).lower()
if search_filter and search_filter not in repo_text:
continue
# Compute status information for the repository.
identifier = get_repo_identifier(repo, all_repos)
executable_path = os.path.join(bin_dir, identifier)
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir, repo)
2025-03-05 10:03:20 +01:00
status_list = []
2025-03-05 10:21:25 +01:00
# Check if the executable exists (Installed).
2025-03-05 10:03:20 +01:00
if os.path.exists(executable_path):
status_list.append("Installed")
else:
status_list.append("Not Installed")
2025-03-05 10:21:25 +01:00
# Check if the repository directory exists (Cloned).
2025-03-05 10:03:20 +01:00
if os.path.exists(repo_dir):
status_list.append("Cloned")
else:
status_list.append("Clonable")
# Mark ignored repositories.
if repo.get("ignore", False):
status_list.append("Ignored")
else:
status_list.append("Active")
# Define installable as cloned but not installed.
if os.path.exists(repo_dir) and not os.path.exists(executable_path):
status_list.append("Installable")
2025-03-05 10:21:25 +01:00
# Build a colored status string.
colored_statuses = [f"{status_colors.get(s, '')}{s}\033[0m" for s in status_list]
status_str = ", ".join(colored_statuses)
2025-03-05 10:03:20 +01:00
# If a status_filter is provided, only display repos whose status contains the filter.
if status_filter and status_filter not in status_str.lower():
continue
2025-03-05 10:21:25 +01:00
# Display repository details:
# Print the identifier in bold.
print(f"\033[1m{identifier}\033[0m")
# Print the description in italic if it exists.
description = repo.get("description")
if description:
print(f"\n\033[3m{description}\033[0m")
print("\nAttributes:")
# Loop through all attributes.
2025-03-05 10:03:20 +01:00
for key, value in repo.items():
formatted_value = str(value)
# Special formatting for "verified" attribute (yellow).
if key == "verified" and value:
formatted_value = f"\033[1;33m{value}\033[0m"
# Special formatting for "ignore" flag (red if True).
if key == "ignore" and value:
formatted_value = f"\033[1;31m{value}\033[0m"
2025-03-05 10:21:25 +01:00
if key == "description":
continue
# Highlight homepage in blue.
if key.lower() == "homepage" and value:
formatted_value = f"\033[1;34m{value}\033[0m"
2025-03-05 10:03:20 +01:00
print(f" {key}: {formatted_value}")
2025-03-05 10:21:25 +01:00
# Always display the computed status.
2025-03-05 10:03:20 +01:00
print(f" Status: {status_str}")
2025-03-05 10:21:25 +01:00
# If the repository is installed, display a hint for more info.
if os.path.exists(executable_path):
2025-03-05 14:58:28 +01:00
print(f"\nMore information and help: \033[1;4mpkgmgr {identifier} --help\033[0m\n")
2025-03-05 10:03:20 +01:00
print("-" * 40)
2025-03-04 15:41:39 +01:00
# Main program.
2025-03-04 13:17:57 +01:00
if __name__ == "__main__":
2025-03-04 15:41:39 +01:00
config_merged = load_config()
2025-03-05 11:20:59 +01:00
repositories_base_dir = os.path.expanduser(config_merged["directories"]["repositories"])
all_repos_list = config_merged["repositories"]
2025-03-05 10:03:20 +01:00
description_text = """\
2025-03-05 10:26:01 +01:00
\033[1;32mPackage Manager 🤖📦\033[0m
2025-03-05 14:26:43 +01:00
\033[3mKevin's Package Manager ist drafted by and designed for:
\033[1;34mKevin Veen-Birkenbach
\033[0m\033[4mhttps://www.veen.world/\033[0m
2025-03-04 13:17:57 +01:00
2025-03-05 10:26:01 +01:00
\033[1mOverview:\033[0m
2025-03-05 10:03:20 +01:00
A configurable Python tool to manage multiple repositories via a unified command-line interface.
2025-03-05 14:26:43 +01:00
This tool automates common Git operations (clone, pull, push, status, etc.) and creates executable wrappers and custom aliases to simplify your workflow.
2025-03-05 10:03:20 +01:00
2025-03-05 10:26:01 +01:00
\033[1mFeatures:\033[0m
\033[1;33mAuto-install & Setup:\033[0m Automatically detect and set up repositories.
\033[1;33mGit Command Integration:\033[0m Execute Git commands with extra parameters.
\033[1;33mExplorer & Terminal Support:\033[0m Open repositories in your file manager or a new terminal tab.
\033[1;33mComprehensive Configuration:\033[0m Manage settings via YAML files (default & user-specific).
2025-03-04 13:17:57 +01:00
2025-03-05 10:03:20 +01:00
For detailed help on each command, use:
2025-03-05 10:26:01 +01:00
\033[1m pkgmgr <command> --help\033[0m
2025-03-05 10:03:20 +01:00
"""
2025-03-05 10:26:01 +01:00
2025-03-05 10:03:20 +01:00
parser = argparse.ArgumentParser(description=description_text,formatter_class=argparse.RawTextHelpFormatter)
subparsers = parser.add_subparsers(dest="command", help="Subcommands")
2025-03-04 13:17:57 +01:00
def add_identifier_arguments(subparser):
subparser.add_argument("identifiers", nargs="*", help="Identifier(s) for repositories")
2025-03-05 09:11:45 +01:00
subparser.add_argument("--all", action="store_true", default=False, help="Apply to all repositories in the config")
2025-03-04 13:17:57 +01:00
subparser.add_argument("--preview", action="store_true", help="Preview changes without executing commands")
subparser.add_argument("--list", action="store_true", help="List affected repositories (with preview or status)")
2025-03-04 14:32:05 +01:00
subparser.add_argument("extra_args", nargs=argparse.REMAINDER, help="Extra arguments for the git command")
2025-03-04 13:17:57 +01:00
install_parser = subparsers.add_parser("install", help="Install repository/repositories")
add_identifier_arguments(install_parser)
install_parser.add_argument("-q", "--quiet", action="store_true", help="Suppress warnings and info messages")
2025-03-05 09:11:45 +01:00
install_parser.add_argument("--no-verification", default=False, action="store_true", help="Disable verification of repository commit")
2025-03-04 13:17:57 +01:00
deinstall_parser = subparsers.add_parser("deinstall", help="Deinstall repository/repositories")
add_identifier_arguments(deinstall_parser)
delete_parser = subparsers.add_parser("delete", help="Delete repository directory for repository/repositories")
add_identifier_arguments(delete_parser)
update_parser = subparsers.add_parser("update", help="Update (pull + install) repository/repositories")
add_identifier_arguments(update_parser)
update_parser.add_argument("--system", action="store_true", help="Include system update commands")
update_parser.add_argument("-q", "--quiet", action="store_true", help="Suppress warnings and info messages")
2025-03-05 09:11:45 +01:00
update_parser.add_argument("--no-verification", action="store_true", default=False, help="Disable verification of repository commit")
2025-03-04 13:17:57 +01:00
status_parser = subparsers.add_parser("status", help="Show status for repository/repositories or system")
add_identifier_arguments(status_parser)
status_parser.add_argument("--system", action="store_true", help="Show system status")
2025-03-04 14:11:56 +01:00
config_parser = subparsers.add_parser("config", help="Manage configuration")
config_subparsers = config_parser.add_subparsers(dest="subcommand", help="Config subcommands", required=True)
config_show = config_subparsers.add_parser("show", help="Show configuration")
add_identifier_arguments(config_show)
config_add = config_subparsers.add_parser("add", help="Interactively add a new repository entry")
config_edit = config_subparsers.add_parser("edit", help="Edit configuration file with nano")
2025-03-04 15:08:25 +01:00
config_init_parser = config_subparsers.add_parser("init", help="Initialize user configuration by scanning the base directory")
2025-03-04 15:58:37 +01:00
config_delete = config_subparsers.add_parser("delete", help="Delete repository entry from user config")
add_identifier_arguments(config_delete)
config_ignore = config_subparsers.add_parser("ignore", help="Set ignore flag for repository entries in user config")
add_identifier_arguments(config_ignore)
config_ignore.add_argument("--set", choices=["true", "false"], required=True, help="Set ignore to true or false")
2025-03-04 17:04:33 +01:00
path_parser = subparsers.add_parser("path", help="Print the path(s) of repository/repositories")
add_identifier_arguments(path_parser)
2025-03-05 09:33:16 +01:00
explor_parser = subparsers.add_parser("explor", help="Open repository in Nautilus file manager")
add_identifier_arguments(explor_parser)
terminal_parser = subparsers.add_parser("terminal", help="Open repository in a new GNOME Terminal tab")
add_identifier_arguments(terminal_parser)
2025-03-05 11:20:59 +01:00
code_parser = subparsers.add_parser("code", help="Open repository workspace with VS Code")
add_identifier_arguments(code_parser)
2025-03-05 10:03:20 +01:00
list_parser = subparsers.add_parser("list", help="List all repositories with details and status")
list_parser.add_argument("--search", default="", help="Filter repositories that contain the given string")
list_parser.add_argument("--status", type=str, default="", help="Filter repositories by status (case insensitive)")
2025-03-05 09:11:45 +01:00
# Proxies the default git commands
for git_command in GIT_DEFAULT_COMMANDS:
add_identifier_arguments(
2025-03-05 11:48:32 +01:00
subparsers.add_parser(
git_command,
help=f"Proxies 'git {git_command}' to one repository/repositories",
description=f"Executes 'git {git_command}' for the identified repos.\nTo recieve more help execute 'git {git_command} --help'",
formatter_class=argparse.RawTextHelpFormatter
)
2025-03-05 09:11:45 +01:00
)
2025-03-04 13:17:57 +01:00
2025-03-04 14:11:56 +01:00
args = parser.parse_args()
2025-03-04 13:43:23 +01:00
2025-03-04 15:41:39 +01:00
# Dispatch commands.
2025-03-04 14:11:56 +01:00
if args.command == "install":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
install_repos(selected,repositories_base_dir, BIN_DIR, all_repos_list, args.no_verification, preview=args.preview, quiet=args.quiet)
2025-03-05 09:11:45 +01:00
elif args.command in GIT_DEFAULT_COMMANDS:
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
2025-03-05 09:11:45 +01:00
if args.command == "clone":
2025-03-05 11:20:59 +01:00
clone_repos(selected, repositories_base_dir, all_repos_list, args.preview)
2025-03-05 09:11:45 +01:00
else:
2025-03-05 11:20:59 +01:00
git_default_exec(selected, repositories_base_dir, all_repos_list, args.extra_args, args.command, preview=args.preview)
2025-03-05 10:03:20 +01:00
elif args.command == "list":
2025-03-05 11:20:59 +01:00
list_repositories(all_repos_list, repositories_base_dir, BIN_DIR, search_filter=args.search, status_filter=args.status)
2025-03-04 13:17:57 +01:00
elif args.command == "deinstall":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
deinstall_repos(selected,repositories_base_dir, BIN_DIR, all_repos_list, preview=args.preview)
2025-03-04 13:17:57 +01:00
elif args.command == "delete":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
delete_repos(selected,repositories_base_dir, all_repos_list, preview=args.preview)
2025-03-04 13:17:57 +01:00
elif args.command == "update":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
update_repos(selected,repositories_base_dir, BIN_DIR, all_repos_list, args.no_verification, system_update=args.system, preview=args.preview, quiet=args.quiet)
2025-03-04 13:17:57 +01:00
elif args.command == "status":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
status_repos(selected,repositories_base_dir, all_repos_list, args.extra_args, list_only=args.list, system_status=args.system, preview=args.preview)
2025-03-05 09:33:16 +01:00
elif args.command == "explor":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all, all_repos_list, args.identifiers)
2025-03-05 09:33:16 +01:00
for repo in selected:
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir, repo)
2025-03-05 09:33:16 +01:00
run_command(f"nautilus {repo_dir}")
2025-03-05 11:20:59 +01:00
elif args.command == "code":
2025-03-05 11:50:54 +01:00
selected = get_selected_repos(args.all, all_repos_list, args.identifiers)
2025-03-05 11:20:59 +01:00
if not selected:
print("No repositories selected.")
else:
identifiers = [get_repo_identifier(repo, all_repos_list) for repo in selected]
sorted_identifiers = sorted(identifiers)
workspace_name = "_".join(sorted_identifiers) + ".code-workspace"
workspaces_dir = os.path.expanduser(config_merged.get("directories").get("workspaces"))
os.makedirs(workspaces_dir, exist_ok=True)
workspace_file = os.path.join(workspaces_dir, workspace_name)
folders = []
for repo in selected:
repo_dir = os.path.expanduser(get_repo_dir(repositories_base_dir, repo))
folders.append({"path": repo_dir})
workspace_data = {
"folders": folders,
"settings": {}
}
2025-03-05 13:54:19 +01:00
if not os.path.exists(workspace_file):
with open(workspace_file, "w") as f:
json.dump(workspace_data, f, indent=4)
print(f"Created workspace file: {workspace_file}")
else:
print(f"Using existing workspace file: {workspace_file}")
2025-03-05 11:20:59 +01:00
run_command(f'code "{workspace_file}"')
2025-03-05 09:33:16 +01:00
elif args.command == "terminal":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all, all_repos_list, args.identifiers)
2025-03-05 09:33:16 +01:00
for repo in selected:
2025-03-05 11:20:59 +01:00
repo_dir = get_repo_dir(repositories_base_dir, repo)
2025-03-05 09:33:16 +01:00
run_command(f'gnome-terminal --tab --working-directory="{repo_dir}"')
2025-03-04 17:04:33 +01:00
elif args.command == "path":
2025-03-05 11:48:32 +01:00
selected = get_selected_repos(args.all,all_repos_list,args.identifiers)
2025-03-04 17:04:33 +01:00
paths = [
2025-03-05 11:20:59 +01:00
get_repo_dir(repositories_base_dir,repo)
2025-03-04 17:04:33 +01:00
for repo in selected
]
print(" ".join(paths))
2025-03-04 14:11:56 +01:00
elif args.command == "config":
if args.subcommand == "show":
if args.all or (not args.identifiers):
show_config([], full_config=True)
else:
selected = resolve_repos(args.identifiers, all_repos_list)
if selected:
show_config(selected, full_config=False)
elif args.subcommand == "add":
2025-03-04 15:41:39 +01:00
interactive_add(config_merged)
2025-03-04 14:11:56 +01:00
elif args.subcommand == "edit":
edit_config()
2025-03-04 15:08:25 +01:00
elif args.subcommand == "init":
if os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, 'r') as f:
user_config = yaml.safe_load(f) or {}
else:
2025-03-05 11:20:59 +01:00
user_config = {"repositories": []}
2025-03-04 15:41:39 +01:00
config_init(user_config, config_merged, BIN_DIR)
2025-03-04 15:58:37 +01:00
elif args.subcommand == "delete":
# Load user config from USER_CONFIG_PATH.
if os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, 'r') as f:
2025-03-05 11:20:59 +01:00
user_config = yaml.safe_load(f) or {"repositories": []}
2025-03-04 15:58:37 +01:00
else:
2025-03-05 11:20:59 +01:00
user_config = {"repositories": []}
2025-03-04 15:58:37 +01:00
if args.all or not args.identifiers:
print("You must specify identifiers to delete.")
else:
2025-03-05 11:20:59 +01:00
to_delete = resolve_repos(args.identifiers, user_config.get("repositories", []))
new_repos = [entry for entry in user_config.get("repositories", []) if entry not in to_delete]
user_config["repositories"] = new_repos
2025-03-04 15:58:37 +01:00
save_user_config(user_config)
print(f"Deleted {len(to_delete)} entries from user config.")
elif args.subcommand == "ignore":
# Load user config from USER_CONFIG_PATH.
if os.path.exists(USER_CONFIG_PATH):
with open(USER_CONFIG_PATH, 'r') as f:
2025-03-05 11:20:59 +01:00
user_config = yaml.safe_load(f) or {"repositories": []}
2025-03-04 15:58:37 +01:00
else:
2025-03-05 11:20:59 +01:00
user_config = {"repositories": []}
2025-03-04 15:58:37 +01:00
if args.all or not args.identifiers:
print("You must specify identifiers to modify ignore flag.")
else:
2025-03-05 11:20:59 +01:00
to_modify = resolve_repos(args.identifiers, user_config.get("repositories", []))
for entry in user_config["repositories"]:
2025-03-04 15:58:37 +01:00
key = (entry.get("provider"), entry.get("account"), entry.get("repository"))
for mod in to_modify:
mod_key = (mod.get("provider"), mod.get("account"), mod.get("repository"))
if key == mod_key:
entry["ignore"] = (args.set == "true")
print(f"Set ignore for {key} to {entry['ignore']}")
save_user_config(user_config)
2025-03-04 13:17:57 +01:00
else:
parser.print_help()