mirror of
https://github.com/allthingslinux/tux.git
synced 2024-10-02 16:43:12 +00:00
feat(pyproject.toml): add repology-client dependency to support new feature
feat(tux/cogs/utility/repology.py): add new Repology cog to handle repology commands feat(tux/wrappers/repology.py): add new RepologyWrapper to interact with Repology API
This commit is contained in:
parent
43ddda891d
commit
c37386a7c3
4 changed files with 190 additions and 1 deletions
22
poetry.lock
generated
22
poetry.lock
generated
|
@ -1971,6 +1971,26 @@ files = [
|
|||
{file = "regex-2024.7.24.tar.gz", hash = "sha256:9cfd009eed1a46b27c14039ad5bbc5e71b6367c5b2e6d5f5da0ea91600817506"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "repology-client"
|
||||
version = "0.2.0"
|
||||
description = "Asynchronous wrapper for Repology API"
|
||||
optional = false
|
||||
python-versions = ">=3.11"
|
||||
files = [
|
||||
{file = "repology_client-0.2.0-py3-none-any.whl", hash = "sha256:e87917554ff32d015a2cfbe34f0ba13c9e293e3008b87f93b6243b2fb73e1e0d"},
|
||||
{file = "repology_client-0.2.0.tar.gz", hash = "sha256:495fe3910f7c7f5ca6b5d56b0ff227e369bc822ad67b7307777840b3d27bdfc0"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
aiohttp = ">=3,<4"
|
||||
pydantic = ">=2,<3"
|
||||
pydantic-core = ">=2,<3"
|
||||
|
||||
[package.extras]
|
||||
docs = ["alabaster", "sphinx", "sphinx-prompt"]
|
||||
test = ["pytest", "pytest-asyncio", "pytest-recording"]
|
||||
|
||||
[[package]]
|
||||
name = "requests"
|
||||
version = "2.32.3"
|
||||
|
@ -2422,4 +2442,4 @@ multidict = ">=4.0"
|
|||
[metadata]
|
||||
lock-version = "2.0"
|
||||
python-versions = ">=3.12,<4"
|
||||
content-hash = "e773f58b7548a102f8e9ebc152ec2390d1149a549fe9f77e9c248b37e896f56a"
|
||||
content-hash = "23f2f7a3fc6b0aeac492e32171fa6102ff69f098a20848a05755fc50f4e43ff7"
|
||||
|
|
|
@ -37,6 +37,7 @@ types-psutil = "^6.0.0.20240621"
|
|||
typing-extensions = "^4.12.2"
|
||||
jishaku = "^2.5.2"
|
||||
pytz = "^2024.1"
|
||||
repology-client = "^0.2.0"
|
||||
|
||||
[tool.poetry.group.docs.dependencies]
|
||||
mkdocs-material = "^9.5.30"
|
||||
|
|
70
tux/cogs/utility/repology.py
Normal file
70
tux/cogs/utility/repology.py
Normal file
|
@ -0,0 +1,70 @@
|
|||
import discord
|
||||
from discord.ext import commands
|
||||
from loguru import logger
|
||||
from reactionmenu import ViewButton, ViewMenu
|
||||
from repology_client.types import Package
|
||||
|
||||
from tux.wrappers.repology import RepologyWrapper
|
||||
|
||||
|
||||
class Repology(commands.Cog):
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
self.repology = RepologyWrapper()
|
||||
|
||||
@commands.command(name="repology")
|
||||
async def repology_command(self, ctx: commands.Context[commands.Bot], project: str) -> None:
|
||||
try:
|
||||
packages: set[Package] = await self.repology.get_packages(project)
|
||||
logger.info(f"Repology packages for {project}: {packages}")
|
||||
except Exception as e:
|
||||
await ctx.send(f"An error occurred: {e}")
|
||||
return
|
||||
|
||||
if not packages:
|
||||
await ctx.send("No packages found.")
|
||||
return
|
||||
|
||||
pages: list[discord.Embed] = []
|
||||
|
||||
embed = discord.Embed(title=f"Repology packages for {project}")
|
||||
|
||||
for count, package in enumerate(packages, start=1):
|
||||
logger.info(f"Package: {package}")
|
||||
value = package.visiblename
|
||||
if package.version:
|
||||
value += f" ({package.version})"
|
||||
if package.status:
|
||||
value += f" - {package.status}"
|
||||
if package.summary:
|
||||
value += f"\n> {package.summary}"
|
||||
|
||||
embed.add_field(name=package.repo, value=value, inline=False)
|
||||
|
||||
if count % 10 == 0:
|
||||
pages.append(embed)
|
||||
embed = discord.Embed(title=f"Repology packages for {project}")
|
||||
|
||||
if embed.fields:
|
||||
pages.append(embed)
|
||||
|
||||
await self._send_paginated_response(ctx, pages)
|
||||
|
||||
async def _send_paginated_response(self, ctx: commands.Context[commands.Bot], pages: list[discord.Embed]) -> None:
|
||||
if pages:
|
||||
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()
|
||||
|
||||
|
||||
async def setup(bot: commands.Bot) -> None:
|
||||
await bot.add_cog(Repology(bot))
|
98
tux/wrappers/repology.py
Normal file
98
tux/wrappers/repology.py
Normal file
|
@ -0,0 +1,98 @@
|
|||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from repology_client import exceptions as repology_exceptions
|
||||
from repology_client.types import Package, ResolvePackageType
|
||||
|
||||
|
||||
class RepologyWrapper:
|
||||
BASE_API_V1_URL = "https://repology.org/api/v1"
|
||||
TOOL_PROJECT_BY_URL = "https://repology.org/tools/project-by"
|
||||
API_EXP_URL = "https://repology.org/api/experimental"
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
async def get_packages(self, project: str) -> set[Package]:
|
||||
async with httpx.AsyncClient() as client:
|
||||
try:
|
||||
url = f"{self.BASE_API_V1_URL}/project/{project}"
|
||||
response: httpx.Response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
json_data = response.json()
|
||||
if not json_data:
|
||||
self._raise_empty_response_error("Empty response from get_packages")
|
||||
return {Package(**pkg) for pkg in json_data}
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
except Exception as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
|
||||
async def get_projects(
|
||||
self,
|
||||
start: str = "",
|
||||
end: str = "",
|
||||
count: int = 200,
|
||||
*,
|
||||
filters: dict[str, Any] | None = None,
|
||||
) -> dict[str, set[Package]]:
|
||||
async with httpx.AsyncClient() as client:
|
||||
try:
|
||||
params = {"start": start, "end": end, "count": count}
|
||||
if filters:
|
||||
params |= filters
|
||||
url = f"{self.BASE_API_V1_URL}/projects/"
|
||||
response: httpx.Response = await client.get(url, params=params)
|
||||
response.raise_for_status()
|
||||
json_data = response.json()
|
||||
if not json_data:
|
||||
self._raise_empty_response_error("Empty response from get_projects")
|
||||
return {project: {Package(**pkg) for pkg in pkgs} for project, pkgs in json_data.items()}
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
except Exception as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
|
||||
async def resolve_package(
|
||||
self,
|
||||
repo: str,
|
||||
name: str,
|
||||
name_type: ResolvePackageType = ResolvePackageType.SOURCE,
|
||||
*,
|
||||
autoresolve: bool = True,
|
||||
) -> set[Package]:
|
||||
async with httpx.AsyncClient() as client:
|
||||
try:
|
||||
url = f"{self.TOOL_PROJECT_BY_URL}/{repo}/{name_type.value}/{name}"
|
||||
response: httpx.Response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
json_data = response.json()
|
||||
if not json_data:
|
||||
self._raise_empty_response_error(f"Empty response from resolve_package for {repo}:{name}")
|
||||
if isinstance(json_data, list) and not autoresolve:
|
||||
self._raise_invalid_input_error(f"Multiple packages found for {repo}:{name}")
|
||||
return {Package(**pkg) for pkg in json_data} # type: ignore
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
except Exception as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
|
||||
async def get_distromap(self, fromrepo: str, torepo: str) -> Any:
|
||||
async with httpx.AsyncClient() as client:
|
||||
try:
|
||||
url = f"{self.API_EXP_URL}/distromap/{fromrepo}/{torepo}"
|
||||
response: httpx.Response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
if json_data := response.json():
|
||||
return json_data
|
||||
self._raise_empty_response_error("Empty response from get_distromap")
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
except Exception as e:
|
||||
raise repology_exceptions.InvalidInput(str(e)) from e
|
||||
|
||||
def _raise_empty_response_error(self, message: str):
|
||||
raise repology_exceptions.EmptyResponse(message)
|
||||
|
||||
def _raise_invalid_input_error(self, message: str):
|
||||
raise repology_exceptions.InvalidInput(message)
|
Loading…
Reference in a new issue