utils.py

import io
import json
from collections import defaultdict
from datetime import datetime, timedelta

import aiohttp
import nextcord


def load_json(file_path: str) -> dict:
    with open(file_path, "r", encoding="utf-8") as file:
        return json.load(file)


def save_json(data: dict, file_path: str) -> None:
    with open(file_path, "w", encoding="utf-8") as file:
        json.dump(data, file, indent=4)


def format_datetime(date_str: str) -> str:
    if date_str.endswith("Z"):
        parsed_date = datetime.fromisoformat(date_str[:-1]).timestamp()
    else:
        parsed_date = datetime.fromisoformat(date_str[:-6]).timestamp()
    return f"<t:{round(parsed_date)}>"


async def send_response(interaction: nextcord.Interaction, content: str, title: str, file_name: str) -> None:
    file_extension = file_name.split(".")[-1]
    if len(content) <= 2000:
        await interaction.send(f"```{file_extension}\n{content}```", allowed_mentions=nextcord.AllowedMentions(everyone=False, users=False, roles=False, replied_user=False))
    elif len(content) <= 4000:
        embed = nextcord.Embed(title=title[:256], description=f"```{file_extension}\n{content}```", color=nextcord.Color.blurple())
        await interaction.send(embed=embed)
    else:
        file = nextcord.File(filename=file_name, fp=io.BytesIO(content.encode()))
        await interaction.send(file=file)


async def send_webhook(webhook_url: str, content: str, extension: str) -> None:
    content = f"```{extension}\n{content}```"
    async with aiohttp.ClientSession() as session:
        if len(content) < 2000:
            payload = {"content": content, "username": "Logs"}
            await session.post(webhook_url, json=payload)

        elif len(content) < 4096:
            payload = {
                "embeds": [{
                    "description": content,
                    "color": 3447003
                }],
                "username": "Logs"
            }
            await session.post(webhook_url, json=payload)

        else:
            content = content.replace(f"```{extension}\n", "").replace("```", "").strip()
            file = io.StringIO(content)
            payload = aiohttp.FormData()
            payload.add_field('file', file, filename=f'content.{extension}')
            payload.add_field('username', 'Logs')
            await session.post(webhook_url, data=payload)


def get_guild_embed(guild: nextcord.Guild, title: str, color: nextcord.Color) -> nextcord.Embed:
    embed = nextcord.Embed(title=title, color=color)
    embed.add_field(name="Server ID", value=guild.id, inline=False)
    embed.add_field(name="Owner", value=f"ID: {guild.owner_id}", inline=False)
    embed.add_field(name="Member Count", value=guild.member_count, inline=False)
    embed.add_field(name="Created At", value=f"<t:{round(guild.created_at.timestamp())}>", inline=False)

    if guild.icon:
        embed.set_thumbnail(url=guild.icon.url)

    return embed


class LinkButton(nextcord.ui.View):
    def __init__(self, label: str, url: str, emoji: str = "🔗"):
        super().__init__()
        self.add_item(nextcord.ui.Button(label=label, url=url, emoji=emoji))


class LinksButton(nextcord.ui.View):
    def __init__(self, links: dict):
        super().__init__()
        for label, url in links.items():
            self.add_item(nextcord.ui.Button(label=label, url=url, emoji="🔗"))


class Options:
    visible = nextcord.SlashOption(description="Whether the response should be visible to everyone", required=False, default=True)
    username = nextcord.SlashOption(description="The GitHub username of the user", required=True, max_length=39)
    repo_owner = nextcord.SlashOption(description="The owner of the repository", required=True, max_length=39)
    repository = nextcord.SlashOption(description="The name of the repository", required=True, max_length=100)
    gist_id = nextcord.SlashOption(description="The ID of the gist", required=True, max_length=32)
    state = nextcord.SlashOption(description="Indicates the state of the issues to return", required=False, choices=["open", "closed", "all"], default="open")
    sort_issues = nextcord.SlashOption(description="The sort order of the issues", required=False, choices=["created", "updated", "comments"], default="created")
    sort_prs = nextcord.SlashOption(description="The sort order of the pull requests", required=False, choices=["created", "updated", "popularity", "long-running"], default="created")
    direction = nextcord.SlashOption(description="The direction to sort the results by", required=False, choices={"asc": "ascending", "desc": "descending"}, default="desc")
    author_filter = nextcord.SlashOption(description="GitHub username or email address to use to filter by author", required=False)
    search_query = nextcord.SlashOption(description="The query to search for", required=True, max_length=256)
    search_repositories_sort = nextcord.SlashOption(description="The sort order of the repositories", required=False, choices=["stars", "forks", "help-wanted-issues", "updated"])
    search_users_sort = nextcord.SlashOption(description="The sort order of the users", required=False, choices=["followers", "repositories", "joined"])
    search_commits_sort = nextcord.SlashOption(description="The sort order of the commits", required=False, choices=["author-date", "committer-date"])


class RateLimiter:
    def __init__(self):
        self.command_limits = {"general": [], "search": []}
        self.command_limits = defaultdict(lambda: defaultdict(list))
        self.button_limits = defaultdict(list)
        self.last_cleanup = datetime.now()

    def cleanup_old_data(self):
        current_time = datetime.now()
        if current_time - self.last_cleanup < timedelta(minutes=5):
            return

        one_hour_ago = current_time - timedelta(hours=1)

        for user_id in list(self.command_limits.keys()):
            self.command_limits[user_id]["general"] = [t for t in self.command_limits[user_id]["general"] if t > one_hour_ago]
            self.command_limits[user_id]["search"] = [t for t in self.command_limits[user_id]["search"] if t > one_hour_ago]
            if not self.command_limits[user_id]["general"] and not self.command_limits[user_id]["search"]:
                del self.command_limits[user_id]

        for user_id in list(self.button_limits.keys()):
            self.button_limits[user_id] = [t for t in self.button_limits[user_id] if t > one_hour_ago]
            if not self.button_limits[user_id]:
                del self.button_limits[user_id]

        self.last_cleanup = current_time

    async def check_command_limit(self, user_id, command_name):
        self.cleanup_old_data()

        if command_name == "search":
            user_commands = self.command_limits[user_id]["search"]
            limit = 20
            global_limit = 25
        else:
            user_commands = self.command_limits[user_id]["general"]
            limit = 100
            global_limit = 150

        if len(user_commands) >= limit:
            return False

        if sum(len(data[command_name]) for data in self.command_limits.values()) >= global_limit:
            return False

        user_commands.append(datetime.now())
        return True

    async def check_button_limit(self, user_id):
        self.cleanup_old_data()
        user_buttons = self.button_limits[user_id]

        if len(user_buttons) >= 500:
            return False

        user_buttons.append(datetime.now())
        return True


class Logger:
    def __init__(self, webhook_url):
        self.logs = []
        self.webhook_url = webhook_url

    def add_log(self, user_id: int, username: str, command: str, options: dict, guild_id: int, channel_id: int):
        formatted_options = ', '.join([f"{k}={v}" for k, v in options.items()])
        formatted_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        log_message = f"[{formatted_date}] {user_id} ({username}) used command {command} with options {formatted_options} on guild {guild_id} in channel {channel_id}"
        self.logs.append(log_message)

    async def send_logs(self):
        if not self.logs:
            return

        await send_webhook(self.webhook_url, "\n".join(self.logs), "log")

        with open("logs.txt", "a") as f:
            f.write("\n".join(self.logs) + "\n")

        self.logs.clear()