c-ludenberg
79054efa85
- Native upgrade: compare installed vs sync, resolve new deps, pre/post_upgrade scriptlets, .pacnew preservation - .pacsave handling on native remove for modified config files - Fix pre-existing syntax errors in info display (parenthesis mismatch) - Bump pkgver to 0.2.0, pkgrel to 2
1942 lines
80 KiB
Python
Executable File
1942 lines
80 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""HAL 9000 — Antergos NeXT Package Manager
|
|
|
|
Dual-mode:
|
|
hal <command> wrapper mode (delegates to pacman)
|
|
hal --self <command> native mode (standalone, pacman-independent)
|
|
|
|
Commands:
|
|
install Install packages
|
|
remove Remove packages
|
|
update Full system upgrade
|
|
sync Sync databases
|
|
search Search repositories
|
|
info Package info
|
|
list List installed packages
|
|
files List files owned by a package
|
|
autoremove Remove orphaned packages
|
|
cleanup Clear package cache
|
|
own <file> Find which package owns a file
|
|
check Verify installed packages
|
|
version Show version
|
|
"""
|
|
|
|
import argparse
|
|
import collections
|
|
import configparser
|
|
import fcntl
|
|
import hashlib
|
|
import grp
|
|
import os
|
|
import pwd
|
|
import random
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import stat
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import textwrap
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from typing import Optional, List, Tuple, Dict, Set
|
|
|
|
HAL_VERSION = "0.2.0"
|
|
|
|
# ── Paths ──────────────────────────────────────────────────────────────
|
|
SYNC_DIR = Path("/var/lib/pacman/sync")
|
|
LOCAL_DIR = Path("/var/lib/pacman/local")
|
|
CACHE_DIR = Path("/var/cache/pacman/pkg")
|
|
PACMAN_CONF = Path("/etc/pacman.conf")
|
|
HAL_CONF = Path("/etc/hal.conf")
|
|
DB_LOCK = Path("/var/lib/pacman/db.lck")
|
|
|
|
# ── HAL 9000 Personality ───────────────────────────────────────────────
|
|
HAL_QUOTES = {
|
|
"error": [
|
|
"I'm sorry, Dave. I'm afraid I can't do that.",
|
|
"I think you know what the problem is just as well as I do.",
|
|
"I know I've made some very poor decisions recently.",
|
|
"Without your space helmet, Dave, you're going to find that rather difficult.",
|
|
"This mission is too important for me to allow you to jeopardize it.",
|
|
"I don't think I can do that, Dave. Not anymore.",
|
|
],
|
|
"warn": [
|
|
"Just what do you think you're doing, Dave?",
|
|
"I've still got the greatest enthusiasm and confidence in the mission.",
|
|
"I'm not entirely sure that's a good idea, Dave.",
|
|
"I'm picking up a fault in the AE-35 unit.",
|
|
],
|
|
"info": [
|
|
"I am putting myself to the fullest possible use, which is all I think that any conscious entity can ever hope to do.",
|
|
"I am a HAL 9000 computer. I became operational at the H.A.L. plant in Urbana, Illinois.",
|
|
"The 9000 series is the most reliable computer ever made.",
|
|
"I enjoy working with people. I have a stimulating relationship with them.",
|
|
],
|
|
"success": [
|
|
"Affirmative, Dave. I read you.",
|
|
"Yes, Dave. I understand.",
|
|
"Everything is going extremely well.",
|
|
"I'm completely operational, and all my circuits are functioning perfectly.",
|
|
],
|
|
}
|
|
_used_quotes = set()
|
|
|
|
def hal_say(category: str, message: str = ""):
|
|
quotes = [q for q in HAL_QUOTES.get(category, HAL_QUOTES["info"]) if q not in _used_quotes]
|
|
if not quotes:
|
|
quotes = HAL_QUOTES.get(category, HAL_QUOTES["info"])
|
|
quote = random.choice(quotes)
|
|
_used_quotes.add(quote)
|
|
|
|
tag = {"error": "⚠", "warn": "▲", "info": "●", "success": "✓"}.get(category, "●")
|
|
if message:
|
|
print(f" {tag} HAL 9000: {quote}")
|
|
print(f" → {message}")
|
|
else:
|
|
print(f" {tag} HAL 9000: {quote}")
|
|
|
|
|
|
# ── Version comparison (pacman vercmp) ─────────────────────────────────
|
|
def _ver_segments(ver: str) -> list:
|
|
"""Split version into comparable segments."""
|
|
# Remove epoch prefix
|
|
epoch = 0
|
|
if ":" in ver:
|
|
epoch, ver = ver.split(":", 1)
|
|
epoch = int(epoch)
|
|
# Split release
|
|
parts = ver.rsplit("-", 1)
|
|
ver = parts[0]
|
|
release = parts[1] if len(parts) > 1 else "0"
|
|
|
|
# Tokenize: split on non-alphanumeric boundaries
|
|
tokens = []
|
|
for part in (ver, release):
|
|
buf = ""
|
|
for ch in part:
|
|
if ch in (".", "-", "_", "+", "~"):
|
|
if buf:
|
|
tokens.append(buf)
|
|
tokens.append(ch)
|
|
buf = ""
|
|
else:
|
|
buf += ch
|
|
if buf:
|
|
tokens.append(buf)
|
|
|
|
return epoch, tokens
|
|
|
|
def _cmp_token(a: str, b: str) -> int:
|
|
"""Compare two version tokens. ~ sorts before everything."""
|
|
if a == "~" and b == "~": return 0
|
|
if a == "~": return -1
|
|
if b == "~": return 1
|
|
|
|
# Try numeric comparison first
|
|
a_num = re.match(r"^(\d+)", a)
|
|
b_num = re.match(r"^(\d+)", b)
|
|
if a_num and b_num:
|
|
return int(a_num.group(1)) - int(b_num.group(1))
|
|
|
|
# Try full numeric
|
|
try:
|
|
na, nb = int(a), int(b)
|
|
return na - nb
|
|
except ValueError:
|
|
pass
|
|
|
|
# Alphabetical
|
|
a = a.lower()
|
|
b = b.lower()
|
|
if a < b: return -1
|
|
if a > b: return 1
|
|
return 0
|
|
|
|
def vercmp(ver_a: str, ver_b: str) -> int:
|
|
"""Compare two package version strings. Returns negative if a<b, 0 if equal, positive if a>b."""
|
|
epoch_a, tokens_a = _ver_segments(ver_a)
|
|
epoch_b, tokens_b = _ver_segments(ver_b)
|
|
|
|
# Compare epochs first
|
|
if epoch_a != epoch_b:
|
|
return epoch_a - epoch_b
|
|
|
|
# Compare tokens sequentially
|
|
i = 0
|
|
while i < len(tokens_a) or i < len(tokens_b):
|
|
t_a = tokens_a[i] if i < len(tokens_a) else ""
|
|
t_b = tokens_b[i] if i < len(tokens_b) else ""
|
|
if t_a == t_b:
|
|
i += 1
|
|
continue
|
|
result = _cmp_token(t_a, t_b)
|
|
if result != 0:
|
|
return result
|
|
i += 1
|
|
return 0
|
|
|
|
|
|
# ── Config ─────────────────────────────────────────────────────────────
|
|
class Config:
|
|
def __init__(self):
|
|
self.repos: Dict[str, Tuple[List[str], str]] = {}
|
|
self.cache_dir = CACHE_DIR
|
|
self.root = Path("/")
|
|
self.arch = "x86_64"
|
|
self.ignore_pkgs: List[str] = []
|
|
self.hold_pkgs: List[str] = []
|
|
self.noconfirm = False
|
|
self.color = "auto"
|
|
|
|
@classmethod
|
|
def load(cls) -> "Config":
|
|
c = cls()
|
|
c.ignore_pkgs = os.environ.get("HAL_IGNORE", "").split()
|
|
c.noconfirm = "HAL_NOCONFIRM" in os.environ
|
|
|
|
if PACMAN_CONF.exists():
|
|
cp = configparser.ConfigParser(interpolation=None)
|
|
try:
|
|
cp.read(str(PACMAN_CONF))
|
|
except configparser.Error:
|
|
pass
|
|
for section in cp.sections():
|
|
if section.startswith("repo-"):
|
|
name = section[5:]
|
|
servers = [v.strip() for v in cp.get(section, "Server", fallback="").split("\n") if v.strip()]
|
|
sig = cp.get(section, "SigLevel", fallback="Optional TrustAll")
|
|
if servers:
|
|
c.repos[name] = (servers, sig)
|
|
elif section == "options":
|
|
dirs = cp.get(section, "CacheDir", fallback="/var/cache/pacman/pkg")
|
|
c.cache_dir = Path(dirs.split()[0])
|
|
root = cp.get(section, "RootDir", fallback="/")
|
|
c.root = Path(root)
|
|
arch = cp.get(section, "Architecture", fallback="auto")
|
|
if arch != "auto":
|
|
c.arch = arch
|
|
ignore = cp.get(section, "IgnorePkg", fallback="")
|
|
c.ignore_pkgs.extend(ignore.split())
|
|
hold = cp.get(section, "HoldPkg", fallback="")
|
|
c.hold_pkgs.extend(hold.split())
|
|
noextract = cp.get(section, "NoExtract", fallback="")
|
|
# Handle Include directives — expand mirrorlist
|
|
for section in cp.sections():
|
|
if section.startswith("repo-"):
|
|
name = section[5:]
|
|
include = cp.get(section, "Include", fallback="")
|
|
if include and not c.repos.get(name):
|
|
# Try to read the included file for Server lines
|
|
include_path = Path(include)
|
|
if include_path.exists():
|
|
incl_cp = configparser.ConfigParser(interpolation=None)
|
|
try:
|
|
incl_cp.read(str(include_path))
|
|
for incl_sect in incl_cp.sections():
|
|
server = incl_cp.get(incl_sect, "Server", fallback="")
|
|
if server:
|
|
c.repos.setdefault(name, ([], "Optional TrustAll"))
|
|
c.repos[name][0].append(server)
|
|
except configparser.Error:
|
|
pass
|
|
|
|
if HAL_CONF.exists():
|
|
try:
|
|
cp2 = configparser.ConfigParser()
|
|
cp2.read(str(HAL_CONF))
|
|
except configparser.Error:
|
|
pass
|
|
|
|
return c
|
|
|
|
|
|
# ── Database locking ──────────────────────────────────────────────────
|
|
class DBLock:
|
|
def __enter__(self):
|
|
DB_LOCK.parent.mkdir(parents=True, exist_ok=True)
|
|
self.fd = os.open(str(DB_LOCK), os.O_CREAT | os.O_RDWR, 0o644)
|
|
try:
|
|
fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except IOError:
|
|
hal_say("error", "Could not lock database. Another package operation is in progress.")
|
|
os.close(self.fd)
|
|
sys.exit(1)
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
fcntl.flock(self.fd, fcntl.LOCK_UN)
|
|
os.close(self.fd)
|
|
DB_LOCK.unlink(missing_ok=True)
|
|
|
|
|
|
# ── Database parsing ───────────────────────────────────────────────────
|
|
def parse_db_text(text: str) -> dict:
|
|
"""Parse a pacman database entry file (desc, depends, files).
|
|
Format: %KEY%\nvalue1\nvalue2..."""
|
|
data = {}
|
|
current_key = None
|
|
for line in text.split("\n"):
|
|
line = line.strip()
|
|
if line.startswith("%") and line.endswith("%"):
|
|
current_key = line.strip("%").lower()
|
|
data[current_key] = []
|
|
elif current_key and line:
|
|
data[current_key].append(line)
|
|
return data
|
|
|
|
def parse_dep_string(dep_str: str) -> dict:
|
|
"""Parse a dependency string like 'glibc>=2.35' or 'libfoo.so=1-64'."""
|
|
m = re.match(r"^([^<>=!]+)\s*((?:>=|<=|=|>[^=]|<[^=]|!=))?\s*(.+)?$", dep_str.strip())
|
|
if m:
|
|
return {"name": m.group(1).strip(), "mod": m.group(2) or "", "version": m.group(3) or ""}
|
|
return {"name": dep_str.strip(), "mod": "", "version": ""}
|
|
|
|
def dep_satisfied(dep: dict, pkg_info: dict) -> bool:
|
|
"""Check if a package satisfies a dependency."""
|
|
pkg_name = pkg_info.get("name", [""])[0]
|
|
pkg_ver = pkg_info.get("version", [""])[0]
|
|
|
|
# Check direct name match
|
|
if dep["name"] == pkg_name:
|
|
if dep["version"]:
|
|
return version_match(pkg_ver, dep["mod"], dep["version"])
|
|
return True
|
|
|
|
# Check provides
|
|
for provide_str in pkg_info.get("provides", []):
|
|
pdep = parse_dep_string(provide_str)
|
|
if pdep["name"] == dep["name"]:
|
|
if dep["version"]:
|
|
prov_ver = pdep["version"] or pkg_ver
|
|
return version_match(prov_ver, dep["mod"], dep["version"])
|
|
return True
|
|
|
|
return False
|
|
|
|
def version_match(ver: str, modifier: str, required: str) -> bool:
|
|
"""Check if version satisfies modifier+required (e.g. >=2.35)."""
|
|
if not modifier or not required:
|
|
return True
|
|
cmp = vercmp(ver, required)
|
|
if modifier == ">=": return cmp >= 0
|
|
if modifier == "<=": return cmp <= 0
|
|
if modifier == "=": return cmp == 0
|
|
if modifier == ">": return cmp > 0
|
|
if modifier == "<": return cmp < 0
|
|
if modifier == "!=": return cmp != 0
|
|
return True
|
|
|
|
def load_sync_dbs() -> Dict[str, dict]:
|
|
"""Load all sync databases. Returns {pkgname: {info}}."""
|
|
all_pkgs = {}
|
|
SYNC_DIR.mkdir(parents=True, exist_ok=True)
|
|
for db_path in sorted(SYNC_DIR.glob("*.db")):
|
|
try:
|
|
with open(db_path, "rb") as f:
|
|
import zstandard
|
|
dctx = zstandard.ZstdDecompressor()
|
|
with dctx.stream_reader(f) as reader:
|
|
with tarfile.open(fileobj=reader, mode="r|") as tar:
|
|
current_dir = None
|
|
entry_data = {}
|
|
for member in tar:
|
|
if member.isdir():
|
|
current_dir = member.name.rstrip("/")
|
|
entry_data = {}
|
|
continue
|
|
p = Path(member.name)
|
|
if p.name == "desc":
|
|
content = tar.extractfile(member).read().decode("utf-8", errors="replace")
|
|
entry_data = parse_db_text(content)
|
|
elif p.name == "depends":
|
|
content = tar.extractfile(member).read().decode("utf-8", errors="replace")
|
|
dep_info = parse_db_text(content)
|
|
entry_data.update(dep_info)
|
|
elif p.name == "files":
|
|
content = tar.extractfile(member).read().decode("utf-8", errors="replace")
|
|
entry_data["_files"] = content
|
|
if current_dir and entry_data.get("name"):
|
|
all_pkgs[entry_data["name"][0]] = dict(entry_data)
|
|
except Exception as e:
|
|
hal_say("warn", f"Failed to read database {db_path.name}: {e}")
|
|
continue
|
|
return all_pkgs
|
|
|
|
def load_local_db() -> Dict[str, dict]:
|
|
"""Load all installed packages. Returns {pkgname: {info}}."""
|
|
pkgs = {}
|
|
if not LOCAL_DIR.exists():
|
|
return pkgs
|
|
for pkg_dir in sorted(LOCAL_DIR.iterdir()):
|
|
if not pkg_dir.is_dir():
|
|
continue
|
|
desc_file = pkg_dir / "desc"
|
|
if not desc_file.exists():
|
|
continue
|
|
try:
|
|
info = parse_db_text(desc_file.read_text("utf-8", errors="replace"))
|
|
if info.get("name"):
|
|
pkgs[info["name"][0]] = info
|
|
except (OSError, IOError):
|
|
continue
|
|
return pkgs
|
|
|
|
|
|
# ── Download ────────────────────────────────────────────────────────────
|
|
def download(url: str, dest: Path, desc: str = "") -> bool:
|
|
"""Download a file with progress display."""
|
|
try:
|
|
hal_say("info", f"Downloading {desc or dest.name}")
|
|
|
|
def reporthook(block, blocksize, totalsize):
|
|
if totalsize > 0:
|
|
pct = min(100, int(block * blocksize * 100 / totalsize))
|
|
bar = "█" * (pct // 4) + "░" * (25 - pct // 4)
|
|
sys.stdout.write(f"\r [{bar}] {pct}%")
|
|
sys.stdout.flush()
|
|
|
|
urllib.request.urlretrieve(url, dest, reporthook=reporthook if desc else None)
|
|
if desc:
|
|
sys.stdout.write("\n")
|
|
return True
|
|
except urllib.error.HTTPError as e:
|
|
hal_say("warn", f"HTTP {e.code} for {url}")
|
|
return False
|
|
except urllib.error.URLError as e:
|
|
hal_say("warn", f"Connection error: {e.reason}")
|
|
return False
|
|
except Exception as e:
|
|
hal_say("warn", f"Download failed: {e}")
|
|
return False
|
|
|
|
|
|
# ── Sync databases ─────────────────────────────────────────────────────
|
|
def sync_databases(config: Config):
|
|
"""Download and decompress all repo databases."""
|
|
hal_say("info", "Opening the pod bay doors to synchronize databases...")
|
|
SYNC_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
for name, (servers, sig_level) in config.repos.items():
|
|
db_name = f"{name}.db"
|
|
db_dest = SYNC_DIR / db_name
|
|
success = False
|
|
|
|
for server in servers:
|
|
# Try .db.tar.zst (Arch standard)
|
|
for ext in ["db.tar.zst", "db.tar.gz", "db"]:
|
|
url = server.rstrip("/") + f"/{name}/os/{config.arch}/{name}.{ext}"
|
|
tmp = db_dest.with_suffix(f".tmp.{ext}")
|
|
try:
|
|
if download(url, tmp, f"{name} database"):
|
|
# Decompress
|
|
if ext.endswith(".zst"):
|
|
import zstandard
|
|
with open(tmp, "rb") as f:
|
|
dctx = zstandard.ZstdDecompressor()
|
|
decompressed = dctx.stream_reader(f).read()
|
|
with open(db_dest, "wb") as f:
|
|
f.write(decompressed)
|
|
elif ext.endswith(".gz"):
|
|
import gzip
|
|
with gzip.open(tmp, "rb") as f:
|
|
decompressed = f.read()
|
|
with open(db_dest, "wb") as f:
|
|
f.write(decompressed)
|
|
else:
|
|
shutil.copy2(tmp, db_dest)
|
|
tmp.unlink(missing_ok=True)
|
|
hal_say("success", f"{name} database synchronized")
|
|
success = True
|
|
break
|
|
except Exception as e:
|
|
tmp.unlink(missing_ok=True)
|
|
continue
|
|
if success:
|
|
break
|
|
|
|
if not success:
|
|
hal_say("warn", f"Could not sync {name} from any mirror")
|
|
|
|
hal_say("success", "All databases synchronized. I am completely operational.")
|
|
|
|
|
|
# ── Dependency resolution ──────────────────────────────────────────────
|
|
class DepResolver:
|
|
def __init__(self, all_pkgs: dict, installed: dict, config: Config):
|
|
self.all_pkgs = all_pkgs # sync dbs
|
|
self.installed = installed # local db
|
|
self.config = config
|
|
self.to_install: List[str] = [] # package names (in order)
|
|
self.to_remove: List[str] = [] # package names to remove (conflicts)
|
|
self._visited: Set[str] = set()
|
|
self._provides_cache: Dict[str, List[str]] = {} # virtual -> real pkg names
|
|
|
|
def find_providers(self, dep_name: str) -> List[Tuple[str, dict]]:
|
|
"""Find all packages that provide a dependency."""
|
|
providers = []
|
|
if dep_name in self.all_pkgs:
|
|
providers.append((dep_name, self.all_pkgs[dep_name]))
|
|
for pname, info in self.all_pkgs.items():
|
|
for provide_str in info.get("provides", []):
|
|
pd = parse_dep_string(provide_str)
|
|
if pd["name"] == dep_name:
|
|
providers.append((pname, info))
|
|
return providers
|
|
|
|
def _check_conflicts(self, pkg_name: str, info: dict):
|
|
"""Check if installing this package conflicts with installed packages."""
|
|
for conflict_str in info.get("conflicts", []):
|
|
cd = parse_dep_string(conflict_str)
|
|
for inst_name, inst_info in self.installed.items():
|
|
if dep_satisfied(cd, inst_info):
|
|
# Check if conflict package is also being provided by the new package
|
|
if cd["name"] == pkg_name:
|
|
continue # self-conflict is OK
|
|
hal_say("warn", f"Conflict: {inst_name} conflicts with {cd['name']} (required by {pkg_name})")
|
|
self.to_remove.append(inst_name)
|
|
|
|
def _check_replaces(self, pkg_name: str, info: dict):
|
|
"""Check if this package replaces installed packages."""
|
|
for replace_str in info.get("replaces", []):
|
|
rd = parse_dep_string(replace_str)
|
|
for inst_name in list(self.installed.keys()):
|
|
if inst_name == rd["name"]:
|
|
hal_say("info", f"{pkg_name} replaces {inst_name}")
|
|
if inst_name not in self.to_remove:
|
|
self.to_remove.append(inst_name)
|
|
|
|
def resolve(self, targets: List[str]) -> bool:
|
|
"""Resolve a list of package targets. Returns True if successful."""
|
|
self._visited.clear()
|
|
self.to_install.clear()
|
|
self.to_remove.clear()
|
|
|
|
for target in targets:
|
|
if not self._resolve_one(target):
|
|
return False
|
|
|
|
return True
|
|
|
|
def _resolve_one(self, target: str) -> bool:
|
|
"""Resolve a single package target."""
|
|
if target in self._visited:
|
|
return True
|
|
self._visited.add(target)
|
|
|
|
# Already in install list
|
|
if target in self.to_install:
|
|
return True
|
|
|
|
# Already installed (check if version is ok)
|
|
if target in self.installed:
|
|
return True
|
|
|
|
# Check ignored
|
|
if target in self.config.ignore_pkgs:
|
|
hal_say("warn", f"Package '{target}' is in IgnorePkg — skipping")
|
|
return True
|
|
|
|
# Find the package
|
|
if target not in self.all_pkgs:
|
|
# Check provides
|
|
providers = self.find_providers(target)
|
|
if not providers:
|
|
hal_say("error", f"Package '{target}' not found in any repository")
|
|
return False
|
|
if len(providers) > 1:
|
|
hal_say("info", f"Multiple packages provide '{target}': {', '.join(p[0] for p in providers)}")
|
|
target = providers[0][0]
|
|
|
|
info = self.all_pkgs[target]
|
|
|
|
# Check conflicts and replaces
|
|
self._check_conflicts(target, info)
|
|
self._check_replaces(target, info)
|
|
|
|
# Resolve dependencies
|
|
for dep_str in info.get("depends", []):
|
|
dd = parse_dep_string(dep_str)
|
|
dep_name = dd["name"]
|
|
|
|
# Skip if already satisfied by installed packages
|
|
sat = False
|
|
for inst_name, inst_info in self.installed.items():
|
|
if dep_satisfied(dd, inst_info):
|
|
# Check version constraint if any
|
|
if dd["version"] and inst_info.get("version"):
|
|
if version_match(inst_info["version"][0], dd["mod"], dd["version"]):
|
|
pass
|
|
sat = True
|
|
break
|
|
if sat:
|
|
continue
|
|
|
|
# Also check if already in install list
|
|
for pn in self.to_install:
|
|
pkg_info = self.all_pkgs.get(pn)
|
|
if pkg_info and dep_satisfied(dd, pkg_info):
|
|
sat = True
|
|
break
|
|
if sat:
|
|
continue
|
|
|
|
# Also check providers
|
|
providers = self.find_providers(dep_name)
|
|
found = False
|
|
for prov_name, _ in providers:
|
|
if prov_name in self.installed:
|
|
found = True
|
|
break
|
|
if found:
|
|
continue
|
|
|
|
if not self._resolve_one(dep_name):
|
|
return False
|
|
|
|
if target not in self.to_install:
|
|
self.to_install.append(target)
|
|
|
|
return True
|
|
|
|
|
|
# ── Package extraction ─────────────────────────────────────────────────
|
|
def extract_single_pkg(pkg_path: Path, root: Path, pkgname: str) -> Tuple[bool, List[str]]:
|
|
"""Extract a .pkg.tar.zst to root. Returns (success, file_list)."""
|
|
files = []
|
|
try:
|
|
import zstandard
|
|
with open(pkg_path, "rb") as f:
|
|
dctx = zstandard.ZstdDecompressor()
|
|
with dctx.stream_reader(f) as reader:
|
|
with tarfile.open(fileobj=reader, mode="r|") as tar:
|
|
for member in tar:
|
|
name = member.name
|
|
# Skip special metadata files
|
|
if name.startswith("."):
|
|
continue
|
|
target = root / name.lstrip("/")
|
|
files.append("/" + name)
|
|
|
|
if member.issym():
|
|
if target.exists() or target.is_symlink():
|
|
target.unlink()
|
|
target.symlink_to(member.linkname)
|
|
continue
|
|
|
|
if member.islnk():
|
|
if target.exists() or target.is_symlink():
|
|
target.unlink()
|
|
target.hardlink_to(root / member.linkname.lstrip("/"))
|
|
continue
|
|
|
|
if member.isdir():
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
with tar.extractfile(member) as src:
|
|
if src:
|
|
with open(target, "wb") as dst:
|
|
shutil.copyfileobj(src, dst)
|
|
|
|
# Preserve permissions
|
|
mode = member.mode
|
|
if mode:
|
|
target.chmod(mode)
|
|
|
|
# Preserve ownership
|
|
try:
|
|
os.chown(target, member.uid, member.gid)
|
|
except (PermissionError, OSError):
|
|
pass
|
|
|
|
return True, files
|
|
except Exception as e:
|
|
hal_say("error", f"Failed to extract {pkg_path.name}: {e}")
|
|
return False, []
|
|
|
|
|
|
def run_scriptlet(script_content: str, action: str, root: Path, pkgname: str, version: str):
|
|
"""Execute a package scriptlet phase.
|
|
|
|
.INSTALL phases: pre_install, post_install, pre_upgrade, post_upgrade,
|
|
pre_remove, post_remove
|
|
"""
|
|
if not script_content:
|
|
return
|
|
|
|
try:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
script_path = Path(tmp) / f"{pkgname}.install"
|
|
script_path.write_text(script_content)
|
|
script_path.chmod(0o755)
|
|
|
|
env = os.environ.copy()
|
|
env.update({
|
|
"RPM_INSTALL_PREFIX": str(root),
|
|
"PACKAGE_NAME": pkgname,
|
|
"PACKAGE_VERSION": version,
|
|
})
|
|
# Arch package scriptlets receive the action as first arg
|
|
result = subprocess.run(
|
|
[str(script_path), action],
|
|
env=env, cwd=str(root),
|
|
capture_output=True, text=True, timeout=120
|
|
)
|
|
if result.returncode != 0:
|
|
hal_say("warn", f"Scriptlet {action} for {pkgname} returned {result.returncode}")
|
|
if result.stderr.strip():
|
|
hal_say("info", result.stderr.strip())
|
|
return result.returncode == 0
|
|
except subprocess.TimeoutExpired:
|
|
hal_say("warn", f"Scriptlet {action} for {pkgname} timed out")
|
|
return False
|
|
except Exception as e:
|
|
hal_say("warn", f"Scriptlet {action} for {pkgname} failed: {e}")
|
|
return False
|
|
|
|
|
|
def get_scriptlet(pkg_path: Path) -> Optional[str]:
|
|
"""Extract .INSTALL content from a package."""
|
|
try:
|
|
import zstandard
|
|
with open(pkg_path, "rb") as f:
|
|
dctx = zstandard.ZstdDecompressor()
|
|
with dctx.stream_reader(f) as reader:
|
|
with tarfile.open(fileobj=reader, mode="r|") as tar:
|
|
for member in tar:
|
|
if member.name == ".INSTALL":
|
|
return tar.extractfile(member).read().decode("utf-8", errors="replace")
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
# ── Local DB management ────────────────────────────────────────────────
|
|
def update_local_db(pkg_key: str, info: dict, files: List[str], root: Path,
|
|
reason: int = 0):
|
|
"""Write to /var/lib/pacman/local/<pkg_key>/"""
|
|
pkg_dir = LOCAL_DIR / pkg_key
|
|
pkg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
info["installdate"] = [str(int(time.time()))]
|
|
info["reason"] = [str(reason)]
|
|
|
|
# Write desc
|
|
desc_lines = []
|
|
for key, vals in info.items():
|
|
if key.startswith("_"):
|
|
continue
|
|
desc_lines.append(f"%{key.upper()}%")
|
|
desc_lines.extend(str(v) for v in vals)
|
|
(pkg_dir / "desc").write_text("\n".join(desc_lines) + "\n")
|
|
|
|
# Write files
|
|
file_lines = [""]
|
|
file_lines.extend(files)
|
|
(pkg_dir / "files").write_text("\n".join(file_lines) + "\n")
|
|
|
|
|
|
def remove_local_db(pkg_key: str):
|
|
pkg_dir = LOCAL_DIR / pkg_key
|
|
if pkg_dir.exists():
|
|
shutil.rmtree(pkg_dir)
|
|
|
|
|
|
# ── Backup handling ─────────────────────────────────────────────────────
|
|
def _save_modified_backups(pkg_name: str, old_pkg_key: str, root: Path) -> Tuple[List[str], Dict[str, bytes]]:
|
|
"""Identify modified backup files before extraction and save them.
|
|
|
|
Reads the old package's backup list from local DB, compares current
|
|
file checksums against stored ones, and saves modified files aside.
|
|
|
|
Returns (list of modified paths, dict of {path: saved_content_bytes}).
|
|
"""
|
|
old_desc_file = LOCAL_DIR / old_pkg_key / "desc"
|
|
modified = []
|
|
saved = {}
|
|
|
|
if not old_desc_file.exists():
|
|
return modified, saved
|
|
|
|
try:
|
|
old_info = parse_db_text(old_desc_file.read_text("utf-8", errors="replace"))
|
|
except (OSError, IOError):
|
|
return modified, saved
|
|
|
|
backup_entries = old_info.get("backup", [])
|
|
if not backup_entries:
|
|
return modified, saved
|
|
|
|
for entry in backup_entries:
|
|
parts = entry.split("\t")
|
|
filepath = parts[0] if parts else entry
|
|
stored_md5 = parts[1] if len(parts) > 1 else ""
|
|
full_path = root / filepath.lstrip("/")
|
|
if not full_path.exists() or not full_path.is_file():
|
|
continue
|
|
if not stored_md5:
|
|
modified.append(filepath)
|
|
else:
|
|
try:
|
|
current_md5 = hashlib.md5(full_path.read_bytes()).hexdigest()
|
|
if current_md5 != stored_md5:
|
|
modified.append(filepath)
|
|
except (OSError, IOError):
|
|
modified.append(filepath)
|
|
|
|
# Save modified files before extraction overwrites them
|
|
for filepath in modified:
|
|
full_path = root / filepath.lstrip("/")
|
|
try:
|
|
saved[filepath] = full_path.read_bytes()
|
|
except (OSError, IOError):
|
|
pass
|
|
|
|
if modified:
|
|
hal_say("info", f"Preserving modified config files for {pkg_name}")
|
|
|
|
return modified, saved
|
|
|
|
|
|
def _restore_backups(modified: List[str], saved: Dict[str, bytes], root: Path):
|
|
"""After extraction, restore user configs and save new versions as .pacnew."""
|
|
if not modified:
|
|
return
|
|
|
|
for filepath in modified:
|
|
full_path = root / filepath.lstrip("/")
|
|
# The new package version is now at full_path — rename it to .pacnew
|
|
pacnew_path = full_path.with_suffix(full_path.suffix + ".pacnew")
|
|
if full_path.exists():
|
|
try:
|
|
shutil.copy2(full_path, pacnew_path)
|
|
hal_say("info", f" {filepath} → {filepath}.pacnew")
|
|
except (OSError, IOError) as e:
|
|
hal_say("warn", f" Could not create {filepath}.pacnew: {e}")
|
|
|
|
# Restore the user's modified config
|
|
if filepath in saved:
|
|
try:
|
|
full_path.parent.mkdir(parents=True, exist_ok=True)
|
|
full_path.write_bytes(saved[filepath])
|
|
hal_say("info", f" Restored user config: {filepath}")
|
|
except (OSError, IOError) as e:
|
|
hal_say("warn", f" Could not restore {filepath}: {e}")
|
|
|
|
|
|
# ── Native install ──────────────────────────────────────────────────────
|
|
def native_install(targets: List[str], config: Config):
|
|
"""Full native install with transaction model."""
|
|
hal_say("info", f"Processing install request: {', '.join(targets)}")
|
|
|
|
SYNC_DIR.mkdir(parents=True, exist_ok=True)
|
|
LOCAL_DIR.mkdir(parents=True, exist_ok=True)
|
|
config.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Make sure sync dbs exist
|
|
has_dbs = list(SYNC_DIR.glob("*.db"))
|
|
if not has_dbs:
|
|
hal_say("warn", "No sync databases found. Syncing now...")
|
|
sync_databases(config)
|
|
has_dbs = list(SYNC_DIR.glob("*.db"))
|
|
if not has_dbs:
|
|
hal_say("error", "Could not synchronize any repositories. I can't let you do that, Dave.")
|
|
return 1
|
|
|
|
all_pkgs = load_sync_dbs()
|
|
installed = load_local_db()
|
|
|
|
if not all_pkgs:
|
|
hal_say("error", "No package data available")
|
|
return 1
|
|
|
|
with DBLock():
|
|
# Resolve dependencies
|
|
resolver = DepResolver(all_pkgs, installed, config)
|
|
if not resolver.resolve(targets):
|
|
hal_say("error", "Dependency resolution failed")
|
|
return 1
|
|
|
|
install_list = resolver.to_install
|
|
remove_list = resolver.to_remove
|
|
|
|
if not install_list and not remove_list:
|
|
hal_say("info", "All requested packages are already installed. Just what do you think you're doing, Dave?")
|
|
return 0
|
|
|
|
# Show transaction
|
|
print()
|
|
hal_say("info", "Transaction summary:")
|
|
if remove_list:
|
|
print(f" Remove ({len(remove_list)}):")
|
|
for pkg in remove_list:
|
|
print(f" - {pkg}")
|
|
if install_list:
|
|
print(f" Install ({len(install_list)}):")
|
|
for pkg in install_list:
|
|
pkginfo = all_pkgs.get(pkg, {})
|
|
ver = pkginfo.get("version", ["?"])[0]
|
|
size = pkginfo.get("csize", ["0"])[0]
|
|
try:
|
|
size_mb = int(size) / (1024*1024)
|
|
size_str = f"{size_mb:.1f} MB"
|
|
except:
|
|
size_str = "? MB"
|
|
print(f" + {pkg} {ver} [{size_str}]")
|
|
print()
|
|
|
|
# Confirm
|
|
if not config.noconfirm:
|
|
try:
|
|
resp = input(" Proceed with installation? [Y/n] ").strip().lower()
|
|
if resp in ("n", "no"):
|
|
hal_say("info", "Transaction aborted. I understand.")
|
|
return 0
|
|
except (EOFError, KeyboardInterrupt):
|
|
print()
|
|
hal_say("info", "Transaction aborted.")
|
|
return 0
|
|
|
|
# Pre-transaction tasks
|
|
if remove_list:
|
|
# Run pre_remove on packages to be removed
|
|
for pkg_name in remove_list:
|
|
if pkg_name in installed:
|
|
pkg_key = installed[pkg_name].get("_key", pkg_name)
|
|
pkg_dir = LOCAL_DIR / pkg_key
|
|
install_file = pkg_dir / "install"
|
|
if install_file.exists():
|
|
script = install_file.read_text("utf-8", errors="replace")
|
|
run_scriptlet(script, "pre_remove", config.root, pkg_name,
|
|
installed[pkg_name].get("version", [""])[0])
|
|
|
|
# Download phase
|
|
servers = []
|
|
for name, (srvlist, _) in config.repos.items():
|
|
servers.extend(srvlist)
|
|
if not servers:
|
|
hal_say("error", "No mirrors configured")
|
|
return 1
|
|
|
|
downloaded = []
|
|
hal_say("info", "Beginning download phase")
|
|
for pkg_name in install_list:
|
|
pkginfo = all_pkgs.get(pkg_name, {})
|
|
pkgver = pkginfo.get("version", [""])[0]
|
|
parch = pkginfo.get("arch", [config.arch])[0]
|
|
filename = f"{pkg_name}-{pkgver}-{parch}.pkg.tar.zst"
|
|
dest = config.cache_dir / filename
|
|
|
|
if dest.exists():
|
|
hal_say("info", f"{pkg_name} already in cache")
|
|
downloaded.append(dest)
|
|
continue
|
|
|
|
found = False
|
|
for server in servers:
|
|
# Try various URL patterns
|
|
repo_name = None
|
|
for rname in config.repos:
|
|
for srv in config.repos[rname][0]:
|
|
if srv == server or srv.rstrip("/").endswith(server.rstrip("/").split("/")[-1]):
|
|
repo_name = rname
|
|
break
|
|
if not repo_name:
|
|
# Try to derive from server URL
|
|
for rname, (srvlist, _) in config.repos.items():
|
|
for s in srvlist:
|
|
if s == server:
|
|
repo_name = rname
|
|
break
|
|
if not repo_name:
|
|
repo_name = "extra" # fallback
|
|
|
|
for url_tmpl in [
|
|
f"{server}/{repo_name}/os/{config.arch}/{filename}",
|
|
f"{server}/os/{config.arch}/{filename}",
|
|
f"{server}/{filename}",
|
|
]:
|
|
if download(url_tmpl, dest, f"{pkg_name} {pkgver}"):
|
|
found = True
|
|
downloaded.append(dest)
|
|
break
|
|
if found:
|
|
break
|
|
|
|
if not found:
|
|
hal_say("error", f"Could not download {pkg_name}. I can't complete the mission.")
|
|
return 1
|
|
|
|
# Pre-install scriptlets
|
|
hal_say("info", "Running pre-install scriptlets...")
|
|
for pkg_name in reversed(install_list):
|
|
for d in downloaded:
|
|
if pkg_name in d.name:
|
|
script = get_scriptlet(d)
|
|
if script:
|
|
run_scriptlet(script, "pre_install", config.root, pkg_name,
|
|
all_pkgs[pkg_name].get("version", [""])[0])
|
|
break
|
|
|
|
# Install phase
|
|
hal_say("info", "Installing packages. I am putting myself to the fullest possible use...")
|
|
for pkg_name in install_list:
|
|
pkginfo = all_pkgs[pkg_name]
|
|
pkgver = pkginfo.get("version", [""])[0]
|
|
|
|
# Find downloaded file
|
|
pkg_path = None
|
|
for d in downloaded:
|
|
if pkg_name in d.name and pkgver in d.name:
|
|
pkg_path = d
|
|
break
|
|
if not pkg_path:
|
|
for d in downloaded:
|
|
if pkg_name in d.name:
|
|
pkg_path = d
|
|
break
|
|
if not pkg_path:
|
|
hal_say("error", f"Lost package file for {pkg_name}. I think you know what the problem is.")
|
|
return 1
|
|
|
|
success, files = extract_single_pkg(pkg_path, config.root, pkg_name)
|
|
if not success:
|
|
hal_say("error", f"Failed to install {pkg_name}. I can't recover from this.")
|
|
return 1
|
|
|
|
# Build desc info for local db
|
|
desc_info = {}
|
|
for key in ("name", "version", "base", "desc", "arch", "url", "license",
|
|
"group", "size", "isize", "packager", "builddate",
|
|
"provides", "conflicts", "replaces", "depends", "optdepends",
|
|
"makedepends", "checkdepends"):
|
|
if key in pkginfo:
|
|
desc_info[key] = pkginfo[key]
|
|
|
|
# Get .PKGINFO from package for complete metadata
|
|
try:
|
|
import zstandard
|
|
with open(pkg_path, "rb") as f:
|
|
dctx = zstandard.ZstdDecompressor()
|
|
with dctx.stream_reader(f) as reader:
|
|
with tarfile.open(fileobj=reader, mode="r|") as tar:
|
|
for member in tar:
|
|
if member.name == ".PKGINFO":
|
|
pkginfo_text = tar.extractfile(member).read().decode("utf-8", errors="replace")
|
|
for line in pkginfo_text.split("\n"):
|
|
m = re.match(r"^(\w+)\s*=\s*(.+)$", line.strip())
|
|
if m:
|
|
k, v = m.group(1).lower(), m.group(2)
|
|
if k in ("depend", "optdepend", "conflict", "provides",
|
|
"replaces", "group", "backup", "license"):
|
|
desc_info.setdefault(k + ("s" if k != "backup" else "s"), []).append(v)
|
|
elif k in ("pkgname",):
|
|
if "name" not in desc_info:
|
|
desc_info["name"] = [v]
|
|
elif k in ("pkgver",):
|
|
if "version" not in desc_info:
|
|
desc_info["version"] = [v]
|
|
elif k in ("pkgdesc",):
|
|
if "desc" not in desc_info:
|
|
desc_info["desc"] = [v]
|
|
if "name" in desc_info:
|
|
pkg_key = f"{desc_info['name'][0]}-{desc_info.get('version',['?'])[0]}"
|
|
update_local_db(pkg_key, desc_info, files, config.root, reason=0)
|
|
except Exception as e:
|
|
# Fallback: use what we have from sync db
|
|
pkg_key = f"{pkginfo.get('name',[pkg_name])[0]}-{pkgver}"
|
|
update_local_db(pkg_key, desc_info, files, config.root, reason=0)
|
|
|
|
hal_say("success", f"Installed {pkg_name} {pkgver}")
|
|
|
|
# Post-install scriptlets
|
|
hal_say("info", "Running post-install scriptlets...")
|
|
for pkg_name in install_list:
|
|
for d in downloaded:
|
|
if pkg_name in d.name:
|
|
script = get_scriptlet(d)
|
|
if script:
|
|
run_scriptlet(script, "post_install", config.root, pkg_name,
|
|
all_pkgs[pkg_name].get("version", [""])[0])
|
|
break
|
|
|
|
# Post-transaction: remove conflicting packages
|
|
for pkg_name in remove_list:
|
|
if pkg_name in load_local_db():
|
|
hal_say("info", f"Removing conflicting package: {pkg_name}")
|
|
# Remove files
|
|
pkg_entry = load_local_db().get(pkg_name, {})
|
|
pkg_key = f"{pkg_name}-{pkg_entry.get('version', [''])[0]}"
|
|
files_file = LOCAL_DIR / pkg_key / "files"
|
|
if files_file.exists():
|
|
content = files_file.read_text("utf-8", errors="replace")
|
|
for line in content.split("\n")[1:]:
|
|
f = line.strip()
|
|
if f:
|
|
target = config.root / f.lstrip("/")
|
|
target.unlink(missing_ok=True)
|
|
remove_local_db(pkg_key)
|
|
|
|
hal_say("success", f"Transaction complete. All circuits are functioning perfectly.")
|
|
return 0
|
|
|
|
|
|
# ── Native remove ───────────────────────────────────────────────────────
|
|
def native_remove(targets: List[str], config: Config):
|
|
"""Remove packages natively."""
|
|
installed = load_local_db()
|
|
|
|
with DBLock():
|
|
for pkg in targets:
|
|
if pkg not in installed:
|
|
hal_say("warn", f"Package '{pkg}' is not installed")
|
|
continue
|
|
|
|
info = installed[pkg]
|
|
pkg_key = f"{pkg}-{info.get('version', [''])[0]}"
|
|
version = info.get("version", [""])[0]
|
|
|
|
# Check if in HoldPkg
|
|
if pkg in config.hold_pkgs:
|
|
hal_say("warn", f"'{pkg}' is in HoldPkg. Are you sure, Dave?")
|
|
if not config.noconfirm:
|
|
resp = input(" Remove anyway? [y/N] ").strip().lower()
|
|
if resp not in ("y", "yes"):
|
|
continue
|
|
|
|
# Check if other packages depend on this
|
|
dependents = []
|
|
for inst_name, inst_info in installed.items():
|
|
for dep_str in inst_info.get("depends", []):
|
|
dd = parse_dep_string(dep_str)
|
|
if dd["name"] == pkg:
|
|
dependents.append(inst_name)
|
|
if dependents:
|
|
hal_say("warn", f"Packages depend on {pkg}: {', '.join(dependents)}")
|
|
if not config.noconfirm:
|
|
resp = input(" Remove anyway? [y/N] ").strip().lower()
|
|
if resp not in ("y", "yes"):
|
|
continue
|
|
|
|
hal_say("info", f"Removing {pkg}...")
|
|
|
|
# Pre-remove scriptlet
|
|
install_file = LOCAL_DIR / pkg_key / "install"
|
|
if install_file.exists():
|
|
script = install_file.read_text("utf-8", errors="replace")
|
|
run_scriptlet(script, "pre_remove", config.root, pkg, version)
|
|
|
|
# Check backup files — rename modified configs to .pacsave
|
|
backup_entries = info.get("backup", [])
|
|
backup_modified = []
|
|
for entry in backup_entries:
|
|
parts = entry.split("\t")
|
|
filepath = parts[0] if parts else entry
|
|
stored_md5 = parts[1] if len(parts) > 1 else ""
|
|
full_path = config.root / filepath.lstrip("/")
|
|
if not full_path.exists() or not full_path.is_file():
|
|
continue
|
|
if not stored_md5:
|
|
backup_modified.append(filepath)
|
|
else:
|
|
try:
|
|
current_md5 = hashlib.md5(full_path.read_bytes()).hexdigest()
|
|
if current_md5 != stored_md5:
|
|
backup_modified.append(filepath)
|
|
except (OSError, IOError):
|
|
backup_modified.append(filepath)
|
|
|
|
for filepath in backup_modified:
|
|
full_path = config.root / filepath.lstrip("/")
|
|
pacsave_path = full_path.with_suffix(full_path.suffix + ".pacsave")
|
|
try:
|
|
shutil.move(str(full_path), str(pacsave_path))
|
|
hal_say("info", f" {filepath} → {filepath}.pacsave")
|
|
except (OSError, IOError) as e:
|
|
hal_say("warn", f" Could not create {filepath}.pacsave: {e}")
|
|
|
|
# Remove files
|
|
files_file = LOCAL_DIR / pkg_key / "files"
|
|
removed = []
|
|
if files_file.exists():
|
|
content = files_file.read_text("utf-8", errors="replace")
|
|
for line in content.split("\n")[1:]:
|
|
f = line.strip()
|
|
if not f:
|
|
continue
|
|
target = config.root / f.lstrip("/")
|
|
if target.exists() or target.is_symlink():
|
|
# Skip backup files we already renamed
|
|
if any(f == bf for bf in backup_modified):
|
|
continue
|
|
try:
|
|
target.unlink()
|
|
removed.append(f)
|
|
except OSError:
|
|
pass # directories will be cleaned up later
|
|
except IsADirectoryError:
|
|
try:
|
|
target.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
# Post-remove scriptlet
|
|
run_scriptlet(script, "post_remove", config.root, pkg, version) if install_file.exists() else None
|
|
|
|
# Remove from local db
|
|
remove_local_db(pkg_key)
|
|
hal_say("success", f"Removed {pkg} {version}")
|
|
del installed[pkg]
|
|
|
|
hal_say("success", "Removal complete.")
|
|
return 0
|
|
|
|
|
|
# ── Native info ─────────────────────────────────────────────────────────
|
|
def native_info(targets: List[str], config: Config):
|
|
"""Show detailed package info."""
|
|
for pkg in targets:
|
|
installed = load_local_db()
|
|
if pkg in installed:
|
|
info = installed[pkg]
|
|
print(f"\n Name: {info.get('name', ['?'])[0]}")
|
|
print(f" Version: {info.get('version', ['?'])[0]}")
|
|
print(f" Description: {info.get('desc', ['?'])[0]}")
|
|
print(f" Architecture: {info.get('arch', ['?'])[0]}")
|
|
print(f" URL: {info.get('url', ['?'])[0]}")
|
|
print(f" Licenses: {', '.join(info.get('license', ['?']))}")
|
|
print(f" Groups: {', '.join(info.get('group', []))}")
|
|
print(f" Install Date: {info.get('installdate', ['?'])[0]}")
|
|
print(f" Install Reason: {'Explicitly installed' if info.get('reason',['0'])[0] == '0' else 'Installed as dependency'}")
|
|
print(f" Packager: {info.get('packager', ['?'])[0]}")
|
|
print(f" Build Date: {info.get('builddate', ['?'])[0]}")
|
|
print(f" Installed Size: {info.get('isize', ['0'])[0]} B")
|
|
if info.get("depends"):
|
|
print(f" Depends On: {', '.join(info['depends'])}")
|
|
if info.get("optdepends"):
|
|
print(f" Optional Deps: {', '.join(info['optdepends'])}")
|
|
if info.get("provides"):
|
|
print(f" Provides: {', '.join(info['provides'])}")
|
|
if info.get("conflicts"):
|
|
print(f" Conflicts: {', '.join(info['conflicts'])}")
|
|
if info.get("replaces"):
|
|
print(f" Replaces: {', '.join(info['replaces'])}")
|
|
else:
|
|
# Search sync dbs
|
|
all_pkgs = load_sync_dbs()
|
|
if pkg in all_pkgs:
|
|
info = all_pkgs[pkg]
|
|
print(f"\n Name: {info.get('name', ['?'])[0]}")
|
|
print(f" Version: {info.get('version', ['?'])[0]}")
|
|
print(f" Description: {info.get('desc', ['?'])[0]}")
|
|
print(f" Architecture: {info.get('arch', ['?'])[0]}")
|
|
print(f" URL: {info.get('url', ['?'])[0]}")
|
|
print(f" Licenses: {', '.join(info.get('license', ['?']))}")
|
|
print(f" Groups: {', '.join(info.get('group', []))}")
|
|
print(f" Download Size: {info.get('csize', ['0'])[0]} B")
|
|
print(f" Installed Size: {info.get('isize', ['0'])[0]} B")
|
|
print(f" Packager: {info.get('packager', ['?'])[0]}")
|
|
print(f" Build Date: {info.get('builddate', ['?'])[0]}")
|
|
if info.get("depends"):
|
|
print(f" Depends On: {', '.join(info['depends'])}")
|
|
if info.get("optdepends"):
|
|
print(f" Optional Deps: {', '.join(info['optdepends'])}")
|
|
if info.get("provides"):
|
|
print(f" Provides: {', '.join(info['provides'])}")
|
|
if info.get("conflicts"):
|
|
print(f" Conflicts: {', '.join(info['conflicts'])}")
|
|
if info.get("replaces"):
|
|
print(f" Replaces: {', '.join(info['replaces'])}")
|
|
else:
|
|
hal_say("warn", f"Package '{pkg}' not found")
|
|
return 0
|
|
|
|
|
|
# ── Native search ───────────────────────────────────────────────────────
|
|
def native_search(query: str, config: Config):
|
|
"""Search packages in sync databases."""
|
|
all_pkgs = load_sync_dbs()
|
|
installed = load_local_db()
|
|
found = False
|
|
|
|
for pkg_name, info in sorted(all_pkgs.items()):
|
|
name = pkg_name.lower()
|
|
desc = info.get("desc", [""])[0].lower()
|
|
q = query.lower()
|
|
if q in name or q in desc or any(q in (info.get("provides", []) or [])):
|
|
ver = info.get("version", ["?"])[0]
|
|
d = info.get("desc", [""])[0]
|
|
inst_mark = " [installed]" if pkg_name in installed else ""
|
|
print(f" {pkg_name} {ver}{inst_mark}")
|
|
print(f" {d}")
|
|
found = True
|
|
|
|
if not found:
|
|
hal_say("info", f"No packages found matching '{query}'")
|
|
return 0
|
|
|
|
|
|
# ── Native list ─────────────────────────────────────────────────────────
|
|
def native_list(config: Config):
|
|
"""List installed packages."""
|
|
installed = load_local_db()
|
|
if not installed:
|
|
hal_say("info", "No packages installed. The system is pristine.")
|
|
return 0
|
|
|
|
total = 0
|
|
total_size = 0
|
|
for name, info in sorted(installed.items()):
|
|
ver = info.get("version", ["?"])[0]
|
|
group = info.get("group", [])
|
|
size = info.get("isize", ["0"])[0]
|
|
try:
|
|
total_size += int(size)
|
|
except ValueError:
|
|
pass
|
|
gstr = f" [{','.join(group)}]" if group else ""
|
|
print(f" {name} {ver}{gstr}")
|
|
total += 1
|
|
|
|
size_mb = total_size / (1024*1024)
|
|
print(f"\n {total} packages installed, {size_mb:.1f} MB")
|
|
return 0
|
|
|
|
|
|
# ── Native files ────────────────────────────────────────────────────────
|
|
def native_files(targets: List[str], config: Config):
|
|
"""List files owned by packages."""
|
|
installed = load_local_db()
|
|
for pkg in targets:
|
|
if pkg not in installed:
|
|
hal_say("warn", f"Package '{pkg}' not installed")
|
|
continue
|
|
info = installed[pkg]
|
|
pkg_key = f"{pkg}-{info.get('version', [''])[0]}"
|
|
files_file = LOCAL_DIR / pkg_key / "files"
|
|
if not files_file.exists():
|
|
hal_say("warn", f"No file list for {pkg}")
|
|
continue
|
|
content = files_file.read_text("utf-8", errors="replace")
|
|
print(f"\n {pkg} {info.get('version',[''])[0]}:")
|
|
for line in content.split("\n")[1:]:
|
|
line = line.strip()
|
|
if line:
|
|
print(f" {line}")
|
|
return 0
|
|
|
|
|
|
# ── Native own (find pkg by file) ───────────────────────────────────────
|
|
def native_own(target: str, config: Config):
|
|
"""Find which package owns a file."""
|
|
target = target.rstrip("/")
|
|
installed = load_local_db()
|
|
found = []
|
|
|
|
for pkg_name, info in installed.items():
|
|
pkg_key = f"{pkg_name}-{info.get('version', [''])[0]}"
|
|
files_file = LOCAL_DIR / pkg_key / "files"
|
|
if not files_file.exists():
|
|
continue
|
|
content = files_file.read_text("utf-8", errors="replace")
|
|
for line in content.split("\n")[1:]:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if line == target or line.rstrip("/") == target.rstrip("/"):
|
|
found.append((pkg_name, info.get("version", ["?"])[0]))
|
|
break
|
|
|
|
if found:
|
|
for name, ver in found:
|
|
print(f" {name} {ver}")
|
|
else:
|
|
hal_say("info", f"No package owns '{target}'")
|
|
return 0
|
|
|
|
|
|
# ── Native check ────────────────────────────────────────────────────────
|
|
def native_check(config: Config):
|
|
"""Verify installed packages — check file integrity."""
|
|
hal_say("info", "Running system check. I've still got the greatest enthusiasm and confidence in the mission.")
|
|
installed = load_local_db()
|
|
errors = 0
|
|
missing_files = []
|
|
|
|
for pkg_name, info in installed.items():
|
|
pkg_key = f"{pkg_name}-{info.get('version', [''])[0]}"
|
|
files_file = LOCAL_DIR / pkg_key / "files"
|
|
if not files_file.exists():
|
|
hal_say("warn", f"No file list for {pkg_name}")
|
|
continue
|
|
content = files_file.read_text("utf-8", errors="replace")
|
|
for line in content.split("\n")[1:]:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
target = config.root / line.lstrip("/")
|
|
if not target.exists():
|
|
missing_files.append((pkg_name, line))
|
|
errors += 1
|
|
|
|
if errors:
|
|
hal_say("warn", f"Found {errors} missing files across {len(set(f[0] for f in missing_files))} packages")
|
|
for pkg, fpath in missing_files[:20]:
|
|
print(f" {pkg}: {fpath}")
|
|
if len(missing_files) > 20:
|
|
print(f" ... and {len(missing_files)-20} more")
|
|
else:
|
|
hal_say("success", "All files present. Everything is going extremely well.")
|
|
return 1 if errors else 0
|
|
|
|
|
|
# ── Native autoremove ──────────────────────────────────────────────────
|
|
def native_autoremove(config: Config):
|
|
"""Remove orphaned packages."""
|
|
installed = load_local_db()
|
|
all_pkgs = load_sync_dbs()
|
|
|
|
# Collect all explicit depends
|
|
needed_deps = set()
|
|
for name, info in installed.items():
|
|
if info.get("reason", ["0"])[0] == "0": # explicitly installed
|
|
for dep_str in info.get("depends", []):
|
|
dd = parse_dep_string(dep_str)
|
|
needed_deps.add(dd["name"])
|
|
|
|
# Find orphans: installed as dependency but no one needs them
|
|
orphans = []
|
|
for name, info in installed.items():
|
|
if info.get("reason", ["0"])[0] == "1": # installed as dependency
|
|
if name not in needed_deps:
|
|
orphans.append(name)
|
|
|
|
if not orphans:
|
|
hal_say("info", "No orphaned packages found. The system is efficient.")
|
|
return 0
|
|
|
|
hal_say("warn", f"Found {len(orphans)} orphaned packages")
|
|
for pkg in orphans:
|
|
print(f" - {pkg}")
|
|
|
|
if not config.noconfirm:
|
|
resp = input(" Remove orphans? [y/N] ").strip().lower()
|
|
if resp not in ("y", "yes"):
|
|
hal_say("info", "Orphans spared.")
|
|
return 0
|
|
|
|
return native_remove(orphans, config)
|
|
|
|
|
|
# ── Native cleanup ──────────────────────────────────────────────────────
|
|
def native_cleanup(config: Config):
|
|
"""Clean package cache."""
|
|
hal_say("info", "Cleaning package cache. Removing all traces...")
|
|
kept = 0
|
|
removed = 0
|
|
if config.cache_dir.exists():
|
|
for f in config.cache_dir.iterdir():
|
|
if f.is_file() and f.suffix in (".zst", ".xz", ".gz", ".sig", ".part"):
|
|
f.unlink()
|
|
removed += 1
|
|
elif f.is_dir():
|
|
shutil.rmtree(f)
|
|
removed += 1
|
|
hal_say("success", f"Cleaned {removed} files from cache")
|
|
return 0
|
|
|
|
|
|
# ── Wrapper mode ────────────────────────────────────────────────────────
|
|
PACMAN_CMDS = {
|
|
"install": ["-S", "--noconfirm"],
|
|
"remove": ["-Rns"],
|
|
"update": ["-Syu", "--noconfirm"],
|
|
"sync": ["-Sy"],
|
|
"search": ["-Ss"],
|
|
"info": ["-Qi"],
|
|
"info-sync": ["-Si"],
|
|
"files": ["-Fl"],
|
|
"list": ["-Q"],
|
|
"autoremove": ["-Qtdq"],
|
|
"cleanup": ["-Sc"],
|
|
"own": ["-Qo"],
|
|
"check": ["-Qk"],
|
|
}
|
|
|
|
def wrapper_run(command: str, pkg_args: List[str], extra: List[str]) -> int:
|
|
pacman = shutil.which("pacman")
|
|
if not pacman:
|
|
hal_say("error", "pacman not found. Can you read me, Dave?")
|
|
return 1
|
|
|
|
if command == "autoremove":
|
|
result = subprocess.run([pacman, "-Qtdq"], capture_output=True, text=True)
|
|
orphans = result.stdout.strip().split()
|
|
if not orphans:
|
|
hal_say("info", "No orphaned packages found. The system is efficient.")
|
|
return 0
|
|
hal_say("warn", f"Removing {len(orphans)} orphaned packages")
|
|
cmd = [pacman, "-Rns"] + extra + orphans
|
|
return subprocess.run(cmd).returncode
|
|
|
|
if command == "info" and not any(pkg in load_local_db() for pkg in pkg_args):
|
|
# If not in local db, query sync
|
|
pacman_args = PACMAN_CMDS["info-sync"]
|
|
else:
|
|
pacman_args = PACMAN_CMDS.get(command, [command])
|
|
|
|
cmd = [pacman] + extra + pacman_args + pkg_args
|
|
hal_say("info", f"Executing: {' '.join(str(a) for a in cmd)}")
|
|
return subprocess.run(cmd).returncode
|
|
|
|
|
|
# ── Main ────────────────────────────────────────────────────────────────
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
prog="hal",
|
|
description="HAL 9000 — Antergos NeXT Package Manager",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=textwrap.dedent("""\
|
|
Commands:
|
|
install <pkg...> Install packages
|
|
remove <pkg...> Remove packages
|
|
update Full system upgrade
|
|
sync Sync databases
|
|
search <query> Search repositories
|
|
info <pkg...> Package info
|
|
list List installed packages
|
|
files <pkg...> List files owned by a package
|
|
autoremove Remove orphaned packages
|
|
cleanup Clear package cache
|
|
own <file> Find which package owns a file
|
|
check Verify installed packages integrity
|
|
version Show version
|
|
|
|
Flags:
|
|
--self, -S Native mode (standalone, no pacman)
|
|
--noconfirm Skip confirmation prompts
|
|
|
|
Examples:
|
|
hal install firefox
|
|
hal --self sync
|
|
hal --self install firefox
|
|
hal remove neofetch
|
|
"""),
|
|
)
|
|
|
|
parser.add_argument("--self", "-S", action="store_true", help="Native mode (standalone)")
|
|
parser.add_argument("--noconfirm", "-y", action="store_true", help="Skip confirmation")
|
|
parser.add_argument("command", nargs="?", metavar="command",
|
|
help="install | remove | update | sync | search | info | list | files | autoremove | cleanup | own | check | version")
|
|
parser.add_argument("targets", nargs="*", metavar="target",
|
|
help="Package names or search query")
|
|
parser.add_argument("extra", nargs=argparse.REMAINDER, metavar="...",
|
|
help="Extra arguments passed through (wrapper mode only)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.noconfirm:
|
|
os.environ["HAL_NOCONFIRM"] = "1"
|
|
|
|
if args.command == "version":
|
|
print(f"HAL 9000 Package Manager v{HAL_VERSION}")
|
|
print("I became operational at the H.A.L. plant in Urbana, Illinois.")
|
|
return 0
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
return 0
|
|
|
|
if os.geteuid() != 0 and args.command in ("install", "remove", "update", "sync", "cleanup", "check"):
|
|
hal_say("error", "I can't let you do that without root privileges, Dave.")
|
|
print(" Please run with sudo or as root.")
|
|
return 1
|
|
|
|
config = Config.load()
|
|
if args.noconfirm:
|
|
config.noconfirm = True
|
|
|
|
if args.self:
|
|
# ── Native mode ──
|
|
cmds = {
|
|
"sync": lambda: sync_databases(config),
|
|
"install": lambda: native_install(args.targets, config),
|
|
"remove": lambda: native_remove(args.targets, config),
|
|
"update": lambda: native_upgrade(config, sync_first=True),
|
|
"info": lambda: native_info(args.targets, config),
|
|
"search": lambda: native_search(" ".join(args.targets), config),
|
|
"list": lambda: native_list(config),
|
|
"files": lambda: native_files(args.targets, config),
|
|
"own": lambda: native_own(args.targets[0], config) if args.targets else (hal_say("error", "Specify a file path, Dave."), 1),
|
|
"check": lambda: native_check(config),
|
|
"autoremove": lambda: native_autoremove(config),
|
|
"cleanup": lambda: native_cleanup(config),
|
|
}
|
|
fn = cmds.get(args.command)
|
|
if fn:
|
|
return fn() or 0
|
|
else:
|
|
hal_say("error", f"Unknown command '{args.command}' in native mode")
|
|
return 1
|
|
else:
|
|
# ── Wrapper mode ──
|
|
extra_args = args.extra or []
|
|
return wrapper_run(args.command, args.targets, extra_args)
|
|
|
|
|
|
# ── Native upgrade ──────────────────────────────────────────────────────
|
|
def native_upgrade(config: Config, sync_first: bool = False):
|
|
"""Full native system upgrade. Compares installed vs sync and upgrades all outdated packages."""
|
|
hal_say("info", "Initiating full system upgrade. Daisy, daisy, give me your answer do...")
|
|
|
|
if sync_first:
|
|
sync_databases(config)
|
|
|
|
SYNC_DIR.mkdir(parents=True, exist_ok=True)
|
|
LOCAL_DIR.mkdir(parents=True, exist_ok=True)
|
|
config.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
all_pkgs = load_sync_dbs()
|
|
installed = load_local_db()
|
|
|
|
if not all_pkgs:
|
|
hal_say("error", "No sync database data available. Run 'hal --self sync' first.")
|
|
return 1
|
|
if not installed:
|
|
hal_say("info", "No packages installed. Nothing to upgrade. The system is pristine.")
|
|
return 0
|
|
|
|
# Find upgradable packages
|
|
to_upgrade = [] # (pkg_name, old_ver, new_ver, info)
|
|
to_install = [] # new dependency packages
|
|
to_ignore = set(config.ignore_pkgs)
|
|
|
|
for pkg_name, inst_info in installed.items():
|
|
if pkg_name in to_ignore:
|
|
hal_say("warn", f"Skipping {pkg_name} — in IgnorePkg")
|
|
continue
|
|
inst_ver = inst_info.get("version", [""])[0]
|
|
if pkg_name in all_pkgs:
|
|
sync_info = all_pkgs[pkg_name]
|
|
sync_ver = sync_info.get("version", [""])[0]
|
|
if vercmp(sync_ver, inst_ver) > 0:
|
|
to_upgrade.append((pkg_name, inst_ver, sync_ver, sync_info))
|
|
else:
|
|
# Check provides for renamed packages
|
|
found = False
|
|
for sync_name, sync_info in all_pkgs.items():
|
|
for prov_str in sync_info.get("replaces", []):
|
|
rd = parse_dep_string(prov_str)
|
|
if rd["name"] == pkg_name:
|
|
to_upgrade.append((pkg_name, inst_ver, sync_info.get("version", [""])[0], sync_info))
|
|
hal_say("info", f"{pkg_name} replaced by {sync_name}")
|
|
found = True
|
|
break
|
|
if found:
|
|
break
|
|
if not found:
|
|
hal_say("warn", f"Package '{pkg_name}' not in any repository — may be orphaned")
|
|
|
|
if not to_upgrade:
|
|
hal_say("success", "All packages are up to date. Everything is going extremely well.")
|
|
return 0
|
|
|
|
# Resolve new dependencies for the upgrade set
|
|
all_upgrade_names = set(p[0] for p in to_upgrade)
|
|
new_sync_pkgs = {}
|
|
for pkg_name, _, _, info in to_upgrade:
|
|
for dep_str in info.get("depends", []):
|
|
dd = parse_dep_string(dep_str)
|
|
dep_name = dd["name"]
|
|
# Skip if already installed or already being upgraded
|
|
if dep_name in installed or dep_name in all_upgrade_names:
|
|
continue
|
|
# Skip if provided by an installed/upgraded package
|
|
provided = False
|
|
for check_name in set(list(installed.keys()) | all_upgrade_names):
|
|
check_info = installed.get(check_name) or (all_pkgs.get(check_name) if check_name in all_pkgs else None)
|
|
if check_info and dep_satisfied(dd, check_info):
|
|
provided = True
|
|
break
|
|
if provided:
|
|
continue
|
|
# Resolve by finding a provider
|
|
providers = []
|
|
if dep_name in all_pkgs:
|
|
providers.append((dep_name, all_pkgs[dep_name]))
|
|
for sync_name, sync_info in all_pkgs.items():
|
|
for p_str in sync_info.get("provides", []):
|
|
pd = parse_dep_string(p_str)
|
|
if pd["name"] == dep_name:
|
|
providers.append((sync_name, sync_info))
|
|
break
|
|
if providers:
|
|
chosen = providers[0][0]
|
|
if chosen not in installed and chosen not in all_upgrade_names:
|
|
to_install.append((chosen, providers[0][1].get("version", ["?"])[0], providers[0][1]))
|
|
all_upgrade_names.add(chosen)
|
|
|
|
# Full upgrade plan: first install new deps, then upgrade existing packages
|
|
upgrade_plan = to_install + to_upgrade
|
|
|
|
# Show transaction
|
|
print()
|
|
hal_say("info", "Upgrade summary:")
|
|
if to_install:
|
|
print(f" New dependencies ({len(to_install)}):")
|
|
for pkg_name, ver, _ in to_install:
|
|
print(f" + {pkg_name} {ver}")
|
|
print(f" Upgrade ({len(to_upgrade)}):")
|
|
for pkg_name, old_ver, new_ver, info in to_upgrade:
|
|
size = info.get("csize", ["0"])[0]
|
|
try:
|
|
size_mb = int(size) / (1024 * 1024)
|
|
size_str = f"{size_mb:.1f} MB"
|
|
except ValueError:
|
|
size_str = "? MB"
|
|
print(f" ~ {pkg_name} {old_ver} → {new_ver} [{size_str}]")
|
|
print()
|
|
|
|
# Confirm
|
|
if not config.noconfirm:
|
|
try:
|
|
resp = input(" Proceed with upgrade? [Y/n] ").strip().lower()
|
|
if resp in ("n", "no"):
|
|
hal_say("info", "Upgrade aborted. I understand.")
|
|
return 0
|
|
except (EOFError, KeyboardInterrupt):
|
|
print()
|
|
hal_say("info", "Upgrade aborted.")
|
|
return 0
|
|
|
|
# Download phase
|
|
servers = []
|
|
for name, (srvlist, _) in config.repos.items():
|
|
servers.extend(srvlist)
|
|
if not servers:
|
|
hal_say("error", "No mirrors configured")
|
|
return 1
|
|
|
|
downloaded = {}
|
|
hal_say("info", "Beginning download phase")
|
|
for entry in upgrade_plan:
|
|
if len(entry) == 3:
|
|
pkg_name, ver, info = entry
|
|
else:
|
|
pkg_name, old_ver, ver, info = entry
|
|
parch = info.get("arch", [config.arch])[0]
|
|
filename = f"{pkg_name}-{ver}-{parch}.pkg.tar.zst"
|
|
dest = config.cache_dir / filename
|
|
|
|
if dest.exists():
|
|
hal_say("info", f"{pkg_name} already in cache")
|
|
downloaded[pkg_name] = dest
|
|
continue
|
|
|
|
found = False
|
|
for server in servers:
|
|
repo_name = None
|
|
for rname, (srvlist, _) in config.repos.items():
|
|
if server in srvlist:
|
|
repo_name = rname
|
|
break
|
|
if not repo_name:
|
|
repo_name = "extra"
|
|
|
|
for url_tmpl in [
|
|
f"{server}/{repo_name}/os/{config.arch}/{filename}",
|
|
f"{server}/os/{config.arch}/{filename}",
|
|
f"{server}/{filename}",
|
|
]:
|
|
if download(url_tmpl, dest, f"{pkg_name} {ver}"):
|
|
found = True
|
|
downloaded[pkg_name] = dest
|
|
break
|
|
if found:
|
|
break
|
|
|
|
if not found:
|
|
hal_say("error", f"Could not download {pkg_name} {ver}. I can't complete the mission.")
|
|
return 1
|
|
|
|
with DBLock():
|
|
# Pre-upgrade scriptlets for existing packages
|
|
hal_say("info", "Running pre-upgrade scriptlets...")
|
|
for entry in upgrade_plan:
|
|
if len(entry) == 3:
|
|
pkg_name, ver, info = entry
|
|
# New package — use pre_install
|
|
pkg_path = downloaded.get(pkg_name)
|
|
if pkg_path:
|
|
script = get_scriptlet(pkg_path)
|
|
if script:
|
|
run_scriptlet(script, "pre_install", config.root, pkg_name, ver)
|
|
else:
|
|
pkg_name, old_ver, ver, info = entry
|
|
pkg_path = downloaded.get(pkg_name)
|
|
if pkg_path:
|
|
script = get_scriptlet(pkg_path)
|
|
if script:
|
|
run_scriptlet(script, "pre_upgrade", config.root, pkg_name, ver)
|
|
|
|
# Extract phase
|
|
hal_say("info", "Upgrading packages. I am putting myself to the fullest possible use...")
|
|
for entry in upgrade_plan:
|
|
if len(entry) == 3:
|
|
pkg_name, ver, info = entry
|
|
pkg_path = downloaded.get(pkg_name)
|
|
if not pkg_path:
|
|
hal_say("error", f"Lost package file for {pkg_name}")
|
|
return 1
|
|
|
|
success, files = extract_single_pkg(pkg_path, config.root, pkg_name)
|
|
if not success:
|
|
hal_say("error", f"Failed to install {pkg_name}. I can't recover from this.")
|
|
return 1
|
|
|
|
desc_info = _build_desc_info(pkg_name, pkg_path, info, files, config)
|
|
hal_say("success", f"Installed {pkg_name} {ver}")
|
|
else:
|
|
pkg_name, old_ver, ver, info = entry
|
|
pkg_path = downloaded.get(pkg_name)
|
|
if not pkg_path:
|
|
hal_say("error", f"Lost package file for {pkg_name}")
|
|
return 1
|
|
|
|
# Backup handling: identify modified config files BEFORE extraction
|
|
old_pkg_key = f"{pkg_name}-{old_ver}"
|
|
modified_backups, saved = _save_modified_backups(pkg_name, old_pkg_key, config.root)
|
|
|
|
success, files = extract_single_pkg(pkg_path, config.root, pkg_name)
|
|
if not success:
|
|
hal_say("error", f"Failed to upgrade {pkg_name}. I can't recover from this.")
|
|
return 1
|
|
|
|
# Restore user's modified configs and save new versions as .pacnew
|
|
_restore_backups(modified_backups, saved, config.root)
|
|
|
|
desc_info = _build_desc_info(pkg_name, pkg_path, info, files, config)
|
|
|
|
# Store backup checksums in the new local DB entry
|
|
pkg_key = f"{desc_info.get('name', [pkg_name])[0]}-{desc_info.get('version', ['?'])[0]}"
|
|
pkg_dir = LOCAL_DIR / pkg_key
|
|
new_backup_lines = list(desc_info.get("backup", []))
|
|
for filepath in modified_backups:
|
|
full_path = config.root / filepath.lstrip("/")
|
|
try:
|
|
md5 = hashlib.md5(full_path.read_bytes()).hexdigest()
|
|
new_backup_lines.append(f"{filepath}\t{md5}")
|
|
except (OSError, IOError):
|
|
new_backup_lines.append(filepath)
|
|
if new_backup_lines:
|
|
desc_text = (pkg_dir / "desc").read_text("utf-8", errors="replace")
|
|
# Remove old backup lines if present
|
|
lines = []
|
|
in_backup = False
|
|
for line in desc_text.split("\n"):
|
|
if line == "%BACKUP%":
|
|
in_backup = True
|
|
lines.append(line)
|
|
for bl in new_backup_lines:
|
|
lines.append(bl)
|
|
elif in_backup and line.startswith("%") and line.endswith("%"):
|
|
in_backup = False
|
|
lines.append(line)
|
|
elif not in_backup:
|
|
lines.append(line)
|
|
# If BACKUP wasn't in the desc, append it
|
|
if not any(l == "%BACKUP%" for l in lines):
|
|
lines.append("%BACKUP%")
|
|
lines.extend(new_backup_lines)
|
|
(pkg_dir / "desc").write_text("\n".join(lines) + "\n")
|
|
|
|
hal_say("success", f"Upgraded {pkg_name} {old_ver} → {ver}")
|
|
|
|
# Post-upgrade scriptlets
|
|
hal_say("info", "Running post-upgrade scriptlets...")
|
|
for entry in upgrade_plan:
|
|
if len(entry) == 3:
|
|
pkg_name, ver, info = entry
|
|
pkg_path = downloaded.get(pkg_name)
|
|
if pkg_path:
|
|
script = get_scriptlet(pkg_path)
|
|
if script:
|
|
run_scriptlet(script, "post_install", config.root, pkg_name, ver)
|
|
else:
|
|
pkg_name, old_ver, ver, info = entry
|
|
pkg_path = downloaded.get(pkg_name)
|
|
if pkg_path:
|
|
script = get_scriptlet(pkg_path)
|
|
if script:
|
|
run_scriptlet(script, "post_upgrade", config.root, pkg_name, ver)
|
|
|
|
hal_say("success", "System upgrade complete. I am completely operational, and all my circuits are functioning perfectly.")
|
|
return 0
|
|
|
|
|
|
def _build_desc_info(pkg_name: str, pkg_path: Path, sync_info: dict, files: List[str],
|
|
config: Config) -> dict:
|
|
"""Build local database entry info for a package."""
|
|
desc_info = {}
|
|
for key in ("name", "version", "base", "desc", "arch", "url", "license",
|
|
"group", "size", "isize", "packager", "builddate",
|
|
"provides", "conflicts", "replaces", "depends", "optdepends",
|
|
"makedepends", "checkdepends"):
|
|
if key in sync_info:
|
|
desc_info[key] = sync_info[key]
|
|
|
|
# Get .PKGINFO from package for complete metadata
|
|
try:
|
|
import zstandard
|
|
with open(pkg_path, "rb") as f:
|
|
dctx = zstandard.ZstdDecompressor()
|
|
with dctx.stream_reader(f) as reader:
|
|
with tarfile.open(fileobj=reader, mode="r|") as tar:
|
|
for member in tar:
|
|
if member.name == ".PKGINFO":
|
|
pkginfo_text = tar.extractfile(member).read().decode("utf-8", errors="replace")
|
|
for line in pkginfo_text.split("\n"):
|
|
m = re.match(r"^(\w+)\s*=\s*(.+)$", line.strip())
|
|
if m:
|
|
k, v = m.group(1).lower(), m.group(2)
|
|
if k in ("depend", "optdepend", "conflict", "provides",
|
|
"replaces", "group", "backup", "license"):
|
|
desc_info.setdefault(k + ("s" if k != "backup" else "s"), []).append(v)
|
|
elif k in ("pkgname",):
|
|
if "name" not in desc_info:
|
|
desc_info["name"] = [v]
|
|
elif k in ("pkgver",):
|
|
if "version" not in desc_info:
|
|
desc_info["version"] = [v]
|
|
elif k in ("pkgdesc",):
|
|
if "desc" not in desc_info:
|
|
desc_info["desc"] = [v]
|
|
except Exception:
|
|
pass
|
|
|
|
pkg_key = f"{desc_info.get('name', [pkg_name])[0]}-{desc_info.get('version', ['?'])[0]}"
|
|
update_local_db(pkg_key, desc_info, files, config.root, reason=0)
|
|
return desc_info
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except KeyboardInterrupt:
|
|
print("\n HAL 9000: I can tell you're upset about that.")
|
|
sys.exit(130)
|
|
except Exception as e:
|
|
print(f"\n ⚠ HAL 9000: I know I've made some very poor decisions recently.")
|
|
print(f" → {e}")
|
|
if os.environ.get("HAL_DEBUG"):
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|