506 lines
22 KiB
Python
506 lines
22 KiB
Python
import discord
|
|
import aiohttp
|
|
from datetime import datetime, timedelta
|
|
import asyncio
|
|
import time
|
|
from typing import Optional
|
|
from redbot.core import commands, Config, checks
|
|
from redbot.core.utils.chat_formatting import humanize_number, box
|
|
|
|
BASE_URL = "https://api.modrinth.com/v2"
|
|
RATE_LIMIT_REQUESTS = 300 # Maximum requests per minute as per Modrinth's guidelines
|
|
RATE_LIMIT_PERIOD = 60 # Period in seconds (1 minute)
|
|
|
|
class ModrinthTracker(commands.Cog):
|
|
"""Track Modrinth project updates."""
|
|
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self.config = Config.get_conf(self, identifier=1234567890, force_registration=True)
|
|
self.config.register_guild(tracked_projects={})
|
|
self.session = None
|
|
self.bg_task = None
|
|
self.request_timestamps = []
|
|
self.request_lock = asyncio.Lock()
|
|
|
|
async def cog_load(self):
|
|
"""Initialize the cog."""
|
|
self.session = aiohttp.ClientSession()
|
|
self.bg_task = self.bot.loop.create_task(self.update_checker())
|
|
|
|
async def cog_unload(self):
|
|
"""Clean up when cog shuts down."""
|
|
if self.bg_task:
|
|
self.bg_task.cancel()
|
|
if self.session:
|
|
await self.session.close()
|
|
self.session = None
|
|
|
|
async def _rate_limit(self):
|
|
"""Implements rate limiting for API requests"""
|
|
async with self.request_lock:
|
|
current_time = time.time()
|
|
|
|
# Remove timestamps older than our period
|
|
self.request_timestamps = [ts for ts in self.request_timestamps
|
|
if current_time - ts < RATE_LIMIT_PERIOD]
|
|
|
|
if len(self.request_timestamps) >= RATE_LIMIT_REQUESTS:
|
|
# Calculate sleep time needed
|
|
sleep_time = self.request_timestamps[0] + RATE_LIMIT_PERIOD - current_time
|
|
if sleep_time > 0:
|
|
await asyncio.sleep(sleep_time)
|
|
# After sleep, clean up old timestamps again
|
|
current_time = time.time()
|
|
self.request_timestamps = [ts for ts in self.request_timestamps
|
|
if current_time - ts < RATE_LIMIT_PERIOD]
|
|
|
|
self.request_timestamps.append(current_time)
|
|
|
|
async def _make_request(self, url, params=None):
|
|
"""Make a rate-limited request to the Modrinth API"""
|
|
try:
|
|
# Ensure session is available
|
|
if self.session is None or self.session.closed:
|
|
self.session = aiohttp.ClientSession()
|
|
|
|
async with self.session.get(url, params=params) as response:
|
|
if response.status == 429: # Too Many Requests
|
|
retry_after = int(response.headers.get('Retry-After', 60))
|
|
await asyncio.sleep(retry_after)
|
|
return await self._make_request(url, params)
|
|
|
|
# Store response data before closing
|
|
if response.status == 200:
|
|
return await response.json()
|
|
return None
|
|
|
|
except (aiohttp.ClientConnectorError, aiohttp.ClientError) as e:
|
|
# Close and recreate session
|
|
if self.session and not self.session.closed:
|
|
await self.session.close()
|
|
self.session = aiohttp.ClientSession()
|
|
# Try one more time
|
|
async with self.session.get(url, params=params) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
return None
|
|
except Exception as e:
|
|
self.bot.logger.error(f"Error in _make_request: {str(e)}", exc_info=True)
|
|
raise
|
|
|
|
@commands.group()
|
|
@checks.admin()
|
|
async def modrinth(self, ctx):
|
|
"""Commands for tracking Modrinth projects"""
|
|
pass
|
|
|
|
@modrinth.command()
|
|
async def add(self, ctx, project_id: str, channel: discord.TextChannel):
|
|
"""Add a Modrinth project to track
|
|
|
|
Arguments:
|
|
project_id: The Modrinth project ID or slug
|
|
channel: The channel to send updates to
|
|
"""
|
|
try:
|
|
# Verify the project exists and get its info
|
|
project_data = await self._make_request(f"{BASE_URL}/project/{project_id}")
|
|
if not project_data:
|
|
await ctx.send(f"Error: Project `{project_id}` not found on Modrinth.")
|
|
return
|
|
|
|
# Get the latest version
|
|
versions = await self._make_request(f"{BASE_URL}/project/{project_id}/version")
|
|
if not versions:
|
|
await ctx.send("Error: Could not fetch version information.")
|
|
return
|
|
latest_version = versions[0] if versions else None
|
|
|
|
tracked_projects = await self.config.guild(ctx.guild).tracked_projects()
|
|
if project_id in tracked_projects:
|
|
await ctx.send("This project is already being tracked.")
|
|
return
|
|
|
|
tracked_projects[project_id] = {
|
|
"channel": channel.id,
|
|
"latest_version": latest_version["id"] if latest_version else None,
|
|
"name": project_data["title"]
|
|
}
|
|
await self.config.guild(ctx.guild).tracked_projects.set(tracked_projects)
|
|
await ctx.send(f"Now tracking {project_data['title']} (`{project_id}`) in {channel.mention}.")
|
|
|
|
# Post the current version information
|
|
if latest_version:
|
|
embed = discord.Embed(
|
|
title=f"Current Version of {project_data['title']}",
|
|
description=f"Version: `{latest_version.get('version_number', 'Unknown')}`\n\n{latest_version.get('changelog', 'No changelog provided')}",
|
|
url=f"https://modrinth.com/project/{project_id}",
|
|
color=discord.Color.blue(),
|
|
timestamp=datetime.now()
|
|
)
|
|
# Add project icon as thumbnail if available
|
|
if project_data.get("icon_url"):
|
|
embed.set_thumbnail(url=project_data["icon_url"])
|
|
# Add featured gallery image if available
|
|
if project_data.get("gallery"):
|
|
for image in project_data["gallery"]:
|
|
if image.get("featured", False):
|
|
embed.set_image(url=image["url"])
|
|
break
|
|
|
|
# Add project details
|
|
categories = ", ".join(f"`{cat}`" for cat in project_data.get("categories", []))
|
|
if categories:
|
|
embed.add_field(name="Categories", value=categories, inline=True)
|
|
|
|
downloads = project_data.get("downloads", 0)
|
|
followers = project_data.get("followers", 0)
|
|
stats = f"📥 {downloads:,} Downloads\n👥 {followers:,} Followers"
|
|
embed.add_field(name="Statistics", value=stats, inline=True)
|
|
|
|
# Add version details
|
|
loaders = ", ".join(f"`{loader}`" for loader in latest_version.get("loaders", []))
|
|
if loaders:
|
|
embed.add_field(name="Supported Loaders", value=loaders, inline=True)
|
|
|
|
game_versions = ", ".join(f"`{ver}`" for ver in latest_version.get("game_versions", []))
|
|
if game_versions:
|
|
embed.add_field(name="Game Versions", value=game_versions, inline=True)
|
|
|
|
embed.set_footer(text="Tracking Started")
|
|
await channel.send(embed=embed)
|
|
else:
|
|
await channel.send("No version information is currently available for this project.")
|
|
|
|
except aiohttp.ClientConnectorError:
|
|
await ctx.send("Error: Failed to connect to Modrinth API. Please try again in a few moments.")
|
|
except aiohttp.ClientError as e:
|
|
await ctx.send("Error: Network error occurred while contacting Modrinth API. Please try again in a few moments.")
|
|
except asyncio.TimeoutError:
|
|
await ctx.send("Error: Request to Modrinth API timed out. Please try again.")
|
|
except Exception as e:
|
|
self.bot.logger.error(f"Error in modrinth add command: {str(e)}", exc_info=True)
|
|
await ctx.send("An unexpected error occurred. Please try again later or contact the bot owner if the issue persists.")
|
|
|
|
@modrinth.command()
|
|
async def remove(self, ctx, project_id: str):
|
|
"""Remove a tracked Modrinth project
|
|
|
|
Arguments:
|
|
project_id: The Modrinth project ID or slug to stop tracking
|
|
"""
|
|
tracked_projects = await self.config.guild(ctx.guild).tracked_projects()
|
|
if project_id not in tracked_projects:
|
|
await ctx.send("This project is not being tracked.")
|
|
return
|
|
|
|
project_name = tracked_projects[project_id].get("name", project_id)
|
|
del tracked_projects[project_id]
|
|
await self.config.guild(ctx.guild).tracked_projects.set(tracked_projects)
|
|
await ctx.send(f"Stopped tracking {project_name} (`{project_id}`).")
|
|
|
|
@modrinth.command()
|
|
async def list(self, ctx):
|
|
"""List all tracked Modrinth projects"""
|
|
tracked_projects = await self.config.guild(ctx.guild).tracked_projects()
|
|
if not tracked_projects:
|
|
await ctx.send("No projects are currently being tracked.")
|
|
return
|
|
|
|
embed = discord.Embed(
|
|
title="📋 Tracked Modrinth Projects",
|
|
color=discord.Color.blue(),
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
for project_id, data in tracked_projects.items():
|
|
channel = self.bot.get_channel(data["channel"])
|
|
channel_mention = channel.mention if channel else "Unknown channel"
|
|
|
|
# Get current project info
|
|
try:
|
|
async with self.session.get(f"{BASE_URL}/project/{project_id}") as response:
|
|
if response.status == 200:
|
|
project_data = await response.json()
|
|
downloads = project_data.get("downloads", 0)
|
|
followers = project_data.get("followers", 0)
|
|
description = f"**ID:** `{project_id}`\n**Channel:** {channel_mention}\n📥 {downloads:,} Downloads\n👥 {followers:,} Followers"
|
|
embed.add_field(
|
|
name=data.get("name", project_id),
|
|
value=description,
|
|
inline=False
|
|
)
|
|
else:
|
|
embed.add_field(
|
|
name=data.get("name", project_id),
|
|
value=f"**ID:** `{project_id}`\n**Channel:** {channel_mention}",
|
|
inline=False
|
|
)
|
|
except Exception:
|
|
embed.add_field(
|
|
name=data.get("name", project_id),
|
|
value=f"**ID:** `{project_id}`\n**Channel:** {channel_mention}",
|
|
inline=False
|
|
)
|
|
|
|
embed.set_footer(text=f"Total Projects: {len(tracked_projects)}")
|
|
await ctx.send(embed=embed)
|
|
|
|
@modrinth.command()
|
|
async def search(self, ctx, *, query: str):
|
|
"""Search for Modrinth projects to track.
|
|
|
|
This will return a list of projects matching your search query.
|
|
You can then use the project ID with the add command.
|
|
"""
|
|
try:
|
|
params = {
|
|
"query": query,
|
|
"limit": 5,
|
|
"index": "relevance"
|
|
}
|
|
data = await self._make_request(f"{BASE_URL}/search", params=params)
|
|
if not data:
|
|
await ctx.send("Failed to search Modrinth projects.")
|
|
return
|
|
|
|
if not data["hits"]:
|
|
await ctx.send("No projects found matching your query.")
|
|
return
|
|
|
|
embed = discord.Embed(
|
|
title="🔍 Modrinth Project Search Results",
|
|
color=discord.Color.blue(),
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
for project in data["hits"]:
|
|
description = f"**ID:** `{project['project_id']}`\n"
|
|
description += f"**Downloads:** {humanize_number(project.get('downloads', 0))}\n"
|
|
description += f"**Categories:** {', '.join(f'`{cat}`' for cat in project.get('categories', []))}\n"
|
|
description += f"[View on Modrinth](https://modrinth.com/project/{project['project_id']})"
|
|
|
|
embed.add_field(
|
|
name=f"{project['title']}",
|
|
value=description,
|
|
inline=False
|
|
)
|
|
|
|
embed.set_footer(text=f"Found {len(data['hits'])} results • Use [p]modrinth add <project_id> <channel> to track")
|
|
await ctx.send(embed=embed)
|
|
|
|
except Exception as e:
|
|
self.bot.logger.error(f"Error in modrinth search command: {str(e)}", exc_info=True)
|
|
await ctx.send("An unexpected error occurred while searching. Please try again later.")
|
|
|
|
@modrinth.command()
|
|
async def stats(self, ctx, project_id: str):
|
|
"""Show detailed statistics for a tracked project."""
|
|
try:
|
|
# Get project info
|
|
project_data = await self._make_request(f"{BASE_URL}/project/{project_id}")
|
|
if not project_data:
|
|
await ctx.send(f"Error: Project `{project_id}` not found on Modrinth.")
|
|
return
|
|
|
|
# Get version history
|
|
versions = await self._make_request(f"{BASE_URL}/project/{project_id}/version")
|
|
if not versions:
|
|
await ctx.send("Error: Could not fetch version information.")
|
|
return
|
|
|
|
embed = discord.Embed(
|
|
title=f"📊 {project_data['title']} Statistics",
|
|
url=f"https://modrinth.com/project/{project_id}",
|
|
color=discord.Color.blue(),
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
if project_data.get("icon_url"):
|
|
embed.set_thumbnail(url=project_data["icon_url"])
|
|
|
|
# Project Stats
|
|
stats = [
|
|
f"📥 **Downloads:** {humanize_number(project_data.get('downloads', 0))}",
|
|
f"👥 **Followers:** {humanize_number(project_data.get('followers', 0))}",
|
|
f"⭐ **Rating:** {project_data.get('rating', 0):.1f}/5.0"
|
|
]
|
|
embed.add_field(name="Statistics", value="\n".join(stats), inline=False)
|
|
|
|
# Categories and Tags
|
|
categories = ", ".join(f"`{cat}`" for cat in project_data.get("categories", []))
|
|
if categories:
|
|
embed.add_field(name="Categories", value=categories, inline=True)
|
|
|
|
# Version Info
|
|
if versions:
|
|
latest = versions[0]
|
|
version_info = [
|
|
f"**Latest:** `{latest.get('version_number', 'Unknown')}`",
|
|
f"**Released:** <t:{int(datetime.fromisoformat(latest.get('date_published', '')).timestamp())}:R>",
|
|
f"**Total Versions:** {len(versions)}"
|
|
]
|
|
embed.add_field(name="Version Information", value="\n".join(version_info), inline=True)
|
|
|
|
# Project Description
|
|
if project_data.get("description"):
|
|
desc = project_data["description"]
|
|
if len(desc) > 1024:
|
|
desc = desc[:1021] + "..."
|
|
embed.add_field(name="Description", value=desc, inline=False)
|
|
|
|
await ctx.send(embed=embed)
|
|
|
|
except Exception as e:
|
|
self.bot.logger.error(f"Error in modrinth stats command: {str(e)}", exc_info=True)
|
|
await ctx.send(f"An error occurred while fetching statistics: {str(e)}")
|
|
|
|
@modrinth.command()
|
|
async def versions(self, ctx, project_id: str, limit: Optional[int] = 5):
|
|
"""Show version history for a project.
|
|
|
|
Arguments:
|
|
project_id: The Modrinth project ID or slug
|
|
limit: Number of versions to show (default: 5, max: 10)
|
|
"""
|
|
limit = min(max(1, limit), 10) # Clamp between 1 and 10
|
|
|
|
try:
|
|
# Get project info
|
|
project_data = await self._make_request(f"{BASE_URL}/project/{project_id}")
|
|
if not project_data:
|
|
await ctx.send(f"Error: Project `{project_id}` not found on Modrinth.")
|
|
return
|
|
|
|
# Get version history
|
|
versions = await self._make_request(f"{BASE_URL}/project/{project_id}/version")
|
|
if not versions:
|
|
await ctx.send("Error: Could not fetch version information.")
|
|
return
|
|
|
|
if not versions:
|
|
await ctx.send("No version information available for this project.")
|
|
return
|
|
|
|
embed = discord.Embed(
|
|
title=f"📜 Version History for {project_data['title']}",
|
|
url=f"https://modrinth.com/project/{project_id}",
|
|
color=discord.Color.blue(),
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
if project_data.get("icon_url"):
|
|
embed.set_thumbnail(url=project_data["icon_url"])
|
|
|
|
for version in versions[:limit]:
|
|
version_name = version.get("version_number", "Unknown Version")
|
|
|
|
# Format version info
|
|
info = []
|
|
if version.get("date_published"):
|
|
timestamp = int(datetime.fromisoformat(version["date_published"]).timestamp())
|
|
info.append(f"Released: <t:{timestamp}:R>")
|
|
|
|
if version.get("downloads"):
|
|
info.append(f"Downloads: {humanize_number(version['downloads'])}")
|
|
|
|
if version.get("game_versions"):
|
|
info.append(f"Game Versions: {', '.join(f'`{v}`' for v in version['game_versions'])}")
|
|
|
|
if version.get("loaders"):
|
|
info.append(f"Loaders: {', '.join(f'`{l}`' for l in version['loaders'])}")
|
|
|
|
changelog = version.get("changelog", "No changelog provided.")
|
|
if len(changelog) > 200:
|
|
changelog = changelog[:197] + "..."
|
|
|
|
content = "\n".join(info) + f"\n\n{changelog}"
|
|
|
|
embed.add_field(
|
|
name=f"📦 {version_name}",
|
|
value=content,
|
|
inline=False
|
|
)
|
|
|
|
embed.set_footer(text=f"Showing {min(limit, len(versions))} of {len(versions)} versions")
|
|
await ctx.send(embed=embed)
|
|
|
|
except Exception as e:
|
|
self.bot.logger.error(f"Error in modrinth versions command: {str(e)}", exc_info=True)
|
|
await ctx.send("An unexpected error occurred while fetching version history. Please try again later.")
|
|
|
|
async def update_checker(self):
|
|
await self.bot.wait_until_ready()
|
|
while True:
|
|
try:
|
|
all_guilds = await self.config.all_guilds()
|
|
for guild_id, guild_data in all_guilds.items():
|
|
guild = self.bot.get_guild(guild_id)
|
|
if not guild:
|
|
continue
|
|
|
|
tracked_projects = guild_data.get("tracked_projects", {})
|
|
for project_id, data in tracked_projects.items():
|
|
try:
|
|
# Get project info for the embed
|
|
response = await self._make_request(f"{BASE_URL}/project/{project_id}")
|
|
if response.status != 200:
|
|
continue
|
|
project_data = await response.json()
|
|
|
|
response = await self._make_request(f"{BASE_URL}/project/{project_id}/version")
|
|
if response.status != 200:
|
|
continue
|
|
|
|
versions = await response.json()
|
|
if not versions:
|
|
continue
|
|
|
|
latest_version = versions[0]
|
|
if latest_version["id"] == data.get("latest_version"):
|
|
continue
|
|
|
|
channel = self.bot.get_channel(data["channel"])
|
|
if channel:
|
|
embed = discord.Embed(
|
|
title=f"🆕 New Update for {data.get('name', project_id)}!",
|
|
description=f"**Version:** `{latest_version.get('version_number', 'Unknown')}`\n\n{latest_version.get('changelog', 'No changelog provided')}",
|
|
url=f"https://modrinth.com/project/{project_id}",
|
|
color=discord.Color.green(),
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
# Add project icon as thumbnail
|
|
if project_data.get("icon_url"):
|
|
embed.set_thumbnail(url=project_data["icon_url"])
|
|
|
|
# Add version details
|
|
loaders = ", ".join(f"`{loader}`" for loader in latest_version.get("loaders", []))
|
|
if loaders:
|
|
embed.add_field(name="Supported Loaders", value=loaders, inline=True)
|
|
|
|
game_versions = ", ".join(f"`{ver}`" for ver in latest_version.get("game_versions", []))
|
|
if game_versions:
|
|
embed.add_field(name="Game Versions", value=game_versions, inline=True)
|
|
|
|
# Add download info
|
|
downloads = project_data.get("downloads", 0)
|
|
embed.add_field(name="Total Downloads", value=f"📥 {downloads:,}", inline=True)
|
|
|
|
embed.set_footer(text="Update Released")
|
|
await channel.send(embed=embed)
|
|
|
|
tracked_projects[project_id]["latest_version"] = latest_version["id"]
|
|
await self.config.guild(guild).tracked_projects.set(tracked_projects)
|
|
|
|
except Exception as e:
|
|
continue
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
await asyncio.sleep(300) # Check every 5 minutes
|
|
|
|
async def setup(bot):
|
|
await bot.add_cog(ModrinthTracker(bot))
|