fix(pkg): Fix ksutils-teamspeak-management-tools

* refactor(utils): Clear unused script
* feat(utils): Add next generation of build script

Signed-off-by: KunoiSayami <[email protected]>

3 files changed, 128 insertions, 153 deletions

diff --git a/repo/ksutils-teamspeak-management-tools/[email protected] b/repo/ksutils-teamspeak-management-tools/[email protected]
index 8f59b9c..8f59b9c 100644
--- a/repo/ksutils-teamspeak-management-tools/[email protected]
+++ b/repo/ksutils-teamspeak-management-tools/[email protected]
diff --git a/utils/aur/aur-check-update.py b/utils/aur/aur-check-update.py
deleted file mode 100755
index f17f280..0000000
--- a/utils/aur/aur-check-update.py
+++ /dev/null
@@ -1,153 +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/>.
-from __future__ import annotations
-
-import argparse
-import asyncio
-import logging
-import re
-
-from collections.abc import Coroutine
-from dataclasses import dataclass
-from pathlib import Path
-
-import aiofiles
-import aiohttp
-
-MAX_WORKERS = 32
-
-
-@dataclass
-class Version:
-    pkgver: list[str]
-    pkgrel: list[str]
-    epoch: int
-
-    @classmethod
-    def from_str(cls, s: str) -> Version:
-        match = re.search(r"pkgver=(?P<VER>.*)", s)
-        if not match:
-            raise ValueError('pkgver keyword not found')
-        else:
-            pkgver = match.group('VER').split('.')
-        match = re.search(r"pkgrel=(?P<REL>.*)", s)
-        if not match:
-            pkgrel = ['1']
-        else:
-            pkgrel = match.group('REL').split('.')
-        match = re.search(r"epoch=(?P<EPOCH>\d+)", s)
-        if not match:
-            epoch = 0
-        else:
-            epoch = int(match.group('EPOCH'))
-        return cls(pkgver, pkgrel, epoch)
-
-    @classmethod
-    async def from_file(cls, file: str | Path) -> Version:
-        async with aiofiles.open(file, 'r', encoding='utf-8') as fin:
-            return cls.from_str(await fin.read())
-
-    def __eq__(self, other: Version) -> bool:
-        return self.epoch == other.epoch \
-               and self.pkgver == self.pkgver \
-               and self.pkgrel == other.pkgrel
-
-    def __gt__(self, other: Version) -> bool:
-        return self.epoch > other.epoch \
-               or self.pkgver > other.pkgver \
-               or (self.pkgver == other.pkgver and self.pkgrel > other.pkgrel)
-
-    def __ne__(self, other: Version) -> bool:
-        return not self.__eq__(other)
-
-    def __ge__(self, other: Version) -> bool:
-        return self.__gt__(other) or self.__eq__(other)
-
-    def __lt__(self, other: Version) -> bool:
-        return not self.__ge__(other)
-
-    def __str__(self) -> str:
-        prefix = f'{self.epoch}:' if self.epoch > 0 else ''
-        return f'{prefix}{self.pkgver}-{self.pkgrel}'
-
-
-async def with_sem(sem: asyncio.Semaphore, coro: Coroutine):
-    async with sem:
-        return await coro
-
-
-async def aur_check_update(session: aiohttp.ClientSession, item: Path, dry_run: bool) -> 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
-    version = await Version.from_file(pkgbuild_file)
-    async with session.get(f'https://aur.archlinux.org/cgit/aur.git/plain/PKGBUILD?h={item.name}') as resp:
-        if resp.status != 200:
-            logging.info(f'{item.name} not register in AUR')
-            return
-        new_version = Version.from_str(await resp.text())
-
-    if new_version == version:
-        return
-    elif new_version < version:
-        logging.warning(f'Warning: {version} is newer than AUR')
-        return
-    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}({version}) (local: {new_version})')
-
-
-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('--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
-
-    repo_dir = Path(args.repo_dir)
-
-    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10)) as session:
-        tasks = [asyncio.create_task(
-            with_sem(sem, aur_check_update(session, item, dry_run))
-        ) 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')
-    loop = asyncio.get_event_loop()
-    exit(loop.run_until_complete(main()))
diff --git a/utils/ng/pkgbuild.py b/utils/ng/pkgbuild.py
new file mode 100644
index 0000000..ed1a661
--- /dev/null
+++ b/utils/ng/pkgbuild.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python
+from __future__ import annotations
+import argparse
+import asyncio
+import dataclasses
+import lzma
+import logging
+import os
+import pathlib
+import subprocess
+
+import aiofiles
+
+logger = logging.getLogger()
+logger.setLevel(logging.DEBUG)
+
+
[email protected]
+class PackageVersion:
+    name: str
+    version: str
+
+    def __eq__(self, other: PackageVersion) -> bool:
+        return other.version == self.version
+
+    def __gt__(self, other: PackageVersion) -> bool:
+        return self.version > other.version
+
+    def __lt__(self, other: PackageVersion) -> bool:
+        return self.version < other.version
+
+    def __le__(self, other: PackageVersion) -> bool:
+        return self < other or self == other
+
+    def __ge__(self, other: PackageVersion) -> bool:
+        return self > other or self == other
+
+    def __ne__(self, other: PackageVersion) -> bool:
+        return not self == other
+
+    def __str__(self) -> str:
+        return f'{self.name} {self.version}'
+
+
+def fetch_database(pkg_db: str) -> dict[str, str]:
+    name_read = False
+    version_read = False
+    pkg_name = ''
+    package_storage = {}
+    with lzma.open(pkg_db) as fin:
+        for line in fin:
+            line = line.strip()
+            if name_read:
+                pkg_name = line.decode()
+                name_read = False
+            elif version_read:
+                package_storage.update({pkg_name: line.decode()})
+                #logger.debug(pkg_name, line.decode())
+                version_read = False
+                pkg_name = ''
+            if line == b"%NAME%":
+                assert not name_read
+                name_read = True
+            if line == b"%VERSION%":
+                assert not version_read
+                version_read = True
+            assert (name_read is False and version_read is False) or (version_read != name_read)
+    return package_storage
+
+
+def get_src_info(s: str) -> PackageVersion:
+    base = ''
+    ver = ''
+    rel = ''
+    for line in s.splitlines(False):
+        if (line := line.strip()).startswith('pkgver'):
+            ver = line.split('=', maxsplit=2)[1]
+        elif line.startswith('pkgrel'):
+            rel = line.split('=', maxsplit=2)[1]
+        elif line.startswith('pkgbase'):
+            base = line.split('=', maxsplit=2)[1]
+    return PackageVersion(base.strip(), f'{ver.strip()}-{rel.strip()}')
+
+
+async def parse_srcinfo(path: pathlib.PurePath) -> PackageVersion:
+    if (path := pathlib.Path(path.joinpath('.SRCINFO'))).is_file():
+        async with aiofiles.open(str(path)) as fin:
+            return get_src_info(await fin.read())
+    else:
+        p = await asyncio.create_subprocess_exec('makepkg', '--printsrcinfo', stdout=subprocess.PIPE, cwd=path.parent)
+        (stdout, stderr) = (await p.communicate())
+        return get_src_info(stdout.decode())
+
+
+async def do_work(pkg_db: str):
+
+    repo = fetch_database(pkg_db)
+    home_directory = pathlib.Path(os.getcwd())
+
+    tasks = []
+    pending_build = []
+
+    for folder in os.listdir("."):
+        if folder.startswith("."):
+            continue
+        if os.path.isfile(folder):
+            continue
+
+        tasks.append(asyncio.create_task(parse_srcinfo(home_directory.joinpath(folder))))
+
+    finished, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
+
+    for future in finished:
+        pkg = future.result()
+        if pkg.name in repo:
+            #print(pkg.name, repo[pkg.name], pkg.version)
+            if repo[pkg.name] < pkg.version:
+                pending_build.append(pkg.name)
+        else:
+            pending_build.append(pkg.name)
+
+    print(pending_build)
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='pkgbuild.py')
+    parser.add_argument('db_file')
+    args = parser.parse_args()
+    asyncio.run(do_work(args.db_file))
\ No newline at end of file