arch_agent
150 lines
4.1 KiB
Python
150 lines
4.1 KiB
Python
"""AUR API client — fetches PKGBUILDs and metadata from the AUR."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
AUR_RPC = "https://aur.archlinux.org/rpc/v5"
|
|
AUR_CGIT = "https://aur.archlinux.org/cgit/aur.git/plain"
|
|
|
|
|
|
@dataclass
|
|
class AURPackage:
|
|
name: str
|
|
version: str
|
|
description: str
|
|
url: str
|
|
maintainer: str | None
|
|
num_votes: int
|
|
popularity: float
|
|
last_modified: int
|
|
pkgbase: str
|
|
|
|
|
|
@dataclass
|
|
class AURSource:
|
|
pkgbuild: str
|
|
srcinfo: str
|
|
package: AURPackage
|
|
|
|
|
|
async def aur_info(name: str) -> AURPackage | None:
|
|
"""Fetch package info from AUR RPC API."""
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.get(
|
|
AUR_RPC,
|
|
params=[
|
|
("type", "info"),
|
|
("v", "5"),
|
|
("arg[]", name),
|
|
],
|
|
)
|
|
if resp.status_code == 404:
|
|
return None
|
|
resp.raise_for_status()
|
|
data: dict[str, Any] = resp.json()
|
|
|
|
results = data.get("results", [])
|
|
if not results:
|
|
return None
|
|
|
|
r = results[0]
|
|
return AURPackage(
|
|
name=r.get("Name", name),
|
|
version=r.get("Version", ""),
|
|
description=r.get("Description", ""),
|
|
url=r.get("URL", ""),
|
|
maintainer=r.get("Maintainer"),
|
|
num_votes=r.get("NumVotes", 0),
|
|
popularity=r.get("Popularity", 0.0),
|
|
last_modified=r.get("LastModified", 0),
|
|
pkgbase=r.get("PackageBaseID", ""),
|
|
)
|
|
|
|
|
|
async def aur_sources(name: str) -> AURSource | None:
|
|
"""Fetch PKGBUILD and .SRCINFO for a package."""
|
|
pkg = await aur_info(name)
|
|
if pkg is None:
|
|
return None
|
|
|
|
pkgbase = r.get("PackageBase", name) if (r := await _raw_info(name)) else name
|
|
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
# Fetch PKGBUILD
|
|
pkgbuild_resp = await client.get(
|
|
AUR_CGIT + f"/PKGBUILD",
|
|
params={"h": pkgbase},
|
|
)
|
|
pkgbuild_resp.raise_for_status()
|
|
pkgbuild = pkgbuild_resp.text
|
|
|
|
# Fetch .SRCINFO
|
|
srcinfo_resp = await client.get(
|
|
AUR_CGIT + f"/.SRCINFO",
|
|
params={"h": pkgbase},
|
|
)
|
|
srcinfo = srcinfo_resp.text if srcinfo_resp.status_code == 200 else ""
|
|
|
|
return AURSource(
|
|
pkgbuild=pkgbuild,
|
|
srcinfo=srcinfo,
|
|
package=pkg,
|
|
)
|
|
|
|
|
|
async def _raw_info(name: str) -> dict[str, Any] | None:
|
|
"""Get raw RPC result for a package."""
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.get(
|
|
AUR_RPC,
|
|
params=[
|
|
("type", "info"),
|
|
("v", "5"),
|
|
("arg[]", name),
|
|
],
|
|
)
|
|
if resp.status_code == 404:
|
|
return None
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
results = data.get("results", [])
|
|
return results[0] if results else None
|
|
|
|
|
|
def extract_sources(pkgbuild: str) -> list[str]:
|
|
"""Extract source URLs from a PKGBUILD."""
|
|
sources = []
|
|
in_array = False
|
|
for line in pkgbuild.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("source="):
|
|
in_array = True
|
|
# single-line: source=(url)
|
|
m = re.findall(r'https?://[^\s)\'"]+|git://[^\s)\'"]+|ftp://[^\s)\'"]+',
|
|
stripped)
|
|
sources.extend(m)
|
|
if ")" in stripped and not stripped.endswith("("):
|
|
in_array = False
|
|
elif in_array:
|
|
m = re.findall(r'https?://[^\s)\'"]+|git://[^\s)\'"]+|ftp://[^\s)\'"]+',
|
|
stripped)
|
|
sources.extend(m)
|
|
if ")" in stripped:
|
|
in_array = False
|
|
return sources
|
|
|
|
|
|
def extract_install_hooks(pkgbuild: str) -> list[str]:
|
|
"""Extract post_install/pre_install hooks from PKGBUILD."""
|
|
hooks = []
|
|
funcs = re.findall(
|
|
r'(?:post_install|pre_install|post_upgrade|pre_upgrade|post_remove|pre_remove)\s*\(\)\s*\{[^}]*\}',
|
|
pkgbuild,
|
|
re.DOTALL,
|
|
)
|
|
return funcs |