aboutsummaryrefslogtreecommitdiff
path: root/utils/aur/AURUtils
diff options
context:
space:
mode:
Diffstat (limited to 'utils/aur/AURUtils')
-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
7 files changed, 0 insertions, 293 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