Add /set-image command for updating player fancy cards and headshots. Features: - Single command with fancy-card/headshot choice parameter - Comprehensive URL validation (format + accessibility testing) - Permission system (users can edit org players, admins can edit all) - Preview embed with confirmation dialog before database update - Player name autocomplete prioritizing user's team - HTTP HEAD request to verify URL accessibility and content-type Implementation: - New commands/profile/ package with ImageCommands cog - Two-stage URL validation (format check + accessibility test) - Permission checking via Team.is_same_organization() - Interactive confirmation view with 180s timeout - Updates player.vanity_card or player.headshot field Testing: - 23 comprehensive tests covering validation and permissions - Uses aioresponses for HTTP mocking (project standard) - Test coverage for admin/user permissions and organization checks Documentation: - Comprehensive README.md with usage guide and troubleshooting - Updated PRE_LAUNCH_ROADMAP.md to mark feature complete 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
479 lines
17 KiB
Python
479 lines
17 KiB
Python
"""
|
|
Player Image Management Commands
|
|
|
|
Allows users to update player fancy card and headshot images for players
|
|
on teams they own. Admins can update any player's images.
|
|
"""
|
|
from typing import Optional, List, Tuple
|
|
import asyncio
|
|
import aiohttp
|
|
|
|
import discord
|
|
from discord import app_commands
|
|
from discord.ext import commands
|
|
|
|
from services.player_service import player_service
|
|
from services.team_service import team_service
|
|
from utils.logging import get_contextual_logger
|
|
from utils.decorators import logged_command
|
|
from constants import SBA_CURRENT_SEASON
|
|
from views.embeds import EmbedColors, EmbedTemplate
|
|
from views.base import BaseView
|
|
from models.player import Player
|
|
|
|
|
|
# URL Validation Functions
|
|
|
|
def validate_url_format(url: str) -> Tuple[bool, str]:
|
|
"""
|
|
Validate URL format for image links.
|
|
|
|
Args:
|
|
url: URL to validate
|
|
|
|
Returns:
|
|
Tuple of (is_valid, error_message)
|
|
If valid, error_message is empty string
|
|
"""
|
|
# Length check
|
|
if len(url) > 500:
|
|
return False, "URL too long (max 500 characters)"
|
|
|
|
# Protocol check
|
|
if not url.startswith(('http://', 'https://')):
|
|
return False, "URL must start with http:// or https://"
|
|
|
|
# Image extension check
|
|
valid_extensions = ('.jpg', '.jpeg', '.png', '.gif', '.webp')
|
|
url_lower = url.lower()
|
|
|
|
# Check if URL ends with valid extension (ignore query params)
|
|
base_url = url_lower.split('?')[0] # Remove query parameters
|
|
if not any(base_url.endswith(ext) for ext in valid_extensions):
|
|
return False, f"URL must end with a valid image extension: {', '.join(valid_extensions)}"
|
|
|
|
return True, ""
|
|
|
|
|
|
async def test_url_accessibility(url: str) -> Tuple[bool, str]:
|
|
"""
|
|
Test if URL is accessible and returns image content.
|
|
|
|
Args:
|
|
url: URL to test
|
|
|
|
Returns:
|
|
Tuple of (is_accessible, error_message)
|
|
If accessible, error_message is empty string
|
|
"""
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.head(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
|
|
if response.status != 200:
|
|
return False, f"URL returned status {response.status}"
|
|
|
|
# Check content-type header
|
|
content_type = response.headers.get('content-type', '').lower()
|
|
if content_type and not content_type.startswith('image/'):
|
|
return False, f"URL does not return an image (content-type: {content_type})"
|
|
|
|
return True, ""
|
|
|
|
except aiohttp.ClientError as e:
|
|
return False, f"Could not access URL: {str(e)}"
|
|
except asyncio.TimeoutError:
|
|
return False, "URL request timed out after 5 seconds"
|
|
except Exception as e:
|
|
return False, f"Error testing URL: {str(e)}"
|
|
|
|
|
|
# Permission Checking
|
|
|
|
async def can_edit_player_image(
|
|
interaction: discord.Interaction,
|
|
player: Player,
|
|
season: int,
|
|
logger
|
|
) -> Tuple[bool, str]:
|
|
"""
|
|
Check if user can edit player's image.
|
|
|
|
Args:
|
|
interaction: Discord interaction object
|
|
player: Player to check permissions for
|
|
season: Season to check
|
|
logger: Logger for debug output
|
|
|
|
Returns:
|
|
Tuple of (has_permission, error_message)
|
|
If has permission, error_message is empty string
|
|
"""
|
|
# Admins can edit anyone
|
|
if interaction.user.guild_permissions.administrator:
|
|
logger.debug("User is admin, granting permission", user_id=interaction.user.id)
|
|
return True, ""
|
|
|
|
# Check if player has a team
|
|
if not player.team:
|
|
return False, "Cannot determine player's team ownership"
|
|
|
|
# Get user's teams (all roster types)
|
|
user_teams = await team_service.get_teams_by_owner(interaction.user.id, season)
|
|
|
|
if not user_teams:
|
|
return False, "You don't own any teams in the current season"
|
|
|
|
# Check if any of user's teams are in same organization as player's team
|
|
for user_team in user_teams:
|
|
if user_team.is_same_organization(player.team):
|
|
logger.debug(
|
|
"User owns organization, granting permission",
|
|
user_id=interaction.user.id,
|
|
user_team=user_team.abbrev,
|
|
player_team=player.team.abbrev
|
|
)
|
|
return True, ""
|
|
|
|
# User doesn't own this organization
|
|
player_org = player.team._get_base_abbrev()
|
|
return False, f"You don't own a team in the {player_org} organization"
|
|
|
|
|
|
# Confirmation View
|
|
|
|
class ImageUpdateConfirmView(BaseView):
|
|
"""Confirmation view showing image preview before updating."""
|
|
|
|
def __init__(self, player: Player, image_url: str, image_type: str, user_id: int):
|
|
super().__init__(timeout=180.0, user_id=user_id)
|
|
self.player = player
|
|
self.image_url = image_url
|
|
self.image_type = image_type
|
|
self.confirmed = False
|
|
|
|
@discord.ui.button(label="Confirm Update", style=discord.ButtonStyle.success, emoji="✅")
|
|
async def confirm_button(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
"""Confirm the image update."""
|
|
self.confirmed = True
|
|
|
|
# Disable all buttons
|
|
for item in self.children:
|
|
if hasattr(item, 'disabled'):
|
|
item.disabled = True # type: ignore
|
|
|
|
await interaction.response.edit_message(view=self)
|
|
self.stop()
|
|
|
|
@discord.ui.button(label="Cancel", style=discord.ButtonStyle.secondary, emoji="❌")
|
|
async def cancel_button(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
"""Cancel the image update."""
|
|
self.confirmed = False
|
|
|
|
# Disable all buttons
|
|
for item in self.children:
|
|
if hasattr(item, 'disabled'):
|
|
item.disabled = True # type: ignore
|
|
|
|
await interaction.response.edit_message(view=self)
|
|
self.stop()
|
|
|
|
|
|
# Autocomplete
|
|
|
|
async def player_name_autocomplete(
|
|
interaction: discord.Interaction,
|
|
current: str,
|
|
) -> List[app_commands.Choice[str]]:
|
|
"""Autocomplete for player names, prioritizing user's team players."""
|
|
if len(current) < 2:
|
|
return []
|
|
|
|
try:
|
|
from utils.autocomplete import player_autocomplete_with_team_priority
|
|
return await player_autocomplete_with_team_priority(interaction, current)
|
|
except Exception:
|
|
# Fallback to basic autocomplete
|
|
try:
|
|
players = await player_service.search_players(current, limit=25, season=SBA_CURRENT_SEASON)
|
|
|
|
choices = []
|
|
for player in players[:25]:
|
|
display_name = f"{player.name} ({player.primary_position})"
|
|
if hasattr(player, 'team') and player.team:
|
|
display_name += f" - {player.team.abbrev}"
|
|
|
|
choices.append(app_commands.Choice(
|
|
name=display_name,
|
|
value=player.name
|
|
))
|
|
|
|
return choices
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
# Main Command Cog
|
|
|
|
class ImageCommands(commands.Cog):
|
|
"""Player image management command handlers."""
|
|
|
|
def __init__(self, bot: commands.Bot):
|
|
self.bot = bot
|
|
self.logger = get_contextual_logger(f'{__name__}.ImageCommands')
|
|
self.logger.info("ImageCommands cog initialized")
|
|
|
|
@app_commands.command(
|
|
name="set-image",
|
|
description="Update a player's fancy card or headshot image"
|
|
)
|
|
@app_commands.describe(
|
|
image_type="Type of image to update",
|
|
player_name="Player name (use autocomplete)",
|
|
image_url="Direct URL to the image file"
|
|
)
|
|
@app_commands.choices(image_type=[
|
|
app_commands.Choice(name="Fancy Card", value="fancy-card"),
|
|
app_commands.Choice(name="Headshot", value="headshot")
|
|
])
|
|
@app_commands.autocomplete(player_name=player_name_autocomplete)
|
|
@logged_command("/set-image")
|
|
async def set_image(
|
|
self,
|
|
interaction: discord.Interaction,
|
|
image_type: app_commands.Choice[str],
|
|
player_name: str,
|
|
image_url: str
|
|
):
|
|
"""Update a player's image (fancy card or headshot)."""
|
|
# Defer response for potentially slow operations
|
|
await interaction.response.defer(ephemeral=True)
|
|
|
|
# Get the image type value
|
|
img_type = image_type.value
|
|
field_name = "vanity_card" if img_type == "fancy-card" else "headshot"
|
|
display_name = "Fancy Card" if img_type == "fancy-card" else "Headshot"
|
|
|
|
self.logger.info(
|
|
"Image update requested",
|
|
user_id=interaction.user.id,
|
|
player_name=player_name,
|
|
image_type=img_type
|
|
)
|
|
|
|
# Step 1: Validate URL format
|
|
is_valid_format, format_error = validate_url_format(image_url)
|
|
if not is_valid_format:
|
|
self.logger.warning("Invalid URL format", url=image_url, error=format_error)
|
|
embed = EmbedTemplate.error(
|
|
title="Invalid URL Format",
|
|
description=f"❌ {format_error}\n\n"
|
|
f"**Requirements:**\n"
|
|
f"• Must start with `http://` or `https://`\n"
|
|
f"• Must end with `.jpg`, `.jpeg`, `.png`, `.gif`, or `.webp`\n"
|
|
f"• Maximum 500 characters"
|
|
)
|
|
await interaction.followup.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
# Step 2: Test URL accessibility
|
|
self.logger.debug("Testing URL accessibility", url=image_url)
|
|
is_accessible, access_error = await test_url_accessibility(image_url)
|
|
if not is_accessible:
|
|
self.logger.warning("URL not accessible", url=image_url, error=access_error)
|
|
embed = EmbedTemplate.error(
|
|
title="URL Not Accessible",
|
|
description=f"❌ {access_error}\n\n"
|
|
f"**Please check:**\n"
|
|
f"• URL is correct and not expired\n"
|
|
f"• Image host is online\n"
|
|
f"• URL points directly to an image file\n"
|
|
f"• URL is publicly accessible"
|
|
)
|
|
await interaction.followup.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
# Step 3: Find player
|
|
self.logger.debug("Searching for player", player_name=player_name)
|
|
players = await player_service.get_players_by_name(player_name, SBA_CURRENT_SEASON)
|
|
|
|
if not players:
|
|
self.logger.warning("Player not found", player_name=player_name)
|
|
embed = EmbedTemplate.error(
|
|
title="Player Not Found",
|
|
description=f"❌ No player found matching `{player_name}` in the current season."
|
|
)
|
|
await interaction.followup.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
# Handle multiple matches - try exact match
|
|
player = None
|
|
if len(players) == 1:
|
|
player = players[0]
|
|
else:
|
|
# Try exact match
|
|
for p in players:
|
|
if p.name.lower() == player_name.lower():
|
|
player = p
|
|
break
|
|
|
|
if player is None:
|
|
# Multiple candidates, ask user to be more specific
|
|
player_list = "\n".join([f"• {p.name} ({p.primary_position})" for p in players[:10]])
|
|
embed = EmbedTemplate.info(
|
|
title="Multiple Players Found",
|
|
description=f"🔍 Multiple players match `{player_name}`:\n\n{player_list}\n\n"
|
|
f"Please use the exact name from autocomplete."
|
|
)
|
|
await interaction.followup.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
self.logger.info("Player found", player_id=player.id, player_name=player.name)
|
|
|
|
# Step 4: Check permissions
|
|
has_permission, permission_error = await can_edit_player_image(
|
|
interaction, player, SBA_CURRENT_SEASON, self.logger
|
|
)
|
|
|
|
if not has_permission:
|
|
self.logger.warning(
|
|
"Permission denied",
|
|
user_id=interaction.user.id,
|
|
player_id=player.id,
|
|
error=permission_error
|
|
)
|
|
embed = EmbedTemplate.error(
|
|
title="Permission Denied",
|
|
description=f"❌ {permission_error}\n\n"
|
|
f"You can only update images for players on teams you own."
|
|
)
|
|
await interaction.followup.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
# Step 5: Show preview with confirmation
|
|
self.logger.debug("Creating preview embed")
|
|
preview_embed = EmbedTemplate.create_base_embed(
|
|
title=f"🖼️ Update {display_name} for {player.name}",
|
|
description=f"Preview the new {display_name.lower()} below. Click **Confirm Update** to save this change.",
|
|
color=EmbedColors.INFO
|
|
)
|
|
|
|
# Add current image info
|
|
current_image = getattr(player, field_name, None)
|
|
if current_image:
|
|
preview_embed.add_field(
|
|
name="Current Image",
|
|
value="Will be replaced",
|
|
inline=True
|
|
)
|
|
else:
|
|
preview_embed.add_field(
|
|
name="Current Image",
|
|
value="None set",
|
|
inline=True
|
|
)
|
|
|
|
# Add player info
|
|
preview_embed.add_field(
|
|
name="Player",
|
|
value=f"{player.name} ({player.primary_position})",
|
|
inline=True
|
|
)
|
|
|
|
if hasattr(player, 'team') and player.team:
|
|
preview_embed.add_field(
|
|
name="Team",
|
|
value=player.team.abbrev,
|
|
inline=True
|
|
)
|
|
|
|
# Set the new image as thumbnail for preview
|
|
preview_embed.set_thumbnail(url=image_url)
|
|
|
|
preview_embed.set_footer(text="This preview shows how the image will appear. Confirm to save.")
|
|
|
|
# Create confirmation view
|
|
confirm_view = ImageUpdateConfirmView(
|
|
player=player,
|
|
image_url=image_url,
|
|
image_type=img_type,
|
|
user_id=interaction.user.id
|
|
)
|
|
|
|
await interaction.followup.send(embed=preview_embed, view=confirm_view, ephemeral=True)
|
|
|
|
# Wait for confirmation
|
|
await confirm_view.wait()
|
|
|
|
if not confirm_view.confirmed:
|
|
self.logger.info("Image update cancelled by user", player_id=player.id)
|
|
cancelled_embed = EmbedTemplate.info(
|
|
title="Update Cancelled",
|
|
description=f"No changes were made to {player.name}'s {display_name.lower()}."
|
|
)
|
|
await interaction.edit_original_response(embed=cancelled_embed, view=None)
|
|
return
|
|
|
|
# Step 6: Update database
|
|
self.logger.info(
|
|
"Updating player image",
|
|
player_id=player.id,
|
|
field=field_name,
|
|
url_length=len(image_url)
|
|
)
|
|
|
|
update_data = {field_name: image_url}
|
|
updated_player = await player_service.update_player(player.id, update_data)
|
|
|
|
if updated_player is None:
|
|
self.logger.error("Failed to update player", player_id=player.id)
|
|
error_embed = EmbedTemplate.error(
|
|
title="Update Failed",
|
|
description="❌ An error occurred while updating the player's image. Please try again."
|
|
)
|
|
await interaction.edit_original_response(embed=error_embed, view=None)
|
|
return
|
|
|
|
# Step 7: Send success message
|
|
self.logger.info(
|
|
"Player image updated successfully",
|
|
player_id=player.id,
|
|
field=field_name,
|
|
user_id=interaction.user.id
|
|
)
|
|
|
|
success_embed = EmbedTemplate.success(
|
|
title="✅ Image Updated Successfully!",
|
|
description=f"**{display_name}** for **{player.name}** has been updated."
|
|
)
|
|
|
|
success_embed.add_field(
|
|
name="Player",
|
|
value=f"{player.name} ({player.primary_position})",
|
|
inline=True
|
|
)
|
|
|
|
if hasattr(player, 'team') and player.team:
|
|
success_embed.add_field(
|
|
name="Team",
|
|
value=player.team.abbrev,
|
|
inline=True
|
|
)
|
|
|
|
success_embed.add_field(
|
|
name="Image Type",
|
|
value=display_name,
|
|
inline=True
|
|
)
|
|
|
|
# Show the new image
|
|
success_embed.set_thumbnail(url=image_url)
|
|
|
|
success_embed.set_footer(text=f"Updated by {interaction.user.display_name}")
|
|
|
|
await interaction.edit_original_response(embed=success_embed, view=None)
|
|
|
|
|
|
async def setup(bot: commands.Bot):
|
|
"""Load the image management commands cog."""
|
|
await bot.add_cog(ImageCommands(bot))
|