feat(script): add new aur check

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

8 files changed, 403 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
diff --git a/utils/aur/check-aur-update.py b/utils/aur/check-aur-update.py
new file mode 100644
index 0000000..85f02c2
--- /dev/null
+++ b/utils/aur/check-aur-update.py
@@ -0,0 +1,110 @@
+import asyncio
+import argparse
+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_aur_update(
+        session: aiohttp.ClientSession,
+        item: Path, dry_run: bool,
+        parse_pkgbuild: Optional[Path] = None,
+) -> 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,
+            )
+            async with aiofiles.open(srcinfo_file, 'w') as f:
+                await f.write((await proc.stdout.read()).decode())
+            logging.info(f'extract .SRCINFO in {str(item)}')
+        else:
+            logging.warning(f'.SRCINFO not in {str(item)}, no tool to parse')
+            return
+    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_version = SRCINFO.parse(await resp.text()).get_version()
+        result = vercmp(new_version, version)
+        if result == 0:
+            return
+        elif result < 0:
+            logging.warning(f'Warning: {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})')
+
+
+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', help='dry run', action='store_true')
+    parser.add_argument('--parse-pkgbuild', dest='script', metavar='SCRIPT',
+                        help='a script to parse pkgbuild',
+                        action='store')
+    parser.add_argument('--max-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
+    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, item, dry_run, script_file))
+        ) 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())
\ No newline at end of file