feat(pkg copy.py): add new file to handle package related commands in discord bot

This new file includes commands for searching and getting information about packages from the Arch User Repository (AUR) and official repositories. It also includes functionalities for filtering packages, formatting timestamps, and logging package details for debugging.

feat(pkg.py): add new package search functionality in discord bot

This commit introduces a new feature in the discord bot that allows users to search for packages in the Arch Linux and AUR repositories. The search results are displayed in a paginated format. The commit also includes the setup for the new 'Pkg' cog.

feat(buttons.py): add PkgSourceButton class to create a button for visiting source
feat(arch.py): add new file to handle interactions with Arch Linux package repository
feat(aur.py): add new file to handle interactions with Arch User Repository (AUR)

.archive/pkg copy.py

@@ -0,0 +1,285 @@
+from discord.ext import commands
+import discord
+from reactionmenu import ViewButton, ViewMenu
+from tux.wrappers.aur import AURClient
+from tux.wrappers.arch import ArchRepoClient
+from typing import List, Dict, Any
+from datetime import datetime
+import logging
+
+VOTE_EMOJI_ID = 1278954135374401566
+SOURCE_EMOJI_ID = 1278954134300655759
+OUT_OF_DATE_EMOJI_ID = 1273494919897681930
+REPO_EMOJI_ID = 1278964195550822490
+MAINTAINER_EMOJI_ID = 1278978044660289558
+
+
+def flatten_list(nested_list: List[Any]) -> List[Any]:
+    """Flatten a possibly nested list."""
+    flat_list = []
+    for item in nested_list:
+        if isinstance(item, list):
+            flat_list.extend(flatten_list(item))
+        else:
+            flat_list.append(item)
+    return flat_list
+
+
+class Pkg(commands.Cog):
+    def __init__(self, bot: commands.Bot) -> None:
+        self.bot = bot
+
+    @commands.hybrid_group(
+        name="pkg",
+        usage="pkg <subcommand>",
+    )
+    async def pkg(self, ctx: commands.Context[commands.Bot]) -> None:
+        """
+        Package related commands.
+
+        Parameters
+        ----------
+        ctx : commands.Context[commands.Bot]
+            The context object for the command.
+        """
+        if ctx.invoked_subcommand is None:
+            await ctx.send_help("pkg")
+
+    async def query_aur(self, package: str) -> List[Dict[str, Any]]:
+        """Query the AUR for a package."""
+        aur_client = AURClient()
+        search_result = await aur_client.search(package)
+        if isinstance(search_result, SearchResult):
+            results = search_result.results
+            for pkg in results:
+                pkg.repo = "aur"
+            return results
+        return []
+
+    async def query_arch_repos(self, package: str) -> List[Dict[str, Any]]:
+        """Query the official Arch Linux repositories for a package."""
+        arch_client = ArchRepoClient()
+        search_result = await arch_client.search_packages(name=package)
+        if search_result and search_result.valid:
+            results = search_result.results
+            for pkg in results:
+                pkg.repo = "arch"
+            return results
+        return []
+
+    def filter_packages(self, packages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Filter out packages containing 'i18n' in their name or description."""
+        return [
+            pkg
+            for pkg in packages
+            if "i18n" not in (pkg.get("Name") or pkg.get("pkgname", "")).lower()
+            and "i18n" not in (pkg.get("Description") or pkg.get("pkgdesc", "")).lower()
+        ]
+
+    def format_unix_timestamp(self, timestamp: int) -> str:
+        """Convert a UNIX timestamp to a formatted string."""
+        dt = datetime.utcfromtimestamp(timestamp)
+        return discord.utils.format_dt(dt, style="R")
+
+    def format_iso8601_date(self, date_str: str) -> str:
+        """Convert an ISO 8601 date string to a formatted string."""
+        dt = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
+        return discord.utils.format_dt(dt, style="f")
+
+    async def log_package_details(self, package: Dict[str, Any]) -> None:
+        """Log package details for debugging."""
+        logging.info(f"Package Details:\n{package}")
+
+    @pkg.command(
+        name="search",
+        usage="pkg search <package name>",
+        aliases=["s"],
+    )
+    async def search(self, ctx: commands.Context[commands.Bot], package: str) -> None:
+        """
+        Search for a package on the Arch User Repository (AUR) and official repositories.
+
+        Parameters
+        ----------
+        ctx : commands.Context[commands.Bot]
+            The discord context object.
+        package : str
+            The name of the package to search for.
+        """
+        # Get the emojis
+        vote_emoji = self.bot.get_emoji(VOTE_EMOJI_ID)
+        source_emoji = self.bot.get_emoji(SOURCE_EMOJI_ID)
+        out_of_date_emoji = self.bot.get_emoji(OUT_OF_DATE_EMOJI_ID)
+        repo_emoji = self.bot.get_emoji(REPO_EMOJI_ID)
+        maintainer_emoji = self.bot.get_emoji(MAINTAINER_EMOJI_ID)
+
+        aur_results = await self.query_aur(package)
+        arch_results = await self.query_arch_repos(package)
+
+        combined_results = self.filter_packages(flatten_list(aur_results + arch_results))
+
+        if combined_results:
+            # Ensure all pkgnames are available
+            sorted_results = sorted(
+                combined_results,
+                key=lambda pkg: (pkg.get("pkgname") or pkg.get("Name", "").lower()) == package.lower(),
+                reverse=True,
+            )
+
+            pages = []
+            embed = discord.Embed(
+                title="Package Search Results",
+                description=f"Results for '{package}'",
+                color=discord.Color.blurple(),
+            )
+
+            for pkg in sorted_results:
+                # Log package information for debugging
+                await self.log_package_details(pkg)
+
+                # Handle missing or None values by skipping the field
+                description = pkg.get("Description") or pkg.get("pkgdesc") or "No description provided"
+                votes = pkg.get("NumVotes")
+                source_url = pkg.get("URL") or pkg.get("url")
+                maintainer = pkg.get("Maintainer") or pkg.get("packager")
+                out_of_date = pkg.get("OutOfDate")
+                repo = pkg.get("repo") or "Unknown"
+                name = pkg.get("Name") or pkg.get("pkgname") or "No Name"
+
+                pkg_description = f"{repo_emoji} **{repo}** | "
+
+                if source_url:
+                    pkg_description += f"{source_emoji} **[Source]({source_url})**"
+                if maintainer and maintainer != "Unknown":
+                    pkg_description += f" | {maintainer_emoji} {maintainer}"
+                if out_of_date:
+                    pkg_description += f"\n{out_of_date_emoji} **OUT OF DATE**"
+
+                pkg_description += f"\n> {description}"
+                if repo == "aur" and votes is not None:
+                    pkg_description += f" | {vote_emoji} **Votes**: {votes}"
+                pkg_description += "\n"
+
+                if len(embed.fields) < 5:  # To limit 5 packages per embed
+                    embed.add_field(name=name, value=pkg_description, inline=False)
+                else:
+                    pages.append(embed)
+                    embed = discord.Embed(
+                        title="Package Search Results",
+                        description=f"Results for '{package}'",
+                        color=discord.Color.blurple(),
+                    )
+                    embed.add_field(name=name, value=pkg_description, inline=False)
+
+            if embed.fields:
+                pages.append(embed)
+
+            menu = ViewMenu(ctx, menu_type=ViewMenu.TypeEmbed)
+
+            for page in pages:
+                menu.add_page(page)
+
+            menu.add_button(ViewButton.go_to_first_page())
+            menu.add_button(ViewButton.back())
+            menu.add_button(ViewButton.next())
+            menu.add_button(ViewButton.go_to_last_page())
+            menu.add_button(ViewButton.end_session())
+
+            await menu.start()
+        else:
+            await ctx.send("No packages found.")
+
+    @pkg.command(name="info", usage="pkg info <package name>", aliases=["i"])
+    async def info(self, ctx: commands.Context, package: str) -> None:
+        """Get information about a package on the Arch User Repository (AUR) or official repositories."""
+        # Get the emojis
+        vote_emoji = self.bot.get_emoji(VOTE_EMOJI_ID)
+        source_emoji = self.bot.get_emoji(SOURCE_EMOJI_ID)
+        out_of_date_emoji = self.bot.get_emoji(OUT_OF_DATE_EMOJI_ID)
+        repo_emoji = self.bot.get_emoji(REPO_EMOJI_ID)
+        maintainer_emoji = self.bot.get_emoji(MAINTAINER_EMOJI_ID)
+
+        aur_client = AURClient()
+        arch_client = ArchRepoClient()
+
+        # Fetch from AUR
+        aur_results = await aur_client.get_package_info(package)
+        # Fetch from official repos with exact match
+        arch_results = await arch_client.search_packages(name=package)
+
+        combined_results = self.filter_packages(flatten_list(aur_results.results + arch_results.results))
+
+        if combined_results:
+            pkg = combined_results[0]
+
+            # Exact match logic
+            for p in combined_results:
+                if (p.get("pkgname") or p.get("Name", "").lower()) == package.lower():
+                    pkg = p
+                    break
+
+            # Logging for debugging
+            logging.info(f"Package: {pkg}")
+            logging.info(f"Combined Results: {combined_results}")
+
+            repo = pkg.get("repo") or "Unknown"
+            version = f"{pkg.get('Version', pkg.get('pkgver', ''))}-{pkg.get('pkgrel', '')}".strip("-")
+            maintainer = pkg.get("Maintainer") or pkg.get("packager")
+            description = pkg.get("Description") or pkg.get("pkgdesc") or "No description provided"
+            votes = pkg.get("NumVotes")
+            source_url = pkg.get("URL") or pkg.get("url")
+            out_of_date = pkg.get("OutOfDate")
+            first_submitted = pkg.get("FirstSubmitted") or pkg.get("build_date")
+            last_modified = pkg.get("LastModified") or pkg.get("last_update")
+            licenses = pkg.get("License") or pkg.get("licenses", [])
+            groups = pkg.get("Groups") or pkg.get("groups", [])
+
+            if first_submitted:
+                if isinstance(first_submitted, int):
+                    first_submitted = self.format_unix_timestamp(first_submitted)
+                else:
+                    first_submitted = self.format_iso8601_date(first_submitted)
+
+            if last_modified:
+                if isinstance(last_modified, int):
+                    last_modified = self.format_unix_timestamp(last_modified)
+                else:
+                    last_modified = self.format_iso8601_date(last_modified)
+
+            embed = discord.Embed(
+                title=f"{pkg.get('Name', pkg.get('pkgname', 'No Name'))}",
+                description=f"{repo_emoji} **{repo}** | ",
+                color=discord.Color.blurple(),
+            )
+
+            # Conditional field additions
+            if source_url:
+                embed.description += f"{source_emoji} **[Source]({source_url})**"
+            if maintainer and maintainer != "Unknown":
+                embed.description += f" | {maintainer_emoji} {maintainer}"
+            if out_of_date:
+                embed.description += f"\n{out_of_date_emoji} **OUT OF DATE**"
+            if repo == "aur" and votes is not None:
+                embed.description += f"| {vote_emoji} **Votes**: {votes}"
+
+            embed.description += f"\n> {description}\n"
+
+            # Additional fields
+            if version:
+                embed.add_field(name="Version", value=version, inline=False)
+            if first_submitted:
+                embed.add_field(name="First Submitted", value=first_submitted, inline=False)
+            if last_modified:
+                embed.add_field(name="Last Modified", value=last_modified, inline=False)
+            if licenses:
+                embed.add_field(name="Licenses", value=", ".join(licenses), inline=False)
+            if groups:
+                embed.add_field(name="Groups", value=", ".join(groups), inline=False)
+
+            await ctx.send(embed=embed)
+        else:
+            await ctx.send("No package found.")
+
+
+async def setup(bot: commands.Bot):
+    await bot.add_cog(Pkg(bot))

tux/cogs/utility/pkg.py

@@ -0,0 +1,312 @@
+from datetime import datetime
+from typing import Any
+
+import discord
+import pytz
+from discord.ext import commands
+from loguru import logger
+from reactionmenu import ViewButton, ViewMenu
+
+from tux.wrappers.arch import ArchRepoClient
+from tux.wrappers.aur import AURClient, ErrorResult
+
+VOTE_EMOJI_ID = 1278954135374401566
+SOURCE_EMOJI_ID = 1278954134300655759
+OUT_OF_DATE_EMOJI_ID = 1273494919897681930
+REPO_EMOJI_ID = 1278964195550822490
+MAINTAINER_EMOJI_ID = 1278978044660289558
+LICENSE_EMOJI_ID = 1279086563409531004
+VERSION_EMOJI_ID = 1279090224097656923
+LATEST_EMOJI_ID = 1279093553674457120
+POPULARITY_EMOJI_ID = 1278954133138702378
+
+
+class Pkg(commands.Cog):
+    def __init__(self, bot: commands.Bot) -> None:
+        self.bot = bot
+
+    @commands.hybrid_group(
+        name="pkg",
+        usage="pkg <subcommand>",
+    )
+    async def pkg(self, ctx: commands.Context[commands.Bot]) -> None:
+        if ctx.invoked_subcommand is None:
+            await ctx.send_help("pkg")
+
+    @staticmethod
+    def format_unix_timestamp(timestamp: int, tz: str = "UTC") -> str:
+        dt = datetime.fromtimestamp(timestamp, pytz.timezone(tz))
+        return discord.utils.format_dt(dt, style="R")
+
+    @staticmethod
+    def format_iso8601_date(date_str: str, tz: str = "UTC") -> str:
+        dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(pytz.timezone(tz))
+        return discord.utils.format_dt(dt, style="R")
+
+    async def fetch_aur_package_info(self, term: str) -> list[dict[str, Any]]:
+        try:
+            result = await AURClient.search(term)
+            if isinstance(result, ErrorResult):
+                return []
+            return [pkg.model_dump() for pkg in result.results]
+        except Exception as e:
+            logger.error(e)
+            return []
+
+    async def fetch_arch_package_info(self, term: str) -> list[dict[str, Any]]:
+        try:
+            pkg_search_result = await ArchRepoClient.search_package(term) or await ArchRepoClient.search_package(
+                term,
+                "any",
+            )
+            return [pkg_search_result.model_dump()] if pkg_search_result else []
+        except Exception as e:
+            logger.error(e)
+            return []
+
+    def build_embed(self, pkg_detail: dict[str, Any], repo: str) -> discord.Embed:
+        vote_emoji = self.bot.get_emoji(VOTE_EMOJI_ID)
+        source_emoji = self.bot.get_emoji(SOURCE_EMOJI_ID)
+        out_of_date_emoji = self.bot.get_emoji(OUT_OF_DATE_EMOJI_ID)
+        repo_emoji = self.bot.get_emoji(REPO_EMOJI_ID)
+        maintainer_emoji = self.bot.get_emoji(MAINTAINER_EMOJI_ID)
+        license_emoji = self.bot.get_emoji(LICENSE_EMOJI_ID)
+        version_emoji = self.bot.get_emoji(VERSION_EMOJI_ID)
+        latest_emoji = self.bot.get_emoji(LATEST_EMOJI_ID)
+        popularity_emoji = self.bot.get_emoji(POPULARITY_EMOJI_ID)
+
+        embed = discord.Embed(
+            title=pkg_detail.get("Name") or pkg_detail.get("pkgname"),
+            description=f"> {pkg_detail.get('Description') or pkg_detail.get('pkgdesc')}",
+            color=discord.Color.blue(),
+        )
+
+        if repo.lower() == "aur":
+            embed = self.add_aur_fields(
+                embed,
+                pkg_detail,
+                repo_emoji,
+                source_emoji,
+                latest_emoji,
+                vote_emoji,
+                popularity_emoji,
+                out_of_date_emoji,
+                version_emoji,
+                license_emoji,
+                maintainer_emoji,
+            )
+        else:
+            embed = self.add_arch_fields(
+                embed,
+                pkg_detail,
+                repo_emoji,
+                source_emoji,
+                latest_emoji,
+                version_emoji,
+                maintainer_emoji,
+                license_emoji,
+            )
+
+        return embed
+
+    def add_aur_fields(
+        self,
+        embed: discord.Embed,
+        pkg_detail: dict[str, Any],
+        repo_emoji: discord.Emoji | None,
+        source_emoji: discord.Emoji | None,
+        latest_emoji: discord.Emoji | None,
+        vote_emoji: discord.Emoji | None,
+        popularity_emoji: discord.Emoji | None,
+        out_of_date_emoji: discord.Emoji | None,
+        version_emoji: discord.Emoji | None,
+        license_emoji: discord.Emoji | None,
+        maintainer_emoji: discord.Emoji | None,
+    ) -> discord.Embed:
+        embed.add_field(name=f"{repo_emoji} Repo", value="aur", inline=True)
+        embed.add_field(name=f"{source_emoji} Source", value=f"[View]({pkg_detail['URL']})", inline=True)
+        embed.add_field(
+            name=f"{latest_emoji} Last Updated",
+            value=self.format_unix_timestamp(pkg_detail["LastModified"]),
+            inline=True,
+        )
+        embed.add_field(name=f"{vote_emoji} Votes", value=str(pkg_detail["NumVotes"]), inline=True)
+        embed.add_field(name=f"{popularity_emoji} Popularity", value=str(pkg_detail["Popularity"]), inline=True)
+        embed.add_field(
+            name=f"{out_of_date_emoji} Out of Date",
+            value="Yes" if pkg_detail["OutOfDate"] else "No",
+            inline=True,
+        )
+        embed.add_field(name=f"{version_emoji} Version", value=pkg_detail["Version"].split("+")[0] + "...", inline=True)
+        embed.add_field(name=f"{license_emoji} License", value=", ".join(pkg_detail.get("License", [])), inline=True)
+
+        if maintainers := self.get_maintainers(pkg_detail, "Maintainer", "CoMaintainers"):
+            embed.add_field(name=f"{maintainer_emoji} Maintainers", value=", ".join(maintainers), inline=True)
+
+        return embed
+
+    def add_arch_fields(
+        self,
+        embed: discord.Embed,
+        pkg_detail: dict[str, Any],
+        repo_emoji: discord.Emoji | None,
+        source_emoji: discord.Emoji | None,
+        latest_emoji: discord.Emoji | None,
+        version_emoji: discord.Emoji | None,
+        maintainer_emoji: discord.Emoji | None,
+        license_emoji: discord.Emoji | None,
+    ) -> discord.Embed:
+        embed.add_field(name=f"{repo_emoji} Repo", value=pkg_detail.get("repo"), inline=True)
+        embed.add_field(name=f"{source_emoji} Source", value=f"[View]({pkg_detail['url']})", inline=True)
+        embed.add_field(
+            name=f"{latest_emoji} Last Updated",
+            value=self.format_iso8601_date(pkg_detail["last_update"]),
+            inline=True,
+        )
+        embed.add_field(
+            name=f"{version_emoji} Version",
+            value=f"{pkg_detail['pkgver']}-{pkg_detail['pkgrel']}",
+            inline=True,
+        )
+
+        if maintainers := self.get_maintainers(pkg_detail, "packager", "maintainers"):
+            embed.add_field(name=f"{maintainer_emoji} Maintainers", value=", ".join(maintainers), inline=True)
+
+        embed.add_field(name=f"{license_emoji} License", value=", ".join(pkg_detail.get("licenses", [])), inline=True)
+
+        return embed
+
+    @staticmethod
+    def get_maintainers(pkg_detail: dict[str, Any], primary_key: str, secondary_key: str) -> list[str]:
+        maintainers: list[str] = []
+        if primary := pkg_detail.get(primary_key):
+            maintainers.append(primary)
+        if secondary := pkg_detail.get(secondary_key):
+            maintainers.extend(secondary)
+        return list(set(maintainers))
+
+    @pkg.command(
+        name="info",
+        usage="pkg info [term] [repo]",
+    )
+    async def info(
+        self,
+        ctx: commands.Context[commands.Bot],
+        term: str,
+        repo: str,
+    ) -> None:
+        if ctx.guild is None:
+            await ctx.send("This command can only be used in a server.")
+            return
+
+        pkg_detail: dict[str, Any] | None = None
+
+        if repo.lower() == "aur":
+            aur_results = await self.fetch_aur_package_info(term)
+            if aur_results:
+                pkg_detail = aur_results[0]
+        elif repo.lower() == "arch":
+            arch_results = await self.fetch_arch_package_info(term)
+            if arch_results:
+                pkg_detail = arch_results[0]
+        else:
+            await ctx.send("Invalid repo. Use `aur` or `arch`.")
+            return
+
+        if not pkg_detail:
+            await ctx.send("No results found.")
+            return
+
+        embed = self.build_embed(pkg_detail, repo)
+        await ctx.send(embed=embed)
+
+    def build_paginated_results(
+        self,
+        combined_results: list[dict[str, Any]],
+        package: str,
+    ) -> list[discord.Embed]:
+        vote_emoji = self.bot.get_emoji(VOTE_EMOJI_ID)
+        source_emoji = self.bot.get_emoji(SOURCE_EMOJI_ID)
+        out_of_date_emoji = self.bot.get_emoji(OUT_OF_DATE_EMOJI_ID)
+        repo_emoji = self.bot.get_emoji(REPO_EMOJI_ID)
+        maintainer_emoji = self.bot.get_emoji(MAINTAINER_EMOJI_ID)
+
+        pages: list[discord.Embed] = []
+        embed = discord.Embed(
+            title="Package Search Results",
+            description=f"Results for '{package}'",
+            color=discord.Color.blurple(),
+        )
+
+        for pkg in combined_results:
+            description = pkg.get("Description") or pkg.get("pkgdesc") or "No description provided"
+            votes = pkg.get("NumVotes")
+            source_url = pkg.get("URL") or pkg.get("url")
+            maintainer = pkg.get("Maintainer") or pkg.get("packager")
+            out_of_date = pkg.get("OutOfDate")
+            repo = pkg.get("repo") or "Unknown"
+            name = pkg.get("Name") or pkg.get("pkgname") or "No Name"
+
+            pkg_description = f"{repo_emoji} **{repo}** | "
+
+            if source_url:
+                pkg_description += f"{source_emoji} **[Source]({source_url})**"
+            if maintainer and maintainer != "Unknown":
+                pkg_description += f" | {maintainer_emoji} {maintainer}"
+            if out_of_date:
+                pkg_description += f"\n{out_of_date_emoji} **OUT OF DATE**"
+
+            pkg_description += f"\n> {description}"
+            if repo == "aur" and votes is not None:
+                pkg_description += f" | {vote_emoji} **Votes**: {votes}"
+            pkg_description += "\n"
+
+            if len(embed.fields) >= 5:
+                pages.append(embed)
+                embed = discord.Embed(
+                    title="Package Search Results",
+                    description=f"Results for '{package}'",
+                    color=discord.Color.blurple(),
+                )
+            embed.add_field(name=name, value=pkg_description, inline=False)
+        if embed.fields:
+            pages.append(embed)
+
+        return pages
+
+    @pkg.command(
+        name="search",
+        usage="pkg search <term>",
+    )
+    async def search_paginated(
+        self,
+        ctx: commands.Context[commands.Bot],
+        term: str,
+    ) -> None:
+        if ctx.guild is None:
+            await ctx.send("This command can only be used in a server.")
+            return
+
+        aur_results = await self.fetch_aur_package_info(term)
+        arch_results = await self.fetch_arch_package_info(term)
+
+        if combined_results := aur_results + arch_results:
+            pages = self.build_paginated_results(combined_results, term)
+
+            menu = ViewMenu(ctx, menu_type=ViewMenu.TypeEmbed)
+            for page in pages:
+                menu.add_page(page)
+
+            menu.add_button(ViewButton.go_to_first_page())
+            menu.add_button(ViewButton.back())
+            menu.add_button(ViewButton.next())
+            menu.add_button(ViewButton.go_to_last_page())
+            menu.add_button(ViewButton.end_session())
+
+            await menu.start()
+        else:
+            await ctx.send("No packages found.")
+
+
+async def setup(bot: commands.Bot):
+    await bot.add_cog(Pkg(bot))

tux/ui/buttons.py

@@ -18,3 +18,11 @@ class GithubButton(discord.ui.View):
         self.add_item(
             discord.ui.Button(style=discord.ButtonStyle.link, label="View on Github", url=url),
         )
+
+
+class PkgSourceButton(discord.ui.View):
+    def __init__(self, url: str) -> None:
+        super().__init__()
+        self.add_item(
+            discord.ui.Button(style=discord.ButtonStyle.link, label="Visit source", url=url),
+        )

tux/wrappers/arch.py

@@ -0,0 +1,93 @@
+from typing import Any
+
+import httpx
+from loguru import logger
+from pydantic import BaseModel
+
+
+class Package(BaseModel):
+    pkgname: str
+    pkgbase: str
+    repo: str
+    arch: str
+    pkgver: str
+    pkgrel: str
+    epoch: int
+    pkgdesc: str
+    url: str | None
+    filename: str
+    compressed_size: int
+    installed_size: int
+    build_date: str
+    last_update: str
+    flag_date: str | None
+    maintainers: list[str]
+    packager: str | None
+    groups: list[str]
+    licenses: list[str]
+    conflicts: list[str]
+    provides: list[str]
+    replaces: list[str]
+    depends: list[str]
+    optdepends: list[str]
+    makedepends: list[str]
+    checkdepends: list[str]
+
+
+class SearchResult(BaseModel):
+    version: int
+    limit: int
+    valid: bool
+    results: list[Package]
+    num_pages: int | None = None
+    page: int | None = None
+
+
+class ArchRepoClient:
+    BASE_URL = "https://archlinux.org"
+
+    @staticmethod
+    async def get_arch_pkg_details(term: str, arch: str = "x86_64") -> dict[str, Any] | None:
+        try:
+            pkg_search_result = await ArchRepoClient.search_package(term, arch)
+            if not pkg_search_result:
+                pkg_search_result = await ArchRepoClient.search_package(term, "any")
+                if not pkg_search_result:
+                    return None
+                arch = "any"
+
+            repo_name = pkg_search_result.repo
+            pkg_result = await ArchRepoClient.get_package_details(repo_name, arch, term)
+            return pkg_result.model_dump()
+        except Exception as e:
+            logger.error(e)
+            return None
+
+    @staticmethod
+    async def get_package_details(repo: str, arch: str, package: str) -> Package:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{ArchRepoClient.BASE_URL}/packages/{repo}/{arch}/{package}/json/")
+            logger.debug(f"Received package details response: {response.json()}")
+            response.raise_for_status()
+            return Package(**response.json())
+
+    @staticmethod
+    async def search_package(term: str, arch: str = "x86_64") -> Package | None:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{ArchRepoClient.BASE_URL}/packages/search/json/",
+                params={"name": term, "arch": arch},
+            )
+            logger.debug(f"Received search response: {response.json()}")
+            response.raise_for_status()
+            search_result = SearchResult(**response.json())
+            logger.debug(f"Search result: {search_result}")
+            return search_result.results[0] if search_result.results else None
+
+    @staticmethod
+    async def get_package_files(repo: str, arch: str, package: str) -> dict[str, Any]:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{ArchRepoClient.BASE_URL}/packages/{repo}/{arch}/{package}/files/json/")
+            logger.debug(f"Received package files response: {response.json()}")
+            response.raise_for_status()
+            return response.json()

tux/wrappers/aur.py

@@ -0,0 +1,127 @@
+import httpx
+from loguru import logger
+from pydantic import BaseModel, Field, RootModel
+
+
+class BaseResult(BaseModel):
+    resultcount: int
+    type: str
+    version: int
+
+
+class PackageBasic(BaseModel):
+    ID: int
+    Name: str | None = None
+    Description: str | None = None
+    PackageBaseID: int | None = None
+    PackageBase: str | None = None
+    Maintainer: str | None = None
+    NumVotes: int | None = None
+    Popularity: float | None = None
+    FirstSubmitted: int | None = None
+    LastModified: int | None = None
+    OutOfDate: int | None = None
+    Version: str | None = None
+    URLPath: str | None = None
+    URL: str | None = None
+
+
+class PackageDetailed(PackageBasic):
+    Submitter: str | None = None
+    License: list[str] = Field(default_factory=list)
+    Depends: list[str] = Field(default_factory=list)
+    MakeDepends: list[str] = Field(default_factory=list)
+    OptDepends: list[str] = Field(default_factory=list)
+    CheckDepends: list[str] = Field(default_factory=list)
+    Provides: list[str] = Field(default_factory=list)
+    Conflicts: list[str] = Field(default_factory=list)
+    Replaces: list[str] = Field(default_factory=list)
+    Groups: list[str] = Field(default_factory=list)
+    Keywords: list[str] = Field(default_factory=list)
+    CoMaintainers: list[str] = Field(default_factory=list)
+
+
+class SearchResult(BaseResult):
+    results: list[PackageBasic]
+
+
+class InfoResult(BaseResult):
+    results: list[PackageDetailed]
+
+
+class ErrorResult(BaseResult):
+    error: str
+    results: list[dict[str, str]] = Field(default_factory=list)
+
+
+class PackageNames(RootModel[list[str]]):
+    @classmethod
+    def from_list(cls, lst: list[str]):
+        return cls(root=lst)
+
+
+class AURClient:
+    BASE_URL = "https://aur.archlinux.org"
+
+    @staticmethod
+    async def search(term: str, search_by: str = "name-desc") -> SearchResult | ErrorResult:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{AURClient.BASE_URL}/rpc/v5/search",
+                params={"v": "5", "arg": term, "by": search_by},
+            )
+            if response.status_code == 200:
+                response_data = response.json()
+                logger.debug(f"Received search response: {response_data}")
+                if response_data.get("type") == "error":
+                    return ErrorResult(**response_data)
+                return SearchResult(**response_data)
+            return ErrorResult(type="error", error="Failed to connect", resultcount=0, version=5)
+
+    @staticmethod
+    async def get_package_info(pkg_name: str) -> InfoResult | ErrorResult:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{AURClient.BASE_URL}/rpc/v5/info", params={"v": "5", "arg[]": pkg_name})
+            if response.status_code == 200:
+                response_data = response.json()
+                logger.debug(f"Received package info response: {response_data}")
+                try:
+                    if response_data.get("type") == "error":
+                        return ErrorResult(**response_data)
+                    return InfoResult(**response_data)
+                except Exception as e:
+                    logger.error(f"Error parsing InfoResult: {e} - Response: {response_data}")
+                    raise
+            return ErrorResult(type="error", error="Failed to connect", resultcount=0, version=5)
+
+    @staticmethod
+    async def get_packages_info(pkg_names: list[str]) -> InfoResult | ErrorResult:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{AURClient.BASE_URL}/rpc/v5/info", params={"v": "5", "arg[]": pkg_names})
+            if response.status_code == 200:
+                response_data = response.json()
+                logger.debug(f"Received packages info response: {response_data}")
+                try:
+                    if response_data.get("type") == "error":
+                        return ErrorResult(**response_data)
+                    return InfoResult(**response_data)
+                except Exception as e:
+                    logger.error(f"Error parsing InfoResult: {e} - Response: {response_data}")
+                    raise
+            return ErrorResult(type="error", error="Failed to connect", resultcount=0, version=5)
+
+    @staticmethod
+    async def suggest_packages(term: str) -> PackageNames:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{AURClient.BASE_URL}/rpc/v5/suggest", params={"v": "5", "arg": term})
+            response_json = response.json()
+            logger.debug(f"Received suggest packages response: {response_json}")
+            return PackageNames.from_list(response_json)
+
+    @staticmethod
+    async def suggest_package_bases(term: str) -> PackageNames:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{AURClient.BASE_URL}/rpc/v5/suggest-pkgbase", params={"v": "5", "arg": term})
+            response_json = response.json()
+            logger.debug(f"Received suggest package bases response: {response_json}")
+            return PackageNames.from_list(response_json)