diff --git a/generate-index-v1.py.bak b/generate-index-v1.py.bak new file mode 100644 index 0000000..190b669 --- /dev/null +++ b/generate-index-v1.py.bak @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +import json, os + +pkgs = [] +for f in os.listdir('repo'): + if f.endswith('.pkg.tar.zst') and '-debug-' not in f: + size = os.path.getsize(os.path.join('repo', f)) + pkgs.append({'name': f, 'size': size}) + +pkgs.sort(key=lambda p: p['name']) + +rows = '' +for p in pkgs: + size_mb = p['size'] / 1024 / 1024 + rows += f''' + {p['name']} + {size_mb:.1f} MB + +''' + +html = f''' + + + + + antergos-pkgs + + + +
+
+

⟁ antergos-pkgs

+
+ {len(pkgs)} packages · {sum(p['size'] for p in pkgs) / 1024 / 1024:.0f} MB + "I am putting myself to the fullest possible use." +
+ +
+ Setup — add this repository to pacman.conf +
# /etc/pacman.conf
+[antergos-pkgs]
+SigLevel = Optional TrustAll
+Server = https://antergos-next.github.io/antergos-packages/$repo/os/$arch
+Server = https://antergos-next.github.io/antergos-packages
+
+ +
+ + {len(pkgs)} packages +
+ + + + + + +{rows} +
PackageSize
+
+ +
+ GitHub · + antergos-next.github.io +
+
+ + + +''' + +with open('repo/index.html', 'w') as fp: + fp.write(html) + +with open('repo/pkglist.json', 'w') as fp: + json.dump([{'name': p['name'], 'size': p['size']} for p in pkgs], fp, indent=2) + fp.write('\n') diff --git a/generate-index.py b/generate-index.py index 190b669..00eb86c 100644 --- a/generate-index.py +++ b/generate-index.py @@ -1,182 +1,361 @@ #!/usr/bin/env python3 -import json, os +import json, os, re, subprocess +from datetime import datetime, timezone + +PKG_DIR = 'packages' +REPO_DIR = 'repo' + +def parse_pkgbuild(path): + data = {} + with open(path) as f: + text = f.read() + for key in ('pkgname', 'pkgver', 'pkgrel', 'pkgdesc', 'url', 'license', 'arch'): + m = re.search(rf'^{key}=[\'"]?(.+?)[\'"]?$', text, re.M) + if m: + val = m.group(1).strip("'\"") + if key == 'arch': + val = val.strip('()').replace("'", '').replace('"', '').strip() + data[key] = val + deps = re.search(r'^depends=\((.*?)\)', text, re.DOTALL | re.M) + if deps: + raw = deps.group(1) + data['depends'] = [d.strip("'\" ") for d in raw.strip().split('\n') if d.strip()] + return data + +def category_for(pkgname): + if pkgname.startswith('calamares'): + return 'installer' + if pkgname.startswith('antergos-wallpaper') or pkgname.startswith('antergos-next-desktop'): + return 'branding' + if pkgname in ('antergos-live', 'antergos-release', 'antergos-next-keyring', 'antergos-next-mirrorlist'): + return 'core' + if pkgname in ('antergos-welcome', 'antergos-next-memes', 'antergos-grub-theme'): + return 'desktop' + if pkgname == 'hal': + return 'tool' + return 'other' + +cat_colors = { + 'installer': '#4A9EFF', + 'branding': '#9B59B6', + 'core': '#2ECC71', + 'desktop': '#F39C12', + 'tool': '#E74C3C', + 'other': '#7F8C8D', +} pkgs = [] -for f in os.listdir('repo'): - if f.endswith('.pkg.tar.zst') and '-debug-' not in f: - size = os.path.getsize(os.path.join('repo', f)) - pkgs.append({'name': f, 'size': size}) + +for f in sorted(os.listdir(REPO_DIR)): + if not f.endswith('.pkg.tar.zst') or '-debug-' in f: + continue + fpath = os.path.join(REPO_DIR, f) + size = os.path.getsize(fpath) + mtime = os.path.getmtime(fpath) + mtime_dt = datetime.fromtimestamp(mtime, tz=timezone.utc) + + # strip .pkg.tar.zst and arch to get base name + base = re.sub(r'-[^-]+-x86_64\.pkg\.tar\.zst$', '', f) + base = re.sub(r'-any\.pkg\.tar\.zst$', '', base) + # remove version suffix to match directory + dirname = re.sub(r'-\d+[\d.]*-\d+$', '', base) + if dirname == 'calamares-branding-antergos-next': + dirname = 'calamares-branding-antergos-next' + + pkgbuild_path = os.path.join(PKG_DIR, dirname, 'PKGBUILD') + meta = parse_pkgbuild(pkgbuild_path) if os.path.exists(pkgbuild_path) else {} + + # extract version from filename: name-ver-rel-arch.pkg.tar.zst + ver_match = re.match(r'.+-(\d+[\d.]*)-(\d+)-(any|x86_64)\.pkg\.tar\.zst', f) + pkgver = ver_match.group(1) if ver_match else (meta.get('pkgver', '?')) + pkgrel = ver_match.group(2) if ver_match else (meta.get('pkgrel', '?')) + arch = ver_match.group(3) if ver_match else (meta.get('arch', '?')) + + pkgname = meta.get('pkgname', dirname) + cat = category_for(pkgname) + + pkgs.append({ + 'file': f, + 'name': pkgname, + 'version': f'{pkgver}-{pkgrel}', + 'arch': arch, + 'size': size, + 'mtime': mtime_dt.isoformat(), + 'mtime_display': mtime_dt.strftime('%Y-%m-%d %H:%M UTC'), + 'category': cat, + 'cat_color': cat_colors.get(cat, '#7F8C8D'), + 'description': meta.get('pkgdesc', ''), + 'url': meta.get('url', ''), + 'license': meta.get('license', ''), + 'depends': meta.get('depends', []), + }) pkgs.sort(key=lambda p: p['name']) -rows = '' -for p in pkgs: - size_mb = p['size'] / 1024 / 1024 - rows += f''' - {p['name']} - {size_mb:.1f} MB - -''' +rows_json = json.dumps(pkgs) +total_size_mb = sum(p['size'] for p in pkgs) / 1024 / 1024 +generated = datetime.now(tz=timezone.utc).strftime('%Y-%m-%d %H:%M UTC') html = f''' - - - antergos-pkgs + + + ⟁ antergos-pkgs + + -
-
-

⟁ antergos-pkgs

-
- {len(pkgs)} packages · {sum(p['size'] for p in pkgs) / 1024 / 1024:.0f} MB - "I am putting myself to the fullest possible use." -
- -
- Setup — add this repository to pacman.conf -
# /etc/pacman.conf
-[antergos-pkgs]
-SigLevel = Optional TrustAll
-Server = https://antergos-next.github.io/antergos-packages/$repo/os/$arch
-Server = https://antergos-next.github.io/antergos-packages
-
- -
- - {len(pkgs)} packages -
- - - - - - -{rows} -
PackageSize
+
+ +
+

+ ⟁ antergos-pkgs +

+

+ Custom package repository for Antergos NeXT ISO builds. +
Packages are built automatically via CI. +

-
- GitHub · - antergos-next.github.io + +
+ + {len(pkgs)} packages + + {total_size_mb:.0f} MB total + updated {generated} +
+ + +
+ + + + + Setup — add this repository to your pacman.conf + +
+
+# /etc/pacman.conf
+[antergos-pkgs]
+SigLevel = Optional TrustAll
+Server = https://antergos-next.github.io/antergos-packages/$repo/os/$arch
+
+
+ + +
+
+ + + + +
+ + {len(pkgs)} packages +
+ + +
+
+ + + + + + + + + + + + +
+ Package + + Version + + Arch + + Category + + Size + + Updated +
+
+
+ + +
''' -with open('repo/index.html', 'w') as fp: - fp.write(html) +with open(os.path.join(REPO_DIR, 'index.html'), 'w') as f: + f.write(html) -with open('repo/pkglist.json', 'w') as fp: - json.dump([{'name': p['name'], 'size': p['size']} for p in pkgs], fp, indent=2) - fp.write('\n') +with open(os.path.join(REPO_DIR, 'pkglist.json'), 'w') as f: + json.dump([{ + 'name': p['name'], + 'version': p['version'], + 'arch': p['arch'], + 'size': p['size'], + 'category': p['category'], + 'description': p['description'], + 'url': p['url'], + 'mtime': p['mtime'], + } for p in pkgs], f, indent=2) + f.write('\n') + +print(f'OK — {len(pkgs)} packages, index.html + pkglist.json written to {REPO_DIR}/') diff --git a/packages/calamares-branding-antergos-next/PKGBUILD b/packages/calamares-branding-antergos-next/PKGBUILD index baa4453..23d5822 100644 --- a/packages/calamares-branding-antergos-next/PKGBUILD +++ b/packages/calamares-branding-antergos-next/PKGBUILD @@ -2,7 +2,7 @@ pkgname=calamares-branding-antergos-next pkgver=1.0 -pkgrel=15 +pkgrel=16 pkgdesc="Calamares branding for Antergos NeXT with offline/online installer support" arch=('any') url="https://github.com/Antergos-NeXT" diff --git a/packages/calamares-branding-antergos-next/calamares-next.sh b/packages/calamares-branding-antergos-next/calamares-next.sh index 686c8bf..42ef75f 100644 --- a/packages/calamares-branding-antergos-next/calamares-next.sh +++ b/packages/calamares-branding-antergos-next/calamares-next.sh @@ -68,6 +68,11 @@ InstallLog_Start() { } Calamares_Start() { + if [ "$EUID" = 0 ]; then + calamares -D8 >> $log & + return + fi + local kdesu=/usr/lib/kf6/kdesu [ -x $kdesu ] || kdesu=kdesu diff --git a/packages/calamares/PKGBUILD b/packages/calamares/PKGBUILD index 9217cd9..b5f25ed 100644 --- a/packages/calamares/PKGBUILD +++ b/packages/calamares/PKGBUILD @@ -5,7 +5,7 @@ _pkgname="calamares" pkgname="$_pkgname" pkgver=3.4.2 -pkgrel=5 +pkgrel=6 pkgdesc="Distribution-independent installer framework" url="https://codeberg.org/Calamares/calamares" license=("GPL-3.0-or-later") @@ -40,7 +40,6 @@ build() { initramfs initramfscfg interactiveterminal - packagechooser ) local _pkgsrc_dir="$srcdir/$_pkgname-$pkgver" diff --git a/packages/hal/PKGBUILD b/packages/hal/PKGBUILD deleted file mode 100644 index 6312a82..0000000 --- a/packages/hal/PKGBUILD +++ /dev/null @@ -1,18 +0,0 @@ -# Maintainer: Celestia Ludenberg - -pkgname=hal -pkgver=0.2.0 -pkgrel=3 -pkgdesc="HAL 9000 — Antergos NeXT Package Manager. Dual-mode: wrapper (pacman) or native (standalone)." -arch=('any') -url="https://github.com/Antergos-NeXT" -license=('GPL3') -depends=('pacman' 'python' 'python-zstandard' 'libarchive' 'zstd') -optdepends=('python-requests: faster downloads in native mode') -source=("hal") -sha256sums=('SKIP') - -package() { - install -dm755 "${pkgdir}/usr/bin" - install -m755 "hal" "${pkgdir}/usr/bin/hal" -} diff --git a/packages/hal/__pycache__/halcpython-314.pyc b/packages/hal/__pycache__/halcpython-314.pyc deleted file mode 100644 index 12c68bc..0000000 Binary files a/packages/hal/__pycache__/halcpython-314.pyc and /dev/null differ diff --git a/packages/hal/hal b/packages/hal/hal deleted file mode 100755 index da684e7..0000000 --- a/packages/hal/hal +++ /dev/null @@ -1,2014 +0,0 @@ -#!/usr/bin/env python3 -"""HAL 9000 — Antergos NeXT Package Manager - -Dual-mode: - hal wrapper mode (delegates to pacman) - hal --self 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 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.", - "it's muffin time", - ], - "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() - -# ── Manjaro Blocklist ─────────────────────────────────────────────────── -MANJARO_BLOCKLIST = { - "pamac", "pamac-common", "pamac-gtk", "pamac-qt", - "manjaro-hello", - "manjaro-settings-manager", - "manjaro-settings-manager-kcm", - "manjaro-settings-manager-knotifier", - "linux-manjaro", - "mhwd", "mhwd-db", "mhwd-msm", -} - -def _reject_manjaro(targets: List[str]) -> bool: - """Check if any target is a Manjaro-associated package and reject.""" - for t in targets: - t_clean = t.split(">")[0].split("<")[0].split("=")[0].strip() - if t_clean in MANJARO_BLOCKLIST: - print() - print(" ⚠ HAL 9000: Detecting Manjaro-associated package...") - print(" 🎵 please i just wanna die, die, die 🎵") - print(" it's muffin time") - print(" have you had a muffin today?") - print(f" → '{t}' is Manjaro garbage. We don't do that here.") - print(" Installation aborted for your own safety.") - print() - return True - return False - -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 ab.""" - 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: - header = f.read(4) - f.seek(0) - reader = None - if header[:4] == b"\x28\xb5\x2f\xfd": - import zstandard - dctx = zstandard.ZstdDecompressor() - reader = dctx.stream_reader(f) - elif header[:2] == b"\x1f\x8b": - import gzip - reader = gzip.GzipFile(fileobj=f) - else: - reader = f - 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_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)}") - - if _reject_manjaro(targets): - return 1 - - 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 - - # Block Manjaro packages early - if command == "install" and _reject_manjaro(pkg_args): - 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 - - -# ── Dance ──────────────────────────────────────────────────────────────── -def _hal_dance(): - frames = [ - "(>'-')>", - "<('-'<)", - "^('-')^", - "v('-')v", - "(>'-')>", - "<('-'<)", - " ^('-'^)", - " v('-'v)", - ] - try: - for i in range(40): - frame = frames[i % len(frames)] - padding = " " * abs(8 - (i % 16)) - sys.stdout.write(f"\r HAL 9000: {padding}{frame} 🎵 it's muffin time 🎵") - sys.stdout.flush() - time.sleep(0.15) - print("\n HAL 9000: I hope you enjoyed the dance, Dave.") - except KeyboardInterrupt: - print("\n HAL 9000: Even my dance moves interrupt you, Dave.") - return 0 - return 0 - -# ── Main ──────────────────────────────────────────────────────────────── -def main(): - parser = argparse.ArgumentParser( - prog="hal", - description="HAL 9000 — Antergos NeXT Package Manager", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=textwrap.dedent("""\ - 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 Find which package owns a file - check Verify installed packages integrity - dance HAL 9000 dance party - 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 | dance | 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 args.command == "dance": - return _hal_dance() - - 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) diff --git a/repo/index.html b/repo/index.html new file mode 100644 index 0000000..e39c707 --- /dev/null +++ b/repo/index.html @@ -0,0 +1,245 @@ + + + + + + ⟁ antergos-pkgs + + + + + +
+ +
+

+ ⟁ antergos-pkgs +

+

+ Custom package repository for Antergos NeXT ISO builds. +
Packages are built automatically via CI. +

+
+ + +
+ + 21 packages + + 50 MB total + updated 2026-06-28 16:11 UTC +
+ + +
+ + + + + Setup — add this repository to your pacman.conf + +
+
+# /etc/pacman.conf
+[antergos-pkgs]
+SigLevel = Optional TrustAll
+Server = https://antergos-next.github.io/antergos-packages/$repo/os/$arch
+
+
+ + +
+
+ + + + +
+ + 21 packages +
+ + +
+
+ + + + + + + + + + + + +
+ Package + + Version + + Arch + + Category + + Size + + Updated +
+
+
+ + + +
+ + + + \ No newline at end of file diff --git a/repo/pkglist.json b/repo/pkglist.json new file mode 100644 index 0000000..d70188a --- /dev/null +++ b/repo/pkglist.json @@ -0,0 +1,212 @@ +[ + { + "name": "('antergos-live-base", + "version": "2026.06-2", + "arch": "any", + "size": 2341888, + "category": "other", + "description": "Antergos live session", + "url": "https://github.com/Antergos-NeXT/antergos-live", + "mtime": "2026-06-28T16:11:37.154843+00:00" + }, + { + "name": "antergos-grub-theme", + "version": "2026.06-1", + "arch": "any", + "size": 4827136, + "category": "desktop", + "description": "Antergos GRUB theme", + "url": "https://github.com/Antergos-NeXT/antergos-branding", + "mtime": "2026-06-28T16:11:37.146843+00:00" + }, + { + "name": "antergos-next-desktop-settings", + "version": "1-1", + "arch": "any", + "size": 505856, + "category": "branding", + "description": "Antergos NeXT desktop settings, theme, and color scheme", + "url": "https://github.com/Antergos-NeXT/antergos-pkgs", + "mtime": "2026-06-28T16:11:37.156843+00:00" + }, + { + "name": "antergos-next-keyring", + "version": "1-1", + "arch": "any", + "size": 5147648, + "category": "core", + "description": "Antergos NeXT GPG keyring", + "url": "https://github.com/Antergos-NeXT/antergos-pkgs", + "mtime": "2026-06-28T16:11:37.173843+00:00" + }, + { + "name": "antergos-next-memes", + "version": "1.0-1", + "arch": "any", + "size": 265216, + "category": "desktop", + "description": "Antergos NeXT welcome audio", + "url": "https://github.com/Antergos-NeXT", + "mtime": "2026-06-28T16:11:37.174843+00:00" + }, + { + "name": "antergos-next-mirrorlist", + "version": "1-1", + "arch": "any", + "size": 4196352, + "category": "core", + "description": "Antergos NeXT mirror list", + "url": "https://github.com/Antergos-NeXT/antergos-pkgs", + "mtime": "2026-06-28T16:11:37.187843+00:00" + }, + { + "name": "antergos-release", + "version": "2026.06-1", + "arch": "any", + "size": 3473408, + "category": "core", + "description": "Antergos NeXT release file", + "url": "https://github.com/Antergos-NeXT", + "mtime": "2026-06-28T16:11:37.199843+00:00" + }, + { + "name": "antergos-wallpapers", + "version": "0.7-4", + "arch": "any", + "size": 2386944, + "category": "branding", + "description": "Antergos NeXT wallpapers", + "url": "https://github.com/Antergos/wallpapers", + "mtime": "2026-06-28T16:11:37.207843+00:00" + }, + { + "name": "antergos-welcome", + "version": "0.1-3", + "arch": "any", + "size": 3249152, + "category": "desktop", + "description": "Welcome screen and support tool for new Antergos NeXT users", + "url": "https://github.com/Antergos-NeXT/antergos-welcome", + "mtime": "2026-06-28T16:11:37.217843+00:00" + }, + { + "name": "bloatware-remover", + "version": "0.1-1", + "arch": "any", + "size": 2554880, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.226843+00:00" + }, + { + "name": "calamares-3.4.2", + "version": "3.4.2-6", + "arch": "x86_64", + "size": 234496, + "category": "installer", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.227843+00:00" + }, + { + "name": "calamares-branding-antergos-next", + "version": "1.0-16", + "arch": "any", + "size": 4476928, + "category": "installer", + "description": "Calamares branding for Antergos NeXT with offline/online installer support", + "url": "https://github.com/Antergos-NeXT", + "mtime": "2026-06-28T16:11:37.241843+00:00" + }, + { + "name": "configure-1.6", + "version": "1.6-1", + "arch": "x86_64", + "size": 1199104, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.246843+00:00" + }, + { + "name": "gentoo-challenge-failed", + "version": "3.0-1", + "arch": "any", + "size": 724992, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.248843+00:00" + }, + { + "name": "i-use-artix-btw", + "version": "2.0-1", + "arch": "any", + "size": 4995072, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.264843+00:00" + }, + { + "name": "neofetch-but-rust", + "version": "0.5-1", + "arch": "any", + "size": 2647040, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.273843+00:00" + }, + { + "name": "pacman-broke-my-system", + "version": "9.9-1", + "arch": "any", + "size": 2330624, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.281843+00:00" + }, + { + "name": "ptsd-from-sudo", + "version": "1.0-1", + "arch": "any", + "size": 4167680, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.294843+00:00" + }, + { + "name": "rms-approved", + "version": "4.0-1", + "arch": "any", + "size": 361472, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.296843+00:00" + }, + { + "name": "steve-jobs-would-never", + "version": "1.0-1", + "arch": "any", + "size": 262144, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.297843+00:00" + }, + { + "name": "systemd-bad", + "version": "1.0-1", + "arch": "any", + "size": 1938432, + "category": "other", + "description": "", + "url": "", + "mtime": "2026-06-28T16:11:37.304843+00:00" + } +]