diff --git a/src/common/database.py b/src/common/database.py index 0c469af..58dadfe 100644 --- a/src/common/database.py +++ b/src/common/database.py @@ -5,7 +5,6 @@ import shelve import shutil import threading -from typing import Union # lib imports import git @@ -25,7 +24,7 @@ class Database: - def __init__(self, db_name: str, db_dir: Union[str, Path] = data_dir, use_git: bool = True): + def __init__(self, db_name: str, db_dir: str | Path = data_dir, use_git: bool = True): self.db_name = db_name self.db_dir = db_dir self.repo = None diff --git a/src/common/rank.py b/src/common/rank.py index c20b6b1..412dbb2 100644 --- a/src/common/rank.py +++ b/src/common/rank.py @@ -5,7 +5,7 @@ import random import threading import time -from typing import Dict, List, Optional, Tuple, Union +from typing import Dict, List, Optional, Tuple # lib imports import aiohttp @@ -60,7 +60,7 @@ def __init__( self.last_activity = {} # Tracks last activity time for cooldowns @staticmethod - def get_community_id(platform: str, user: Union[DiscordUser, RedditUser]) -> Optional[Union[int, str]]: + def get_community_id(platform: str, user: DiscordUser | RedditUser) -> int | str | None: """ Get the community ID for a user based on platform. @@ -68,12 +68,12 @@ def get_community_id(platform: str, user: Union[DiscordUser, RedditUser]) -> Opt ---------- platform : str Platform identifier ('discord' or 'reddit') - user : Union[DiscordUser, RedditUser] + user : DiscordUser | RedditUser Discord or Reddit user object Returns ------- - Union[int, str] + int | str Community ID (guild_id for Discord, subreddit_id for Reddit) """ if platform == 'discord': @@ -138,7 +138,7 @@ def calculate_xp_for_level(level: int) -> int: def get_rank_data( self, platform: str, - user: Union[DiscordUser, RedditUser], + user: DiscordUser | RedditUser, create_if_not_exists: bool = False, ) -> dict: """ @@ -148,7 +148,7 @@ def get_rank_data( ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - user : Union[DiscordUser, RedditUser] + user : DiscordUser | RedditUser Discord or Reddit user object create_if_not_exists : bool If True, create a new user entry if it doesn't exist @@ -189,7 +189,7 @@ def get_rank_data( return user_data - def update_rank_data(self, platform: str, user: Union[DiscordUser, RedditUser], data: dict) -> dict: + def update_rank_data(self, platform: str, user: DiscordUser | RedditUser, data: dict) -> dict: """ Update rank data for a user. @@ -197,7 +197,7 @@ def update_rank_data(self, platform: str, user: Union[DiscordUser, RedditUser], ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - user : Union[DiscordUser, RedditUser] + user : DiscordUser | RedditUser Discord or Reddit user object data : dict New data to update @@ -219,7 +219,7 @@ def update_rank_data(self, platform: str, user: Union[DiscordUser, RedditUser], data=data, ) - def award_xp(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optional[dict]: + def award_xp(self, platform: str, user: DiscordUser | RedditUser) -> Optional[dict]: """ Award XP to a user with cooldown enforcement. @@ -227,7 +227,7 @@ def award_xp(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optio ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - user : Union[DiscordUser, RedditUser] + user : DiscordUser | RedditUser Discord or Reddit user object Returns @@ -287,8 +287,8 @@ def award_xp(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optio def get_leaderboard( self, platform: str, - user: Optional[Union[DiscordUser, RedditUser]] = None, - community_id: Optional[Union[int, str]] = None, + user: DiscordUser | RedditUser | None = None, + community_id: int | str | None = None, limit: int = 100, offset: int = 0, ) -> List[dict]: @@ -299,9 +299,9 @@ def get_leaderboard( ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - user : Optional[Union[DiscordUser, RedditUser]] + user : DiscordUser | RedditUser | None User object to determine community_id if not explicitly provided - community_id : Optional[Union[int, str]] + community_id : int | str | None Community identifier (guild_id for Discord, subreddit_id for Reddit) limit : int Maximum number of entries to return @@ -338,7 +338,7 @@ def get_leaderboard( return leaderboard - def get_user_rank_position(self, platform: str, user: Union[DiscordUser, RedditUser]) -> Optional[int]: + def get_user_rank_position(self, platform: str, user: DiscordUser | RedditUser) -> Optional[int]: """ Get the exact rank position of a user on the leaderboard without any limits. @@ -346,7 +346,7 @@ def get_user_rank_position(self, platform: str, user: Union[DiscordUser, RedditU ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - user : Union[DiscordUser, RedditUser] + user : DiscordUser | RedditUser Discord or Reddit user object Returns @@ -377,8 +377,8 @@ def get_user_rank_position(self, platform: str, user: Union[DiscordUser, RedditU def get_migration_status( self, platform: str, - community_id: Union[int, str], - source_id: Union[int, str], + community_id: int | str, + source_id: int | str, ) -> Optional[dict]: """ Check if migration has already been performed for a specific source. @@ -387,9 +387,9 @@ def get_migration_status( ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id or subreddit_id) - source_id : Union[int, str] + source_id : int | str Source identifier (e.g., guild_id for Mee6) Returns @@ -406,8 +406,8 @@ def get_migration_status( def set_migration_completed( self, platform: str, - community_id: Union[int, str], - source_id: Union[int, str], + community_id: int | str, + source_id: int | str, stats: dict, ) -> dict: """ @@ -417,9 +417,9 @@ def set_migration_completed( ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id or subreddit_id) - source_id : Union[int, str] + source_id : int | str Source identifier (e.g., guild_id for Mee6) stats : dict Migration statistics @@ -441,7 +441,7 @@ def migrate_from_reddit_database( reddit_bot, reddit_db, community_id: str, - ) -> Dict[str, Union[int, str]]: + ) -> Dict[str, int | str]: """ Migrate user data from Reddit database. @@ -458,7 +458,7 @@ def migrate_from_reddit_database( Returns ------- - Dict[str, Union[int, str]] + Dict[str, int | str] Migration statistics """ # Acquire migration lock to prevent concurrent migrations @@ -600,7 +600,7 @@ def _do_reddit_migration( reddit_bot, reddit_db, community_id: str, - ) -> Dict[str, Union[int, str]]: + ) -> Dict[str, int | str]: """Internal method that performs the actual Reddit migration.""" total_submissions = 0 total_comments = 0 @@ -681,7 +681,7 @@ def _do_reddit_migration( logger.info(f"Reddit migration completed with stats: {stats}") return stats - async def migrate_from_mee6(self, guild_id: int) -> Dict[str, Union[int, str]]: + async def migrate_from_mee6(self, guild_id: int) -> Dict[str, int | str]: """ Migrate user data from Mee6 API. @@ -692,7 +692,7 @@ async def migrate_from_mee6(self, guild_id: int) -> Dict[str, Union[int, str]]: Returns ------- - Dict[str, Union[int, str]] + Dict[str, int | str] Migration statistics """ import asyncio @@ -818,7 +818,7 @@ def _process_mee6_batch( return new_users, updated_users - async def _do_mee6_migration(self, guild_id: int) -> Dict[str, Union[int, str]]: + async def _do_mee6_migration(self, guild_id: int) -> Dict[str, int | str]: """Internal method that performs the actual Mee6 migration.""" page = 0 total_users = 0 diff --git a/src/common/rank_database.py b/src/common/rank_database.py index 8d76da3..2b1b9fc 100644 --- a/src/common/rank_database.py +++ b/src/common/rank_database.py @@ -1,5 +1,5 @@ # standard imports -from typing import List, Optional, Union +from typing import List, Optional # local imports from src.common.database import Database @@ -39,7 +39,7 @@ def _ensure_tables(self): def get_community_users( self, platform: str, - community_id: Union[int, str], + community_id: int | str, search: Optional[str] = None, ) -> List[dict]: """ @@ -49,7 +49,7 @@ def get_community_users( ---------- platform : str Platform identifier ('discord' or 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id for Discord, subreddit_id for Reddit) search : Optional[str] Optional search string to filter users by username (case-insensitive) @@ -84,8 +84,8 @@ def get_community_users( def get_user_data( self, platform: str, - community_id: Union[int, str], - user_id: Union[int, str], + community_id: int | str, + user_id: int | str, create_if_not_exists: bool = False, ) -> dict: """ @@ -95,9 +95,9 @@ def get_user_data( ---------- platform : str Platform identifier ('discord' or 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id for Discord, subreddit_id for Reddit) - user_id : Union[int, str] + user_id : int | str User identifier create_if_not_exists : bool Whether to create a new user entry if it doesn't exist @@ -140,8 +140,8 @@ def get_user_data( def update_user_data( self, platform: str, - community_id: Union[int, str], - user_id: Union[int, str], + community_id: int | str, + user_id: int | str, data: dict, ) -> dict: """ @@ -151,9 +151,9 @@ def update_user_data( ---------- platform : str Platform identifier ('discord' or 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id or subreddit_id) - user_id : Union[int, str] + user_id : int | str User identifier data : dict New user data @@ -196,7 +196,7 @@ def update_user_data( def get_leaderboard( self, platform: str, - community_id: Union[int, str], + community_id: int | str, limit: int = 100, offset: int = 0, ) -> List[dict]: @@ -207,7 +207,7 @@ def get_leaderboard( ---------- platform : str Platform identifier ('discord' or 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id or subreddit_id) limit : int Maximum number of entries to return @@ -251,8 +251,8 @@ def get_leaderboard( def get_migration_status( self, platform: str, - community_id: Union[int, str], - source_id: Union[int, str], + community_id: int | str, + source_id: int | str, ) -> Optional[dict]: """ Check if migration has already been performed for a specific source. @@ -261,9 +261,9 @@ def get_migration_status( ---------- platform : str Platform identifier (e.g., 'discord', 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id or subreddit_id) - source_id : Union[int, str] + source_id : int | str Source identifier (e.g., guild_id for Mee6) Returns @@ -285,8 +285,8 @@ def get_migration_status( def set_migration_completed( self, platform: str, - community_id: Union[int, str], - source_id: Union[int, str], + community_id: int | str, + source_id: int | str, stats: dict, ) -> dict: """ @@ -296,9 +296,9 @@ def set_migration_completed( ---------- platform : str Platform identifier ('discord' or 'reddit') - community_id : Union[int, str] + community_id : int | str Community identifier (guild_id or subreddit_id) - source_id : Union[int, str] + source_id : int | str Source identifier (e.g., guild_id for Mee6) stats : dict Migration statistics diff --git a/src/common/sponsors.py b/src/common/sponsors.py index 793c5d8..37b1033 100644 --- a/src/common/sponsors.py +++ b/src/common/sponsors.py @@ -1,7 +1,7 @@ # standard imports import logging import os -from typing import Union +from typing import Literal # lib imports import requests @@ -18,13 +18,13 @@ } -def get_github_sponsors() -> Union[dict, False]: +def get_github_sponsors() -> dict | Literal[False]: """ Get list of GitHub sponsors. Returns ------- - Union[dict, False] + dict | Literal[False] JSON response containing the list of sponsors. False if an error occurred. """ token = os.getenv("GITHUB_TOKEN") diff --git a/src/discord_bot/cogs/fun_commands.py b/src/discord_bot/cogs/fun_commands.py index 7c88759..4051f7e 100644 --- a/src/discord_bot/cogs/fun_commands.py +++ b/src/discord_bot/cogs/fun_commands.py @@ -16,10 +16,12 @@ def __init__(self, bot): self.bot = bot async def get_random_quotes(self) -> list[dict]: - async with aiohttp.ClientSession() as session: - async with session.get(url='https://app.lizardbyte.dev/uno/random-quotes/games.json') as response: - response.raise_for_status() - return await response.json() + async with ( + aiohttp.ClientSession() as session, + session.get(url='https://app.lizardbyte.dev/uno/random-quotes/games.json') as response, + ): + response.raise_for_status() + return await response.json() @discord.slash_command( name="random", diff --git a/src/discord_bot/cogs/moderator_commands.py b/src/discord_bot/cogs/moderator_commands.py index 832d879..0a787b3 100644 --- a/src/discord_bot/cogs/moderator_commands.py +++ b/src/discord_bot/cogs/moderator_commands.py @@ -1,6 +1,5 @@ # standard imports import time -from typing import Union # lib imports import discord @@ -32,7 +31,7 @@ async def channel_command( self, ctx: discord.ApplicationContext, recommended_channel: Option( - Union[discord.ForumChannel, discord.TextChannel], + discord.ForumChannel | discord.TextChannel, description=recommended_channel_desc, required=True, ), @@ -45,7 +44,7 @@ async def channel_command( ---------- ctx : discord.ApplicationContext Request message context. - recommended_channel : Union[discord.ForumChannel, discord.TextChannel] + recommended_channel : discord.ForumChannel | discord.TextChannel The recommended channel to move discussion to. """ categories_map = { diff --git a/src/discord_bot/cogs/rank.py b/src/discord_bot/cogs/rank.py index 8a3a977..d233c28 100644 --- a/src/discord_bot/cogs/rank.py +++ b/src/discord_bot/cogs/rank.py @@ -1,7 +1,6 @@ # standard imports import logging import os -from typing import Union # lib imports import discord @@ -397,7 +396,7 @@ def create_leaderboard_buttons(self, page, total_pages, platform): async def get_leaderboard_data( self, platform: str, - community_id: Union[int, str], + community_id: int | str, page: int = 1, per_page: int = 10, ): diff --git a/src/discord_bot/cogs/support_commands.py b/src/discord_bot/cogs/support_commands.py index 95e1188..8782902 100644 --- a/src/discord_bot/cogs/support_commands.py +++ b/src/discord_bot/cogs/support_commands.py @@ -139,11 +139,14 @@ async def project_command(ctx: discord.ApplicationContext, command: str): @staticmethod def render_command_file(command_file: str) -> str: - with open(command_file, "r", encoding='utf-8') as file: - with MarkdownRenderer( - max_line_length=4096, # this must be set to reflow the text - normalize_whitespace=True) as renderer: - return renderer.render(mistletoe.Document(file)) + with ( + open(command_file, "r", encoding='utf-8') as file, + MarkdownRenderer( + max_line_length=4096, # this must be set to reflow the text + normalize_whitespace=True, + ) as renderer, + ): + return renderer.render(mistletoe.Document(file)) @discord.slash_command( name="docs",