"""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)