"""AUR metadata via the RPC v5 API, plus local history for change detection. The API tells us the *current* maintainer, which is not enough to notice that a package changed hands. That needs a record of what we saw last time, so this module keeps a small state file: for each AUR package, the maintainer and the PKGBUILD commit we last reviewed. Everything downstream that says "adopted by a new maintainer" is a comparison against that record. """ from __future__ import annotations import json import re import shutil import subprocess import time import urllib.error import urllib.parse import urllib.request from dataclasses import asdict, dataclass, field from pathlib import Path RPC = "https://aur.archlinux.org/rpc/v5/info" #: the RPC is a GET with repeated arg[] params, so names are batched to keep the #: query string within sane limits BATCH = 150 @dataclass(frozen=True) class AurPackage: name: str version: str = "" description: str = "" maintainer: str | None = None comaintainers: tuple[str, ...] = () out_of_date: int | None = None first_submitted: int = 0 last_modified: int = 0 votes: int = 0 popularity: float = 0.0 url: str = "" package_base: str = "" depends: tuple[str, ...] = () make_depends: tuple[str, ...] = () @property def is_orphaned(self) -> bool: return not self.maintainer @property def is_flagged(self) -> bool: return bool(self.out_of_date) @property def age_days(self) -> float: return (time.time() - self.last_modified) / 86400 if self.last_modified else 0.0 @classmethod def from_rpc(cls, raw: dict) -> AurPackage: return cls( name=raw.get("Name", ""), version=raw.get("Version", ""), description=raw.get("Description") or "", maintainer=raw.get("Maintainer"), comaintainers=tuple(raw.get("CoMaintainers") or ()), out_of_date=raw.get("OutOfDate"), first_submitted=int(raw.get("FirstSubmitted") or 0), last_modified=int(raw.get("LastModified") or 0), votes=int(raw.get("NumVotes") or 0), popularity=float(raw.get("Popularity") or 0.0), url=raw.get("URL") or "", package_base=raw.get("PackageBase") or "", depends=tuple(raw.get("Depends") or ()), make_depends=tuple(raw.get("MakeDepends") or ()), ) class AurError(RuntimeError): pass def _default_opener(url: str, timeout: float): return urllib.request.urlopen(url, timeout=timeout) # noqa: S310 - fixed https host #: how long a batch of RPC answers may be reused. Short on purpose: this is the #: authoritative source for "what version is in the AUR", so staleness matters, #: but a drill-down asks the same question repeatedly within seconds. RPC_TTL = 300 def _rpc_cache_path(chunk: list[str]) -> Path: import hashlib from . import fetch key = hashlib.sha256("\n".join(chunk).encode()).hexdigest()[:16] return fetch.cache_dir() / "aur-rpc" / f"{key}.json" def fetch( names: list[str], *, opener=_default_opener, timeout: float = 20.0, use_cache: bool = True ) -> dict[str, AurPackage]: """Look up AUR packages by name. Names the AUR does not know are omitted. A missing name is meaningful rather than an error: it is how a package that was deleted from the AUR shows up. Batches are cached for a few minutes. The plan asks once for every foreign package, but opening a package asks again for that one alone, and the second question has the same answer as the first. """ out: dict[str, AurPackage] = {} unique = sorted(set(names)) for start in range(0, len(unique), BATCH): chunk = unique[start : start + BATCH] path = _rpc_cache_path(chunk) if use_cache else None payload = None if path is not None: try: if time.time() - path.stat().st_mtime < RPC_TTL: payload = json.loads(path.read_text()) except (OSError, ValueError): payload = None if payload is None: query = urllib.parse.urlencode([("arg[]", n) for n in chunk]) try: with opener(f"{RPC}?{query}", timeout) as response: payload = json.load(response) except (urllib.error.URLError, OSError, ValueError, TimeoutError) as exc: raise AurError(f"AUR RPC unavailable: {exc}") from exc if path is not None: try: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(payload)) tmp.replace(path) except (OSError, TypeError): pass # a cache that cannot be written is not a failure if not isinstance(payload, dict): # json.loads accepts a list or a scalar; the RPC contract does not, # and the AttributeError escaped as an unhandled crash. raise AurError("the AUR RPC returned something that is not an object") for raw in payload.get("results", []): pkg = AurPackage.from_rpc(raw) out[pkg.name] = pkg return out # -------------------------------------------------------------------------- # on-disk cache # -------------------------------------------------------------------------- def state_dir() -> Path: import os base = os.environ.get("XDG_STATE_HOME") or (Path.home() / ".local/state") return Path(base) / "updater" @dataclass class Entry: """What we knew about an AUR package the last time we looked at it.""" maintainer: str | None = None commit: str = "" version: str = "" @dataclass class State: """Local memory of AUR packages, so changes can be described as changes.""" path: Path = field(default_factory=lambda: state_dir() / "aur-state.json") entries: dict[str, Entry] = field(default_factory=dict) @classmethod def load(cls, path: Path | None = None) -> State: path = path or (state_dir() / "aur-state.json") try: raw = json.loads(path.read_text()) except (OSError, ValueError): return cls(path=path) if not isinstance(raw, dict): return cls(path=path) packages = raw.get("packages") if not isinstance(packages, dict): packages = {} entries = { name: Entry(**{k: v for k, v in body.items() if k in Entry.__dataclass_fields__}) for name, body in packages.items() if isinstance(body, dict) } return cls(path=path, entries=entries) def save(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) payload = {"version": 1, "packages": {n: asdict(e) for n, e in self.entries.items()}} tmp = self.path.with_suffix(".tmp") tmp.write_text(json.dumps(payload, indent=1, sort_keys=True)) tmp.replace(self.path) def previous_maintainer(self, name: str) -> str | None: """The maintainer recorded at the last successful build. `None` means no record at all; `""` means the package was orphaned when it was last seen. Conflating the two hid the case that matters most: adopting an orphaned package is the classic AUR account-takeover vector, and with both encoded as None there was nothing to compare against. """ entry = self.entries.get(name) if entry is None: return None return entry.maintainer or "" def previous_commit(self, name: str) -> str: entry = self.entries.get(name) return entry.commit if entry else "" def record( self, name: str, *, maintainer: str | None, commit: str = "", version: str = "" ) -> None: self.entries[name] = Entry(maintainer=maintainer, commit=commit, version=version) DEBUG_SUFFIX = "-debug" #: `name old -> new`, the shape every pacman-family query produces _UPGRADE_LINE = re.compile(r"^(\S+)\s+(\S+)\s+->\s+(\S+)\s*$") def devel_upgrades(*, timeout: int = 120) -> tuple[dict[str, tuple[str, str]], str]: """AUR packages paru itself would rebuild, and why if it could not say. A `-git` package's recorded pkgver is whatever the maintainer last committed, which is routinely *behind* the version paru generated the last time it built the thing. Comparing versions therefore says nothing: on this machine 18 of 21 devel packages differ, and not one of them differs in the direction that means "there is something new". Only paru knows. It keeps a development database recording the commit each devel package was built from and checks upstream for movement, and `-Qua` is that check without changing anything. Returns `(upgrades, note)`; `note` is empty on success and explains the shortfall otherwise. """ exe = shutil.which("paru") if exe is None: return {}, "paru is not installed, so devel packages cannot be checked" try: proc = subprocess.run( [exe, "-Qua"], capture_output=True, text=True, errors="replace", timeout=timeout, check=False, ) except subprocess.TimeoutExpired: return {}, f"paru -Qua timed out after {timeout}s" except OSError as exc: return {}, f"could not run paru: {exc}" out: dict[str, tuple[str, str]] = {} for line in proc.stdout.splitlines(): match = _UPGRADE_LINE.match(line.strip()) if match: out[match.group(1)] = (match.group(2), match.group(3)) if not out and proc.returncode != 0: detail = (proc.stderr.strip().splitlines() or ["no output"])[0] return {}, f"paru -Qua failed: {detail[:120]}" return out, "" def deleted( names: list[str], packages: dict[str, AurPackage], known_bases: set[str] | None = None ) -> list[str]: """Names the AUR no longer returns, i.e. packages that have been removed. `-debug` packages are split out by makepkg when a base package is built with debug symbols; they were never AUR entries of their own, so a missing one says nothing. Without this, every machine with debug packages installed reports phantom deletions. """ bases = set(known_bases or ()) | set(packages) out: list[str] = [] for name in names: if name in packages: continue if name.endswith(DEBUG_SUFFIX) and name[: -len(DEBUG_SUFFIX)] in bases: continue out.append(name) return sorted(out)