aboutsummaryrefslogtreecommitdiff
path: root/utils/aur
diff options
context:
space:
mode:
Diffstat (limited to 'utils/aur')
-rw-r--r--utils/aur/AURUtils/__init__.py0
-rw-r--r--utils/aur/AURUtils/aur_deps.py9
-rw-r--r--utils/aur/AURUtils/errors.py16
-rw-r--r--utils/aur/AURUtils/rpc.py88
-rw-r--r--utils/aur/AURUtils/srcinfo.py97
-rw-r--r--utils/aur/AURUtils/types.py78
-rw-r--r--utils/aur/AURUtils/version.py5
-rwxr-xr-xutils/aur/check-aur-dependencies.py80
-rwxr-xr-xutils/aur/check-aur-update.py145
-rw-r--r--utils/aur/requirements.txt2
10 files changed, 0 insertions, 520 deletions
diff --git a/utils/aur/AURUtils/__init__.py b/utils/aur/AURUtils/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/utils/aur/AURUtils/__init__.py
+++ /dev/null
diff --git a/utils/aur/AURUtils/aur_deps.py b/utils/aur/AURUtils/aur_deps.py
deleted file mode 100644
index 3a4bc44..0000000
--- a/utils/aur/AURUtils/aur_deps.py
+++ /dev/null
@@ -1,9 +0,0 @@
-import re
-
-signs = [r'\>\=', r'\<\=', r'\=\=', r'\>', r'\<']
-signs_or = r'|'.join(signs)
-deps_pattern = re.compile(rf'({signs_or}).*')
-
-
-def depends_strip(deps: str) -> str:
- return deps_pattern.sub('', deps) \ No newline at end of file
diff --git a/utils/aur/AURUtils/errors.py b/utils/aur/AURUtils/errors.py
deleted file mode 100644
index 2e1fd87..0000000
--- a/utils/aur/AURUtils/errors.py
+++ /dev/null
@@ -1,16 +0,0 @@
-class RPCError(Exception):
-
- def __init__(self, method: str, message: str):
- self.method = method
- self.message = message
-
- def __str__(self):
- return f'Method({self.method}): {self.message}'
-
-
-class PackageNotFound(Exception):
- pass
-
-
-class MaxRetries(Exception):
- pass
diff --git a/utils/aur/AURUtils/rpc.py b/utils/aur/AURUtils/rpc.py
deleted file mode 100644
index 6120d41..0000000
--- a/utils/aur/AURUtils/rpc.py
+++ /dev/null
@@ -1,88 +0,0 @@
-import asyncio
-import random
-
-import aiohttp
-
-from .errors import RPCError, MaxRetries
-from .types import Results, UrlParameterType
-
-DEFAULT_AURWEB_VERSION = 5
-MAX_WORKERS = 10
-
-METHODS = {'search', 'info', 'multiinfo', 'msearch', 'suggest',
- 'suggest-pkgbase', 'get-comment-form'}
-FIELDS = {'name', 'name-desc', 'maintainer', 'depends',
- 'makedepends', 'optdepends', 'checkdepends'}
-
-
-class AsyncRPC:
- """
- https://wiki.archlinux.org/title/Aurweb_RPC_interface
- """
-
- RPC_URL = 'https://aur.archlinux.org/rpc/'
-
- def __init__(self, max_workers: int = MAX_WORKERS):
- self._client = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10))
- self.sem = asyncio.Semaphore(max_workers)
-
- async def __aenter__(self):
- return self
-
- async def __aexit__(self, exc_type, exc_val, exc_tb):
- await self.close()
-
- async def close(self):
- await self._client.close()
-
- async def _post(self, params: dict) -> dict:
- async with self.sem:
- for i in range(5):
- try:
- async with self._client.get(self.RPC_URL, params=params) as resp:
- resp.raise_for_status()
- return await resp.json()
- except aiohttp.ServerTimeoutError:
- await asyncio.sleep(3 * random.random())
- continue
- raise MaxRetries
-
- async def request(self,
- method: str,
- method_params: dict[str, UrlParameterType],
- ):
- params = {
- 'v': DEFAULT_AURWEB_VERSION,
- 'type': method,
- **method_params
- }
- json_response = await self._post(params)
- if 'error' in json_response:
- raise RPCError(method, json_response['error'])
- return Results.parse(json_response)
-
- async def search(self,
- arg: UrlParameterType,
- search_by: str = None,
- ):
- method_params = {
- 'by': search_by or [],
- 'arg': arg,
- }
- return await self.request('search', method_params)
-
- async def info(self,
- arg: UrlParameterType,
- ):
- return await self.multiinfo(arg)
-
- async def multiinfo(self,
- arg: UrlParameterType,
- ):
- method_params = {
- 'arg[]': arg,
- }
- return await self.request('multiinfo', method_params)
-
- async def msearch(self, arg: UrlParameterType):
- return await self.search(arg, search_by='maintainer')
diff --git a/utils/aur/AURUtils/srcinfo.py b/utils/aur/AURUtils/srcinfo.py
deleted file mode 100644
index 34d2240..0000000
--- a/utils/aur/AURUtils/srcinfo.py
+++ /dev/null
@@ -1,97 +0,0 @@
-import re
-from collections import defaultdict
-from itertools import chain
-from typing import Optional, Union
-
-from .errors import PackageNotFound
-from .aur_deps import depends_strip
-
-key_value_pattern = re.compile(r'(?P<KEY>.+?)=(?P<VALUE>.+)')
-multivalued_attrs = {
- 'source', 'provides', 'conflicts', 'depends', 'replaces', 'optdepends', 'makedepends', 'checkdepends',
- 'md5sums', 'sha1sums', 'sha224sums', 'sha256sums', 'sha384sums', 'sha512sums', 'arch'
-}
-
-
-class SRCINFO:
-
- def __init__(self, pkgbase: str, pkgbase_opt: dict, pkgs_opt: dict):
- self.pkgbase = pkgbase
- self.pkgbase_opt = pkgbase_opt
- self.pkgs_opt = pkgs_opt
-
- @classmethod
- def parse(cls, source: str):
- pkgbase = None
- pkgbase_opt = defaultdict(list)
- pkgs_opt = {}
- handler = None
- for l in source.splitlines():
- m = key_value_pattern.search(l.strip())
- if m is None:
- continue
- key, value = m.group('KEY').strip(), m.group('VALUE').strip()
- if key == 'pkgbase':
- pkgbase = value
- handler = pkgbase_opt
- elif key == 'pkgname':
- pkgs_opt[value] = defaultdict(list)
- handler = pkgs_opt[value]
- else:
- if key in multivalued_attrs:
- handler[key].append(value)
- else:
- handler[key] = value
- return SRCINFO(pkgbase, pkgbase_opt, pkgs_opt)
-
- @property
- def pkgs(self) -> dict:
- return {
- k: {i: self.get_pkg_opt(k, i) for i in chain(self.pkgbase_opt.keys(), v.keys())}
- for k, v in self.pkgs_opt.items()
- }
-
- @property
- def pkgs_name(self) -> list:
- return list(self.pkgs_opt.keys())
-
- def _chain_pkg_opt(self, pkgname: str, opt: str) -> chain:
- if pkgname not in self.pkgs_name:
- raise PackageNotFound(f'{pkgname} not found in {self.pkgbase}')
- return chain(self.pkgbase_opt.get(opt, []), self.pkgs_opt[pkgname].get(opt, []))
-
- def _chain_opt(self, opt: str) -> chain:
- return chain(self.pkgbase_opt.get(opt, []), *(i.get(opt, []) for i in self.pkgs_opt.values()))
-
- def get_pkg_opt(self, pkgname: str, opt: str) -> Union[list, str]:
- if pkgname not in self.pkgs_name:
- raise PackageNotFound(f'{pkgname} not found in {self.pkgbase}')
- if opt in multivalued_attrs:
- return list(self._chain_pkg_opt(pkgname, opt))
- return self.pkgbase_opt.get(opt)
-
- def get_opt(self, opt: str) -> Union[list, str]:
- if opt in multivalued_attrs:
- return list(self._chain_opt(opt))
- return self.pkgbase_opt.get(opt)
-
- @property
- def depends(self) -> list[str]:
- return list(map(depends_strip, self._chain_opt('depends')))
-
- @property
- def makedepends(self) -> list[str]:
- return list(map(depends_strip, self._chain_opt('makedepends')))
-
- @property
- def optdepends(self) -> list[str]:
- return list(self._chain_opt('optdepends'))
-
- @property
- def version(self) -> tuple[Optional[str], str, str]:
- return self.pkgbase_opt.get('epoch', None), \
- self.pkgbase_opt['pkgver'], self.pkgbase_opt['pkgrel']
-
- def get_version(self) -> str:
- epoch = f"{self.pkgbase_opt['epoch']}:" if self.pkgbase_opt.get('epoch') else ''
- return f"{epoch}{self.pkgbase_opt['pkgver']}-{self.pkgbase_opt['pkgrel']}"
diff --git a/utils/aur/AURUtils/types.py b/utils/aur/AURUtils/types.py
deleted file mode 100644
index 9830b08..0000000
--- a/utils/aur/AURUtils/types.py
+++ /dev/null
@@ -1,78 +0,0 @@
-from dataclasses import dataclass
-from datetime import datetime
-from collections.abc import Iterator
-from typing import Union
-
-from .aur_deps import depends_strip
-
-UrlParameterType = Union[str, list[str]]
-
-
-@dataclass()
-class Package:
-
- ID: int
- Name: str
- PackageBaseID: int
- PackageBase: str
- Version: str
- Description: str
- URL: str
- NumVotes: int
- Popularity: float
- OutOfDate: int
- Maintainer: str
- FirstSubmitted: int
- LastModified: int
- URLPath: str
- Conflicts: list[str] = None
- Provides: list[str] = None
- Depends: list[str] = None
- MakeDepends: list[str] = None
- OptDepends: list[str] = None
- License: list[str] = None
- Keywords: list[str] = None
-
- @property
- def depends(self) -> list[str]:
- return list(map(depends_strip, self.Depends))
-
- @property
- def makedepends(self) -> list[str]:
- return list(map(depends_strip, self.MakeDepends))
-
- @property
- def optdepends(self) -> list[str]:
- return list(self.OptDepends)
-
- @property
- def first_submitted(self) -> datetime:
- return datetime.fromtimestamp(self.FirstSubmitted)
-
- @property
- def last_modified(self) -> datetime:
- return datetime.fromtimestamp(self.LastModified)
-
- def get_version(self) -> str:
- return self.Version
-
-
-@dataclass
-class Results:
-
- resultcount: int
- results: list
- type: str
- version: int
-
- @classmethod
- def parse(cls, json_response: dict):
- return Results(**json_response)
-
- def empty(self) -> bool:
- return self.resultcount == 0
-
- def packages(self) -> Iterator[Package]:
- if self.type in {'multiinfo', 'search'}:
- return [Package(**i) for i in self.results]
- return []
diff --git a/utils/aur/AURUtils/version.py b/utils/aur/AURUtils/version.py
deleted file mode 100644
index b3d5329..0000000
--- a/utils/aur/AURUtils/version.py
+++ /dev/null
@@ -1,5 +0,0 @@
-import pyalpm # type: ignore
-
-
-def vercmp(ver1: str, ver2: str) -> int:
- return pyalpm.vercmp(ver1, ver2) \ No newline at end of file
diff --git a/utils/aur/check-aur-dependencies.py b/utils/aur/check-aur-dependencies.py
deleted file mode 100755
index 38b6436..0000000
--- a/utils/aur/check-aur-dependencies.py
+++ /dev/null
@@ -1,80 +0,0 @@
-#!/usr/bin/env python3
-# Copyright (C) 2021 KunoiSayami and contributors
-#
-# This module is part of KunoiSayami/repos and is released under
-# the AGPL v3 License: https://www.gnu.org/licenses/agpl-3.0.txt
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with this program. If not, see <https://www.gnu.org/licenses/>.
-import argparse
-import asyncio
-import heapq
-import logging
-import re
-from pathlib import Path
-
-import aiofiles
-import aiohttp
-
-
-async def check_pkg(session: aiohttp.ClientSession, num: int, pkg: str) -> tuple[int, bool]:
- logging.info(f'Checking {pkg}')
- ret = await session.head(f'https://aur.archlinux.org/cgit/aur.git/plain/PKGBUILD?h={pkg}')
- return num, ret.status == 200
-
-
-async def main() -> int:
- parser = argparse.ArgumentParser(description='check-aur-dependencies.py')
- parser.add_argument('paths', metavar='sub_path', type=str, nargs='*', help='sub-path for repo root')
-
- args = parser.parse_args()
- repo_dir = Path(*args.paths)
- pkgbuild_file = repo_dir.joinpath('PKGBUILD')
- if not pkgbuild_file.exists():
- logging.error("Can't locate PKGBUILD file, please specify directory in arguments or chdir to folder")
- return 1
-
- async with aiofiles.open(pkgbuild_file, 'r', encoding='utf-8') as f:
- pkgbuild_text = await f.read()
-
- runtime_deps_match = re.search(r"depends=\((?P<DEPS>([^)]|\n)*)\)", pkgbuild_text, re.M)
- if not runtime_deps_match:
- logging.warning("Can't found 'depends' keyword, if you think this is mistake, please report issue")
-
- make_deps_match = re.search(r"makedepends=\((?P<DEPS>([^)]|\n)*)\)", pkgbuild_text, re.M)
- if not make_deps_match:
- logging.warning("Can't found `makedepends' keyword, if you think this is mistake, please report issue")
-
- deps = [dep.strip('\'') for m in {runtime_deps_match, make_deps_match} if m for dep in m.group('DEPS').split()]
-
- async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10)) as session:
- tasks = [asyncio.create_task(check_pkg(session, n, pkg)) for n, pkg in enumerate(deps)]
- yay_dep_nums = []
- for task in asyncio.as_completed(tasks):
- n, r = await task
- if r:
- heapq.heappush(yay_dep_nums, n)
-
- yaydeps_dir = repo_dir.parent.joinpath('.yaydeps')
-
- if len(yay_dep_nums) > 0:
- async with aiofiles.open(yaydeps_dir.joinpath(repo_dir.stem), 'w', encoding='utf-8') as f:
- await f.write('\n'.join([*(deps[heapq.heappop(yay_dep_nums)] for _ in range(len(yay_dep_nums))), '']))
-
- return 0
-
-
-if __name__ == '__main__':
- logging.basicConfig(level=logging.DEBUG,
- format='%(asctime)s - %(levelname)s - %(funcName)s - %(lineno)d - %(message)s')
- exit(asyncio.run(main()))
diff --git a/utils/aur/check-aur-update.py b/utils/aur/check-aur-update.py
deleted file mode 100755
index 4be4e09..0000000
--- a/utils/aur/check-aur-update.py
+++ /dev/null
@@ -1,145 +0,0 @@
-#!/usr/bin/env python3
-
-import asyncio
-import argparse
-import heapq
-import logging
-from collections.abc import Coroutine
-from pathlib import Path
-from typing import Optional
-
-import aiohttp
-import aiofiles
-
-from AURUtils.srcinfo import SRCINFO
-from AURUtils.version import vercmp
-
-
-MAX_WORKERS = 16
-
-
-async def check_pkg(session: aiohttp.ClientSession, num: int, pkg: str) -> tuple[int, bool]:
- logging.info(f'Checking {pkg}')
- ret = await session.head(f'https://aur.archlinux.org/cgit/aur.git/plain/PKGBUILD?h={pkg}')
- return num, ret.status == 200
-
-
-async def check_aur_update(
- session: aiohttp.ClientSession,
- item: Path, dry_run: bool,
- parse_pkgbuild: Optional[Path] = None,
- check_deps: bool = True
-) -> None:
- if item.name.startswith('.') or not item.is_dir():
- return
- pkgbuild_file = item.joinpath('PKGBUILD')
- if not pkgbuild_file.exists():
- logging.warning(f'PKGBUILD not in {str(item)}, SKIP!')
- return
- srcinfo_file = item.joinpath('.SRCINFO')
- if not srcinfo_file.exists():
- if parse_pkgbuild.exists():
- logging.warning(f'.SRCINFO not in {str(item)}, try to parse')
- proc = await asyncio.create_subprocess_exec(
- f'{str(parse_pkgbuild.resolve())}',
- f'{str(pkgbuild_file)}',
- stdout=asyncio.subprocess.PIPE,
- )
- srcinfo = SRCINFO.parse((await proc.stdout.read()).decode())
- logging.info(f'extract .SRCINFO from {srcinfo.pkgbase}')
- else:
- logging.warning(f'.SRCINFO not in {str(item)}, no tool to parse')
- return
- else:
- async with aiofiles.open(srcinfo_file, 'r') as f:
- srcinfo = SRCINFO.parse(await f.read())
- version = srcinfo.get_version()
- async with session.get(f'https://aur.archlinux.org/cgit/aur.git/plain/.SRCINFO?h={srcinfo.pkgbase}') as resp:
- if resp.status != 200:
- logging.info(f'{item.name} not register in AUR')
- return
- new_srcinfo = SRCINFO.parse(await resp.text())
- new_version = new_srcinfo.get_version()
- result = vercmp(new_version, version)
- if result == 0:
- return
- elif result < 0:
- logging.warning(f'Warning: {item.name}({version}) is newer than AUR')
- else:
- if item.joinpath('.git').exists():
- if not dry_run:
- await asyncio.create_subprocess_exec('git', '-C', f'{str(item)}', 'pull', 'origin', 'master')
- logging.info(f'Upgrade {item.name} from {version} to {new_version}')
- else:
- logging.info(f'Found update {item.name}({new_version}) (local: {version})')
- if check_deps:
- deps = [*new_srcinfo.makedepends, *new_srcinfo.depends]
- async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10)) as session:
- tasks = [asyncio.create_task(check_pkg(session, n, pkg)) for n, pkg in enumerate(deps)]
- yay_dep_nums = []
- for task in asyncio.as_completed(tasks):
- n, r = await task
- if r:
- heapq.heappush(yay_dep_nums, n)
-
- yaydeps_dir = item.parent.joinpath('.yaydeps')
-
- if len(yay_dep_nums) > 0:
- async with aiofiles.open(yaydeps_dir.joinpath(item.stem), 'w', encoding='utf-8') as f:
- await f.write(
- '\n'.join([*(deps[heapq.heappop(yay_dep_nums)] for _ in range(len(yay_dep_nums))), '']))
-
-
-async def with_sem(sem: asyncio.Semaphore, coro: Coroutine):
- async with sem:
- return await coro
-
-
-async def main() -> int:
- parser = argparse.ArgumentParser(description='aur-check-update.py')
- parser.add_argument('--dry-run', dest='dry', help='dry run', action='store_true')
- parser.add_argument('--check-deps', dest='deps', help='check deps', action='store_true')
- parser.add_argument('--parse-pkgbuild', dest='script', metavar='SCRIPT',
- help='a script to parse pkgbuild',
- action='store')
- parser.add_argument('--miax-workers', dest='workers', metavar='MAX_WORKERS',
- help=f'aur check max workers, default: {MAX_WORKERS}',
- default=MAX_WORKERS, action='store', type=int)
- parser.add_argument(dest='repo_dir', metavar='REPO_DIR', help='repository directory', nargs='?')
-
- args = parser.parse_args()
-
- dry_run = args.dry
- check_deps = args.deps
- sem = asyncio.Semaphore(args.workers)
-
- if args.repo_dir is None:
- logging.error('repository directory not found')
- return 1
-
- script_file = Path(args.script)
- if args.script is None or not (script_file.is_file() and script_file.exists()):
- logging.info(f'Parse script {str(script_file)+" " if args.script is None else ""} not found')
- script_file = None
-
- repo_dir = Path(args.repo_dir)
-
- async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10)) as session:
- tasks = [asyncio.create_task(
- with_sem(sem, check_aur_update(
- session=session,
- item=item,
- dry_run=dry_run,
- parse_pkgbuild=script_file,
- check_deps=check_deps
- ))
- ) for item in repo_dir.iterdir()]
- await asyncio.gather(*tasks)
-
- return 0
-
-
-if __name__ == '__main__':
- logging.basicConfig(level=logging.DEBUG,
- format='%(asctime)s - %(levelname)s - %(funcName)s - %(lineno)d - %(message)s')
- asyncio.run(main())
diff --git a/utils/aur/requirements.txt b/utils/aur/requirements.txt
deleted file mode 100644
index 785d447..0000000
--- a/utils/aur/requirements.txt
+++ /dev/null
@@ -1,2 +0,0 @@
-aiofiles~=0.7.0
-aiohttp~=3.8.0
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-20 02:36:42 +0000