diff --git a/bot.py b/bot.py index 6c0f24f..97b0f5f 100644 --- a/bot.py +++ b/bot.py @@ -11,17 +11,18 @@ from services.health_monitor import HealthMonitor from services.metadata_monitor import MetadataMonitor from services.state_manager import StateManager +from services.personal_favorites_manager import PersonalFavoritesManager from pls_parser import parse_pls import shout_errors import urllib_hack from dotenv import load_dotenv from pathlib import Path from streamscrobbler import streamscrobbler -from favorites_manager import get_favorites_manager -from permissions import get_permission_manager, can_set_favorites_check, can_remove_favorites_check, can_manage_roles_check -from stream_validator import get_stream_validator -from input_validator import get_input_validator -from ui_components import FavoritesView, create_favorites_embed, create_favorites_list_embed, create_role_setup_embed, ConfirmationView +# from favorites_manager import get_favorites_manager +# from permissions import get_permission_manager, can_set_favorites_check, can_remove_favorites_check, can_manage_roles_check +# from stream_validator import get_stream_validator +# from input_validator import get_input_validator +# from ui_components import FavoritesView, create_favorites_embed, create_favorites_list_embed, create_role_setup_embed, ConfirmationView load_dotenv() # take environment variables from .env. @@ -89,7 +90,8 @@ _active_heartbeats = {} # TODO: Clean this up? -STATE_MANAGER = None +STATE_MANAGER: StateManager = None +PERSONAL_FAVORITES_MANAGER: PersonalFavoritesManager = None MONITORS = [] async def init(): @@ -97,6 +99,8 @@ async def init(): # Create State Manager to manage the state global STATE_MANAGER STATE_MANAGER = await StateManager.create_state_manager(bot=bot) + global PERSONAL_FAVORITES_MANAGER + PERSONAL_FAVORITES_MANAGER = PersonalFavoritesManager(logger=logger) # Create list of monitors global MONITORS MONITORS = [ @@ -192,7 +196,7 @@ async def predicate(interaction: discord.Interaction): name='play', description="Begin playback of a shoutcast/icecast stream" ) -@discord.app_commands.checks.cooldown(rate=1, per=5, key=None) +@discord.app_commands.checks.cooldown(rate=1, per=5, key=lambda i: i.guild_id) @is_channel() @bot_has_channel_permissions(permissions=discord.Permissions(send_messages=True)) @bot_not_in_maintenance() @@ -211,7 +215,7 @@ async def play(interaction: discord.Interaction, url: str, private_stream: bool name='leave', description="Remove the bot from the current call" ) -@discord.app_commands.checks.cooldown(rate=1, per=5, key=None) +@discord.app_commands.checks.cooldown(rate=1, per=5, key=lambda i: i.guild_id) @is_channel() @bot_not_in_maintenance() async def leave(interaction: discord.Interaction, force: bool = False): @@ -266,7 +270,7 @@ async def song(interaction: discord.Interaction): name="refresh", description="Refresh the stream. Bot will leave and come back. Song updates will start displaying in this channel" ) -@discord.app_commands.checks.cooldown(rate=1, per=5, key=None) +@discord.app_commands.checks.cooldown(rate=1, per=5, key=lambda i: i.guild_id) @is_channel() @bot_has_channel_permissions(permissions=discord.Permissions(send_messages=True)) @bot_not_in_maintenance() @@ -367,7 +371,7 @@ async def debug(interaction: discord.Interaction, page: int = 0, per_page: int = name='maint', description="Toggle maintenance mode! (Bot maintainer only)" ) -@discord.app_commands.checks.cooldown(rate=1, per=5, key=None) +@discord.app_commands.checks.cooldown(rate=1, per=5, key=lambda i: i.guild_id) @is_channel() @bot_has_channel_permissions(permissions=discord.Permissions(send_messages=True)) async def maint(interaction: discord.Interaction, status: bool = True): @@ -423,249 +427,107 @@ async def maint(interaction: discord.Interaction, status: bool = True): await interaction.response.send_message("Awww look at you, how cute") ### FAVORITES COMMANDS ### - @bot.tree.command( - name='set-favorite', - description="Add a radio station to favorites" + name='add-favorite', + description="Add a radio station to my favorites" ) @discord.app_commands.checks.cooldown(rate=1, per=5) +@bot_not_in_maintenance() @is_channel() -async def set_favorite(interaction: discord.Interaction, url: str, name: str = None): - # Check permissions - perm_manager = get_permission_manager() - if not perm_manager.can_set_favorites(interaction.guild.id, interaction.user): - await interaction.response.send_message( - "❌ You don't have permission to set favorites. Ask an admin to assign you the appropriate role.", - ephemeral=True - ) - return - - # Validate URL format first - if not is_valid_url(url): - await interaction.response.send_message("❌ Please provide a valid URL.", ephemeral=True) - return - - await interaction.response.send_message("🔍 Validating stream and adding to favorites...") - - try: - favorites_manager = get_favorites_manager() - result = await favorites_manager.add_favorite( - guild_id=interaction.guild.id, - url=url, - name=name, - user_id=interaction.user.id - ) - - if result['success']: - await interaction.edit_original_response( - content=f"✅ Added **{result['station_name']}** as favorite #{result['favorite_number']}" - ) - else: - await interaction.edit_original_response( - content=f"❌ Failed to add favorite: {result['error']}" - ) - - except Exception as e: - logger.error(f"Error in set_favorite command: {e}") - await interaction.edit_original_response( - content="❌ An unexpected error occurred while adding the favorite." - ) +async def add_favorite(interaction: discord.Interaction, url: str, station_name: str = None): + logger.debug("[%s] Attempting to add %s to %s favorites", interaction.guild_id, url, interaction.user.name) + if await PERSONAL_FAVORITES_MANAGER.create_user_favorite(interaction.user.id, stream_url=url, station_name=station_name): + logger.debug("[%s] Add success", interaction.guild_id) + await interaction.response.send_message(content="Favorite added!", ephemeral=True) + else: + logger.debug("[%s] Add failed", interaction.guild_id) + await interaction.response.send_message(content="Failed to add favorite", ephemeral=True) @bot.tree.command( - name='play-favorite', - description="Play a favorite radio station by number" + name='list-favorites', + description="List my favorited radio stations" ) @discord.app_commands.checks.cooldown(rate=1, per=5) +@bot_not_in_maintenance() @is_channel() -async def play_favorite(interaction: discord.Interaction, number: int): - try: - favorites_manager = get_favorites_manager() - favorite = favorites_manager.get_favorite_by_number(interaction.guild.id, number) - - if not favorite: - await interaction.response.send_message(f"❌ Favorite #{number} not found.", ephemeral=True) - return +async def list_favorites(interaction: discord.Interaction): + favorites = await PERSONAL_FAVORITES_MANAGER.retrieve_user_favorites(user_id=interaction.user.id) + logger.debug('[%s] %s\'s favorites are: %s', interaction.guild_id, interaction.user.name, favorites) + embed_data = { + 'title': "ðŸ“ŧ Your favorites", + 'color': 0x0099ff, + 'description': "", + # 'timestamp': str(datetime.datetime.now(datetime.UTC)), + } - await interaction.response.send_message( - f"ðŸŽĩ Starting favorite #{number}: **{favorite['station_name']}**" - ) - await play_stream(interaction, favorite['stream_url']) + if not favorites: + embed_data['description'] = '`No Favorites`' + for i,(fav, *_) in enumerate(favorites): - except Exception as e: - logger.error(f"Error in play_favorite command: {e}") - if interaction.response.is_done(): - await interaction.followup.send("❌ An error occurred while playing the favorite.", ephemeral=True) + if fav.station_name: + embed_data['description'] += f"`{i+1}`:\t[{fav.station_name}]({fav.stream_url})\n" else: - await interaction.response.send_message("❌ An error occurred while playing the favorite.", ephemeral=True) + embed_data['description'] += f"`{i+1}`:\t{url_slicer(fav.stream_url)}\n" -@bot.tree.command( - name='favorites', - description="Show favorites with clickable buttons" -) -@discord.app_commands.checks.cooldown(rate=1, per=10) -@is_channel() -async def favorites(interaction: discord.Interaction): - try: - favorites_manager = get_favorites_manager() - favorites_list = favorites_manager.get_favorites(interaction.guild.id) - - if not favorites_list: - await interaction.response.send_message( - "ðŸ“ŧ No favorites set for this server yet! Use `/set-favorite` to add some.", - ephemeral=True - ) - return - - # Create embed and view with buttons - embed = create_favorites_embed(favorites_list, 0, interaction.guild.name) - view = FavoritesView(favorites_list, 0) - - await interaction.response.send_message(embed=embed, view=view) + embed = discord.Embed.from_dict(embed_data) + embed.add_field(name="Total:", value=f"`{len(favorites)} Favorites`", inline=True) + embed.add_field(name="Add More:", value="`/add-favorite`", inline=True) + embed.add_field(name="Play:", value=f"`/play-favorite `", inline=True) - except Exception as e: - logger.error(f"Error in favorites command: {e}") - await interaction.response.send_message("❌ An error occurred while loading favorites.", ephemeral=True) + await interaction.response.send_message(embed=embed, ephemeral=False) @bot.tree.command( - name='list-favorites', - description="List all favorites (text only, mobile-friendly)" + name='play-favorite', + description="Play one of my favorited radio stations" ) -@discord.app_commands.checks.cooldown(rate=1, per=5) +@discord.app_commands.checks.cooldown(rate=1, per=5, key=lambda i: i.guild_id) +@bot_not_in_maintenance() @is_channel() -async def list_favorites(interaction: discord.Interaction): - try: - favorites_manager = get_favorites_manager() - favorites_list = favorites_manager.get_favorites(interaction.guild.id) - - embed = create_favorites_list_embed(favorites_list, interaction.guild.name) - await interaction.response.send_message(embed=embed) +async def play_favorite(interaction: discord.Interaction, index: int): + favorites = await PERSONAL_FAVORITES_MANAGER.retrieve_user_favorites(user_id=interaction.user.id) + index = index-1 - except Exception as e: - logger.error(f"Error in list_favorites command: {e}") - await interaction.response.send_message("❌ An error occurred while listing favorites.", ephemeral=True) - -@bot.tree.command( - name='remove-favorite', - description="Remove a favorite radio station" -) -@discord.app_commands.checks.cooldown(rate=1, per=5) -@is_channel() -async def remove_favorite(interaction: discord.Interaction, number: int): - # Check permissions - perm_manager = get_permission_manager() - if not perm_manager.can_remove_favorites(interaction.guild.id, interaction.user): - await interaction.response.send_message( - "❌ You don't have permission to remove favorites. Ask an admin to assign you the appropriate role.", - ephemeral=True - ) + if len(favorites) <= 0: + await interaction.response.send_message(content="ðŸĨ€ Looks like you don't have any favorites? Try adding one first", ephemeral=True) return + if index < 0 or index >= len(favorites): + await interaction.response.send_message(content="ðŸ‘Ļ‍ðŸĶŊ‍➡ïļ I can't find that favorite. Try again?", ephemeral=True) + return + fav_url = favorites[index][0].stream_url - try: - favorites_manager = get_favorites_manager() - - # Check if favorite exists first - favorite = favorites_manager.get_favorite_by_number(interaction.guild.id, number) - if not favorite: - await interaction.response.send_message(f"❌ Favorite #{number} not found.", ephemeral=True) - return - - # Create confirmation view - view = ConfirmationView("remove", f"favorite #{number}: {favorite['station_name']}") - await interaction.response.send_message( - f"⚠ïļ Are you sure you want to remove favorite #{number}: **{favorite['station_name']}**?\n" - f"This will reorder all subsequent favorites.", - view=view - ) - - # Wait for confirmation - await view.wait() - - if view.confirmed: - result = favorites_manager.remove_favorite(interaction.guild.id, number) - if result['success']: - await interaction.followup.send( - f"✅ Removed **{result['station_name']}** from favorites. Subsequent favorites have been renumbered." - ) - else: - await interaction.followup.send(f"❌ Failed to remove favorite: {result['error']}") + response_message = f"ðŸ“Ą Starting your favorite channel {fav_url}" + await interaction.response.send_message(response_message, ephemeral=True) - except Exception as e: - logger.error(f"Error in remove_favorite command: {e}") - if interaction.response.is_done(): - await interaction.followup.send("❌ An error occurred while removing the favorite.", ephemeral=True) - else: - await interaction.response.send_message("❌ An error occurred while removing the favorite.", ephemeral=True) + if await play_stream(interaction, fav_url): + STATE_MANAGER.set_state(interaction.guild_id, 'private_stream', False) @bot.tree.command( - name='setup-roles', - description="Configure which Discord roles can manage favorites" + name='remove-favorite', + description="Remove one of my favorited radio stations" ) -@discord.app_commands.checks.cooldown(rate=1, per=5) +@discord.app_commands.checks.cooldown(rate=1, per=5, key=lambda i: i.guild_id) +@bot_not_in_maintenance() @is_channel() -async def setup_roles(interaction: discord.Interaction, role: discord.Role = None, permission_level: str = None): - # Check permissions - perm_manager = get_permission_manager() - if not perm_manager.can_manage_roles(interaction.guild.id, interaction.user): - await interaction.response.send_message( - "❌ You don't have permission to manage role assignments. Ask an admin to assign you the appropriate role.", - ephemeral=True - ) - return +async def remove_favorite(interaction: discord.Interaction, index: int): + favorites = await PERSONAL_FAVORITES_MANAGER.retrieve_user_favorites(user_id=interaction.user.id) + index = index-1 - try: - # If no parameters provided, show current setup - if not role and not permission_level: - role_assignments = perm_manager.get_server_role_assignments(interaction.guild.id) - available_roles = perm_manager.get_available_permission_roles() - - embed = create_role_setup_embed(role_assignments, available_roles, interaction.guild.name) - await interaction.response.send_message(embed=embed) - return - - # Both parameters required for assignment - if not role or not permission_level: - await interaction.response.send_message( - "❌ Please provide both a role and permission level.\n" - "Example: `/setup-roles @DJ dj`\n" - "Available levels: user, dj, radio manager, admin", - ephemeral=True - ) - return - - # Validate permission level - available_roles = perm_manager.get_available_permission_roles() - valid_levels = [r['role_name'] for r in available_roles] - - if permission_level.lower() not in valid_levels: - await interaction.response.send_message( - f"❌ Invalid permission level. Available levels: {', '.join(valid_levels)}", - ephemeral=True - ) - return - - # Assign the role - success = perm_manager.assign_role_permission( - guild_id=interaction.guild.id, - role_id=role.id, - role_name=permission_level.lower() - ) + if len(favorites) <= 0: + await interaction.response.send_message(content="ðŸĨ€ Looks like you don't have any favorites? Try adding one first", ephemeral=True) + return + if index < 0: + await interaction.response.send_message(content="🗅 Favorite index must be greater than 0", ephemeral=True) + return + if index >= len(favorites): + await interaction.response.send_message(content="ðŸ‘Ļ‍ðŸĶŊ‍➡ïļ You don't have that many favorites in your favorites list", ephemeral=True) + return - if success: - await interaction.response.send_message( - f"✅ Assigned role {role.mention} to permission level **{permission_level}**" - ) - else: - await interaction.response.send_message( - "❌ Failed to assign role permission. Please check the permission level is valid.", - ephemeral=True - ) + favorite_to_delete = favorites[index][0] + if await PERSONAL_FAVORITES_MANAGER.delete_user_favorite(favorite_to_delete): + await interaction.response.send_message(content="ðŸŠĶ Favorite deleted", ephemeral=True) + else: + await interaction.response.send_message(content="Something went wrong while deleting your favorite", ephemeral=True) - except Exception as e: - logger.error(f"Error in setup_roles command: {e}") - if interaction.response.is_done(): - await interaction.followup.send("❌ An error occurred while setting up roles.", ephemeral=True) - else: - await interaction.response.send_message("❌ An error occurred while setting up roles.", ephemeral=True) ### END FAVORITES COMMANDS ### diff --git a/input_validator.py b/input_validator.py index 2c9cf5a..5bd146e 100644 --- a/input_validator.py +++ b/input_validator.py @@ -6,31 +6,30 @@ import re import logging from typing import Dict, Any, Optional +import urllib import validators logger = logging.getLogger('discord') class InputValidator: """Validates and sanitizes user inputs for security""" - + # Maximum lengths for various inputs MAX_STATION_NAME_LENGTH = 100 - MAX_URL_LENGTH = 2048 - MAX_SEARCH_TERM_LENGTH = 50 - + # Allowed characters for station names (alphanumeric, spaces, common punctuation) - STATION_NAME_PATTERN = re.compile(r'^[a-zA-Z0-9\s\-_.,!()&\'"]+$') - + STATION_NAME_PATTERN = re.compile(r'^[a-zA-Z0-9\s\-_\.,!\(\)&\'"]+$') + def __init__(self): pass - + def validate_url(self, url: str) -> Dict[str, Any]: """ Validate and sanitize a stream URL - + Args: url: URL to validate - + Returns: Dict with 'valid', 'sanitized_url', 'error' keys """ @@ -40,18 +39,10 @@ def validate_url(self, url: str) -> Dict[str, Any]: 'sanitized_url': None, 'error': 'URL cannot be empty' } - + # Remove leading/trailing whitespace url = url.strip() - - # Check length - if len(url) > self.MAX_URL_LENGTH: - return { - 'valid': False, - 'sanitized_url': None, - 'error': f'URL too long (max {self.MAX_URL_LENGTH} characters)' - } - + # Validate URL format if not validators.url(url): return { @@ -59,38 +50,31 @@ def validate_url(self, url: str) -> Dict[str, Any]: 'sanitized_url': None, 'error': 'Invalid URL format' } - + # Check for allowed protocols - allowed_protocols = ['http', 'https'] - protocol = url.split('://')[0].lower() - if protocol not in allowed_protocols: - return { - 'valid': False, - 'sanitized_url': None, - 'error': f'Protocol not allowed. Use: {", ".join(allowed_protocols)}' - } - - # Basic security checks - if self._contains_suspicious_patterns(url): + allowed_schemes = ['http', 'https'] + sliced_url = urllib.parse.urlparse(url) + scheme = sliced_url.scheme + if scheme not in allowed_schemes: return { 'valid': False, 'sanitized_url': None, - 'error': 'URL contains suspicious patterns' + 'error': f'Protocol not allowed. Use: {", ".join(allowed_schemes)}' } - + return { 'valid': True, 'sanitized_url': url, 'error': None } - + def validate_station_name(self, name: str) -> Dict[str, Any]: """ Validate and sanitize a station name - + Args: name: Station name to validate - + Returns: Dict with 'valid', 'sanitized_name', 'error' keys """ @@ -100,10 +84,10 @@ def validate_station_name(self, name: str) -> Dict[str, Any]: 'sanitized_name': None, 'error': 'Station name cannot be empty' } - + # Remove leading/trailing whitespace and normalize name = name.strip() - + # Check length if len(name) > self.MAX_STATION_NAME_LENGTH: return { @@ -111,7 +95,7 @@ def validate_station_name(self, name: str) -> Dict[str, Any]: 'sanitized_name': None, 'error': f'Station name too long (max {self.MAX_STATION_NAME_LENGTH} characters)' } - + # Check for allowed characters if not self.STATION_NAME_PATTERN.match(name): return { @@ -119,70 +103,23 @@ def validate_station_name(self, name: str) -> Dict[str, Any]: 'sanitized_name': None, 'error': 'Station name contains invalid characters' } - + # Remove excessive whitespace sanitized_name = re.sub(r'\s+', ' ', name) - + return { 'valid': True, 'sanitized_name': sanitized_name, 'error': None } - - def validate_search_term(self, search_term: str) -> Dict[str, Any]: - """ - Validate and sanitize a search term - - Args: - search_term: Search term to validate - - Returns: - Dict with 'valid', 'sanitized_term', 'error' keys - """ - if not search_term: - return { - 'valid': False, - 'sanitized_term': None, - 'error': 'Search term cannot be empty' - } - - # Remove leading/trailing whitespace - search_term = search_term.strip() - - # Check length - if len(search_term) > self.MAX_SEARCH_TERM_LENGTH: - return { - 'valid': False, - 'sanitized_term': None, - 'error': f'Search term too long (max {self.MAX_SEARCH_TERM_LENGTH} characters)' - } - - # Basic sanitization - remove SQL wildcards that could be abused - sanitized_term = search_term.replace('%', '').replace('_', '') - - # Remove excessive whitespace - sanitized_term = re.sub(r'\s+', ' ', sanitized_term) - - if not sanitized_term: - return { - 'valid': False, - 'sanitized_term': None, - 'error': 'Search term contains only invalid characters' - } - - return { - 'valid': True, - 'sanitized_term': sanitized_term, - 'error': None - } - + def validate_favorite_number(self, number: int) -> Dict[str, Any]: """ Validate a favorite number - + Args: number: Favorite number to validate - + Returns: Dict with 'valid', 'error' keys """ @@ -191,31 +128,31 @@ def validate_favorite_number(self, number: int) -> Dict[str, Any]: 'valid': False, 'error': 'Favorite number must be an integer' } - + if number < 1: return { 'valid': False, 'error': 'Favorite number must be positive' } - + if number > 9999: # Reasonable upper limit return { 'valid': False, 'error': 'Favorite number too large (max 9999)' } - + return { 'valid': True, 'error': None } - + def validate_role_name(self, role_name: str) -> Dict[str, Any]: """ Validate a permission role name - + Args: role_name: Role name to validate - + Returns: Dict with 'valid', 'sanitized_name', 'error' keys """ @@ -225,54 +162,25 @@ def validate_role_name(self, role_name: str) -> Dict[str, Any]: 'sanitized_name': None, 'error': 'Role name cannot be empty' } - + # Normalize to lowercase role_name = role_name.strip().lower() - + # Whitelist valid role names valid_roles = {'user', 'dj', 'radio manager', 'admin'} - + if role_name not in valid_roles: return { 'valid': False, 'sanitized_name': None, 'error': f'Invalid role name. Valid options: {", ".join(valid_roles)}' } - + return { 'valid': True, 'sanitized_name': role_name, 'error': None } - - def _contains_suspicious_patterns(self, url: str) -> bool: - """ - Check for suspicious patterns in URLs that might indicate attacks - - Args: - url: URL to check - - Returns: - True if suspicious patterns found - """ - suspicious_patterns = [ - # SQL injection attempts - r'(\bUNION\b|\bSELECT\b|\bINSERT\b|\bDELETE\b|\bDROP\b)', - # Script injection - r'( bool: + # not a url + # if not self.input_validator.validate_url(stream_url)['valid']: + # self.logger.warning('Failed to validate url %s as url', stream_url) + # return False + # # station name has invalid characters + # if not self.input_validator.validate_station_name(station_name)['valid']: + # self.logger.warning('Failed to validate station_name %s as an allowed name', station_name) + # return False + # # not a real station + # if not self.stream_validator.validate_stream(stream_url)['valid']: + # self.logger.warning('Failed to validate stream %s as a radio station', stream_url) + # return False + + async with self.ASYNC_SESSION_LOCAL() as session: + stmt = select(func.count()).where(PersonalFavorite.user_id == user_id).where(PersonalFavorite.stream_url == stream_url) + matched_favorites = await session.execute(stmt) + stmt = select(func.count()).where(PersonalFavorite.user_id == user_id) + fav_count = await session.execute(stmt) + await session.commit() + + # station url already exists + if matched_favorites.scalar() > 0: + self.logger.error('User already has %s as a favorited station', stream_url) + return False + # user has max favorites + if fav_count.scalar() > self.MAX_FAVORITES: + self.logger.error('%s already has maximum favorites', user_id) + return False + + personal_favorite = PersonalFavorite(user_id=user_id, stream_url=stream_url, station_name=station_name) + async with self.ASYNC_SESSION_LOCAL() as session: + session.add(personal_favorite) + await session.commit() + return True + async def retrieve_user_favorites(self, user_id: int) -> list[PersonalFavorite]: + async with self.ASYNC_SESSION_LOCAL() as session: + stmt = self._all_user_favorites_statement(user_id) + user_favorites = await session.execute(stmt) + await session.commit() + + return user_favorites.all() + async def delete_user_favorite(self, favorite_to_delete: PersonalFavorite) -> bool: + if not favorite_to_delete: + self.logger.error("User {user_id} tried to delete favorite index {favorite_index}") + return False + + self.logger.debug(favorite_to_delete) + self.logger.debug(favorite_to_delete.id) + + async with self.ASYNC_SESSION_LOCAL() as session: + await session.delete(favorite_to_delete) + await session.commit() + return True + + + + + @staticmethod + def _all_user_favorites_statement(user_id: int): + return select(PersonalFavorite).where(PersonalFavorite.user_id == user_id).order_by(PersonalFavorite.__table__.c.creation_date) diff --git a/stream_validator.py b/stream_validator.py index 836922f..87ef0a1 100644 --- a/stream_validator.py +++ b/stream_validator.py @@ -13,7 +13,7 @@ class StreamValidator: """Validates radio streams and extracts metadata""" - async def validate_stream(self, url: str) -> Dict[str, Any]: + def validate_stream(self, url: str) -> Dict[str, Any]: """ Validate stream and return metadata