feat(script): add new aur check

Signed-off-by: 67au <[email protected]>

7 files changed, 293 insertions, 0 deletions

diff --git a/utils/aur/AURUtils/__init__.py b/utils/aur/AURUtils/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/utils/aur/AURUtils/__init__.py
diff --git a/utils/aur/AURUtils/aur_deps.py b/utils/aur/AURUtils/aur_deps.py
new file mode 100644
index 0000000..3a4bc44
--- /dev/null
+++ b/utils/aur/AURUtils/aur_deps.py
@@ -0,0 +1,9 @@
+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
new file mode 100644
index 0000000..2e1fd87
--- /dev/null
+++ b/utils/aur/AURUtils/errors.py
@@ -0,0 +1,16 @@
+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
new file mode 100644
index 0000000..6120d41
--- /dev/null
+++ b/utils/aur/AURUtils/rpc.py
@@ -0,0 +1,88 @@
+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
new file mode 100644
index 0000000..34d2240
--- /dev/null
+++ b/utils/aur/AURUtils/srcinfo.py
@@ -0,0 +1,97 @@
+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
new file mode 100644
index 0000000..9830b08
--- /dev/null
+++ b/utils/aur/AURUtils/types.py
@@ -0,0 +1,78 @@
+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
new file mode 100644
index 0000000..b3d5329
--- /dev/null
+++ b/utils/aur/AURUtils/version.py
@@ -0,0 +1,5 @@
+import pyalpm  # type: ignore
+
+
+def vercmp(ver1: str, ver2: str) -> int:
+    return pyalpm.vercmp(ver1, ver2)
\ No newline at end of file