Travis Vasceannie
3acb779569
- Deleted .env.example file as it is no longer needed. - Added .gitignore to manage ignored files and directories. - Introduced CLAUDE.md for AI provider integration documentation. - Created dev.sh for development setup and scripts. - Updated Dockerfile and Dockerfile.production for improved build processes. - Added multiple test files and directories for comprehensive testing. - Introduced new utility and service files for enhanced functionality. - Organized codebase with new directories and files for better maintainability.
1325 lines
49 KiB
Python
1325 lines
49 KiB
Python
"""
|
||
Discord UI Components for Voice Chat Quote Bot
|
||
|
||
Implements interactive Discord UI elements including embeds, buttons, select menus,
|
||
modals, and views for enhanced user engagement and bot interaction.
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
from collections.abc import Callable
|
||
from datetime import datetime, timezone
|
||
from typing import TYPE_CHECKING, Any, Awaitable
|
||
|
||
import discord
|
||
from discord import ui
|
||
from discord.ext import commands
|
||
|
||
from config.consent_templates import ConsentMessages, ConsentTemplates
|
||
from core.database import DatabaseManager
|
||
from core.memory_manager import MemoryManager
|
||
from services.quotes.quote_analyzer import QuoteAnalyzer
|
||
|
||
if TYPE_CHECKING:
|
||
from core.consent_manager import ConsentManager
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class EmbedBuilder:
|
||
"""Comprehensive utility class for building Discord embeds"""
|
||
|
||
@staticmethod
|
||
def create_quote_embed(
|
||
quote_data: dict[str, Any], include_analysis: bool = True
|
||
) -> discord.Embed:
|
||
"""Create a rich embed for displaying quotes"""
|
||
try:
|
||
embed = discord.Embed(
|
||
title="💬 Memorable Quote",
|
||
description=f"*\"{quote_data['quote']}\"*",
|
||
color=EmbedBuilder._get_score_color(quote_data.get("overall_score", 0)),
|
||
timestamp=quote_data.get("timestamp", datetime.now(timezone.utc)),
|
||
)
|
||
|
||
# Add user information
|
||
if quote_data.get("username"):
|
||
embed.set_author(
|
||
name=quote_data["username"], icon_url=quote_data.get("avatar_url")
|
||
)
|
||
|
||
# Add score information
|
||
overall_score = quote_data.get("overall_score", 0)
|
||
embed.add_field(
|
||
name="📊 Overall Score",
|
||
value=f"**{overall_score:.1f}/10**",
|
||
inline=True,
|
||
)
|
||
|
||
# Add category scores
|
||
if include_analysis:
|
||
scores = []
|
||
score_emojis = {
|
||
"funny_score": "😂",
|
||
"dark_score": "🖤",
|
||
"silly_score": "🤪",
|
||
"suspicious_score": "🤔",
|
||
"asinine_score": "🙄",
|
||
}
|
||
|
||
for score_type, emoji in score_emojis.items():
|
||
score = quote_data.get(score_type, 0)
|
||
if score > 5.0:
|
||
scores.append(f"{emoji} {score:.1f}")
|
||
|
||
if scores:
|
||
embed.add_field(
|
||
name="🎯 Category Scores", value=" | ".join(scores), inline=True
|
||
)
|
||
|
||
# Add laughter information
|
||
laughter_duration = quote_data.get("laughter_duration", 0)
|
||
if laughter_duration > 0:
|
||
embed.add_field(
|
||
name="😄 Laughter", value=f"{laughter_duration:.1f}s", inline=True
|
||
)
|
||
|
||
# Add audio clip info if available
|
||
if quote_data.get("audio_clip_path"):
|
||
embed.add_field(
|
||
name="🎵 Audio Available", value="Audio clip recorded", inline=True
|
||
)
|
||
|
||
embed.set_footer(text="Quote Analysis • Powered by AI")
|
||
|
||
return embed
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating quote embed: {e}")
|
||
return EmbedBuilder.error("Error", "Failed to create quote display")
|
||
|
||
@staticmethod
|
||
def quote_embed(
|
||
quote_data: dict[str, Any], show_scores: bool = True
|
||
) -> discord.Embed:
|
||
"""Build embed for displaying a quote (alternative format)"""
|
||
embed = discord.Embed(
|
||
title=f"📝 Quote from {quote_data.get('speaker_label', 'Unknown')}",
|
||
description=f'"{quote_data.get("quote", "")}"',
|
||
color=EmbedBuilder._get_score_color(quote_data.get("overall_score", 0)),
|
||
)
|
||
|
||
if show_scores:
|
||
scores_text = (
|
||
f"**Funny:** {quote_data.get('funny_score', 0)}/10\n"
|
||
f"**Overall:** {quote_data.get('overall_score', 0)}/10"
|
||
)
|
||
embed.add_field(name="📊 Scores", value=scores_text, inline=True)
|
||
|
||
# Add timestamp
|
||
if quote_data.get("timestamp"):
|
||
embed.timestamp = quote_data["timestamp"]
|
||
|
||
return embed
|
||
|
||
@staticmethod
|
||
def create_personality_embed(personality_data: dict[str, Any]) -> discord.Embed:
|
||
"""Create embed for personality profile display"""
|
||
try:
|
||
embed = discord.Embed(
|
||
title="🧠 Personality Profile",
|
||
description="AI-generated personality insights based on conversation history",
|
||
color=0x9B59B6,
|
||
timestamp=personality_data.get(
|
||
"last_updated", datetime.now(timezone.utc)
|
||
),
|
||
)
|
||
|
||
# Humor preferences
|
||
humor_prefs = personality_data.get("humor_preferences", {})
|
||
if humor_prefs:
|
||
humor_text = ""
|
||
for humor_type, score in humor_prefs.items():
|
||
emoji = {
|
||
"funny": "😂",
|
||
"dark": "🖤",
|
||
"silly": "🤪",
|
||
"suspicious": "🤔",
|
||
"asinine": "🙄",
|
||
}.get(humor_type, "📊")
|
||
bar = "█" * int(score) + "░" * (10 - int(score))
|
||
humor_text += (
|
||
f"{emoji} **{humor_type.title()}**: {bar} {score:.1f}/10\n"
|
||
)
|
||
|
||
embed.add_field(
|
||
name="😄 Humor Preferences", value=humor_text, inline=False
|
||
)
|
||
|
||
# Communication style
|
||
comm_style = personality_data.get("communication_style", {})
|
||
if comm_style:
|
||
style_text = ""
|
||
for style, score in comm_style.items():
|
||
if score > 0.3:
|
||
percentage = int(score * 100)
|
||
style_text += f"• **{style.title()}**: {percentage}%\n"
|
||
|
||
if style_text:
|
||
embed.add_field(
|
||
name="💬 Communication Style", value=style_text, inline=True
|
||
)
|
||
|
||
# Activity patterns
|
||
activity = personality_data.get("activity_periods", [])
|
||
if activity:
|
||
# Analyze most active hours
|
||
hour_counts = {}
|
||
for period in activity[-20:]: # Last 20 activities
|
||
hour = period.get("hour", 0)
|
||
hour_counts[hour] = hour_counts.get(hour, 0) + 1
|
||
|
||
if hour_counts:
|
||
most_active = max(hour_counts.items(), key=lambda x: x[1])
|
||
embed.add_field(
|
||
name="⏰ Most Active Time",
|
||
value=f"{most_active[0]:02d}:00 - {most_active[0]+1:02d}:00",
|
||
inline=True,
|
||
)
|
||
|
||
# Topic interests
|
||
interests = personality_data.get("topic_interests", [])
|
||
if interests:
|
||
embed.add_field(
|
||
name="🔍 Topics of Interest",
|
||
value=", ".join(interests[:8]),
|
||
inline=False,
|
||
)
|
||
|
||
embed.set_footer(
|
||
text="Profile updates automatically based on your conversations"
|
||
)
|
||
|
||
return embed
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating personality embed: {e}")
|
||
return EmbedBuilder.error(
|
||
"Error", "Failed to create personality profile display"
|
||
)
|
||
|
||
@staticmethod
|
||
def error_embed(
|
||
title: str, description: str, error_type: str = "error"
|
||
) -> discord.Embed:
|
||
"""Build embed for error messages"""
|
||
colors = {"error": 0xFF0000, "warning": 0xFF9900, "info": 0x0099FF}
|
||
|
||
embed = discord.Embed(
|
||
title=f"{'❌' if error_type == 'error' else '⚠️' if error_type == 'warning' else 'ℹ️'} {title}",
|
||
description=description,
|
||
color=colors.get(error_type, 0xFF0000),
|
||
)
|
||
|
||
return embed
|
||
|
||
@staticmethod
|
||
def success_embed(title: str, description: str) -> discord.Embed:
|
||
"""Build embed for success messages"""
|
||
embed = discord.Embed(
|
||
title=f"✅ {title}", description=description, color=0x00FF00
|
||
)
|
||
|
||
return embed
|
||
|
||
@staticmethod
|
||
def info(title: str, description: str) -> discord.Embed:
|
||
"""Build embed for informational messages"""
|
||
return EmbedBuilder.error_embed(title, description, error_type="info")
|
||
|
||
@staticmethod
|
||
def error(title: str, description: str) -> discord.Embed:
|
||
"""Build embed for error messages"""
|
||
return EmbedBuilder.error_embed(title, description, error_type="error")
|
||
|
||
@staticmethod
|
||
def success(title: str, description: str) -> discord.Embed:
|
||
"""Build embed for success messages"""
|
||
return EmbedBuilder.success_embed(title, description)
|
||
|
||
@staticmethod
|
||
def warning(title: str, description: str) -> discord.Embed:
|
||
"""Build embed for warning messages"""
|
||
return EmbedBuilder.error_embed(title, description, error_type="warning")
|
||
|
||
@staticmethod
|
||
def _get_score_color(score: float) -> int:
|
||
"""Get color based on quote score"""
|
||
if score >= 8:
|
||
return 0xFF6B6B # Red for high scores
|
||
elif score >= 6:
|
||
return 0xFF9F43 # Orange for medium-high scores
|
||
elif score >= 4:
|
||
return 0xFFD93D # Yellow for medium scores
|
||
elif score >= 2:
|
||
return 0x6BCF7F # Green for low scores
|
||
else:
|
||
return 0x4DABF7 # Blue for very low scores
|
||
|
||
|
||
class ConsentView(ui.View):
|
||
"""Interactive consent collection view with buttons"""
|
||
|
||
def __init__(
|
||
self,
|
||
consent_manager: "ConsentManager",
|
||
guild_id: int,
|
||
timeout: int = 300,
|
||
on_consent_granted: Callable[[int, int], Awaitable[None]] | None = None,
|
||
):
|
||
super().__init__(timeout=timeout)
|
||
self.consent_manager = consent_manager
|
||
self.guild_id = guild_id
|
||
self.responses: dict[int, str] = {} # Track user responses
|
||
self.on_consent_granted = on_consent_granted # Callback when consent is granted
|
||
|
||
@ui.button(label="Give Consent", style=discord.ButtonStyle.green, emoji="✅")
|
||
async def give_consent(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Handle consent approval"""
|
||
# Immediately acknowledge the interaction to prevent timeout
|
||
await interaction.response.defer(ephemeral=True)
|
||
|
||
try:
|
||
user_id = interaction.user.id
|
||
|
||
# Check if user has already responded
|
||
if user_id in self.responses:
|
||
await interaction.followup.send(
|
||
"You've already responded to this consent request.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
# Check if user has global opt-out first
|
||
if user_id in self.consent_manager.global_opt_outs:
|
||
await interaction.followup.send(
|
||
ConsentMessages.GLOBAL_OPT_OUT, ephemeral=True
|
||
)
|
||
return
|
||
|
||
# Grant consent with timeout protection
|
||
try:
|
||
success = await asyncio.wait_for(
|
||
self.consent_manager.grant_consent(user_id, self.guild_id),
|
||
timeout=10.0,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.error(
|
||
f"Timeout granting consent for user {user_id} in guild {self.guild_id}"
|
||
)
|
||
await interaction.followup.send(
|
||
"❌ Request timed out. Please try again later.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
if success:
|
||
self.responses[user_id] = "granted"
|
||
await interaction.followup.send(
|
||
ConsentMessages.CONSENT_GRANTED, ephemeral=True
|
||
)
|
||
logger.info(
|
||
f"Consent granted by user {user_id} in guild {self.guild_id}"
|
||
)
|
||
|
||
# Trigger callback if provided
|
||
if self.on_consent_granted:
|
||
try:
|
||
await self.on_consent_granted(user_id, self.guild_id)
|
||
except Exception as callback_error:
|
||
logger.error(
|
||
f"Error in consent granted callback: {callback_error}"
|
||
)
|
||
else:
|
||
await interaction.followup.send(
|
||
"❌ Failed to grant consent. Please try again or contact an administrator.",
|
||
ephemeral=True,
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error handling consent approval: {e}")
|
||
try:
|
||
await interaction.followup.send(
|
||
"❌ An error occurred while processing your consent.",
|
||
ephemeral=True,
|
||
)
|
||
except Exception as followup_error:
|
||
logger.error(f"Failed to send error message: {followup_error}")
|
||
|
||
@ui.button(label="Learn More", style=discord.ButtonStyle.gray, emoji="ℹ️")
|
||
async def learn_more(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Show detailed privacy information"""
|
||
try:
|
||
embed = ConsentTemplates.get_privacy_info_embed()
|
||
await interaction.response.send_message(embed=embed, ephemeral=True)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error showing privacy info: {e}")
|
||
try:
|
||
await interaction.response.send_message(
|
||
"❌ Failed to load privacy information.", ephemeral=True
|
||
)
|
||
except Exception:
|
||
# If we can't respond, try followup
|
||
try:
|
||
await interaction.followup.send(
|
||
"❌ Failed to load privacy information.", ephemeral=True
|
||
)
|
||
except Exception as final_error:
|
||
logger.error(
|
||
f"Failed all attempts to send error message: {final_error}"
|
||
)
|
||
|
||
@ui.button(label="Decline", style=discord.ButtonStyle.red, emoji="❌")
|
||
async def decline_consent(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
):
|
||
"""Handle consent decline"""
|
||
try:
|
||
user_id = interaction.user.id
|
||
|
||
if user_id in self.responses:
|
||
await interaction.response.send_message(
|
||
"You've already responded to this consent request.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
self.responses[user_id] = "declined"
|
||
await interaction.response.send_message(
|
||
"🚫 **Consent declined.** You will not be included in voice recordings.\n\n"
|
||
"You can change your mind anytime using `/give_consent`.",
|
||
ephemeral=True,
|
||
)
|
||
|
||
logger.info(f"Consent declined by user {user_id} in guild {self.guild_id}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error handling consent decline: {e}")
|
||
try:
|
||
await interaction.response.send_message(
|
||
"❌ An error occurred while processing your response.",
|
||
ephemeral=True,
|
||
)
|
||
except Exception:
|
||
try:
|
||
await interaction.followup.send(
|
||
"❌ An error occurred while processing your response.",
|
||
ephemeral=True,
|
||
)
|
||
except Exception as final_error:
|
||
logger.error(
|
||
f"Failed all attempts to send error message: {final_error}"
|
||
)
|
||
|
||
async def on_timeout(self):
|
||
"""Handle view timeout"""
|
||
try:
|
||
# Disable all buttons
|
||
for item in self.children:
|
||
item.disabled = True
|
||
|
||
# Update the original message if possible
|
||
if hasattr(self, "message") and self.message:
|
||
timeout_embed = discord.Embed(
|
||
title="⏰ Consent Request Expired",
|
||
description=ConsentMessages.CONSENT_TIMEOUT,
|
||
color=0xFF9900,
|
||
)
|
||
await self.message.edit(embed=timeout_embed, view=self)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error handling consent view timeout: {e}")
|
||
|
||
|
||
class QuoteBrowserView(ui.View):
|
||
"""
|
||
Interactive view for browsing quotes with pagination and filtering
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
db_manager: "DatabaseManager",
|
||
user_id: int,
|
||
guild_id: int,
|
||
quotes: list[dict[str, Any]],
|
||
page_size: int = 5,
|
||
) -> None:
|
||
super().__init__(timeout=600) # 10 minute timeout
|
||
self.db_manager = db_manager
|
||
self.user_id = user_id
|
||
self.guild_id = guild_id
|
||
self.quotes = quotes
|
||
self.page_size = page_size
|
||
self.current_page = 0
|
||
self.total_pages = max(1, (len(quotes) + page_size - 1) // page_size)
|
||
|
||
# Update button states
|
||
self._update_buttons()
|
||
|
||
def _update_buttons(self) -> None:
|
||
"""Update button states based on current page"""
|
||
self.previous_page.disabled = self.current_page == 0
|
||
self.next_page.disabled = self.current_page >= self.total_pages - 1
|
||
|
||
def _get_current_page_quotes(self) -> list[dict[str, Any]]:
|
||
"""Get quotes for current page"""
|
||
start_idx = self.current_page * self.page_size
|
||
end_idx = start_idx + self.page_size
|
||
return self.quotes[start_idx:end_idx]
|
||
|
||
def _create_page_embed(self) -> discord.Embed:
|
||
"""Create embed for current page"""
|
||
page_quotes = self._get_current_page_quotes()
|
||
|
||
embed = discord.Embed(
|
||
title=f"📝 Your Quotes (Page {self.current_page + 1}/{self.total_pages})",
|
||
description=f"Showing {len(page_quotes)} of {len(self.quotes)} quotes",
|
||
color=0x3498DB,
|
||
)
|
||
|
||
for i, quote in enumerate(page_quotes, 1):
|
||
# Build score display
|
||
scores = []
|
||
if quote["funny_score"] > 5:
|
||
scores.append(f"😂 {quote['funny_score']:.1f}")
|
||
if quote["dark_score"] > 5:
|
||
scores.append(f"🖤 {quote['dark_score']:.1f}")
|
||
if quote["silly_score"] > 5:
|
||
scores.append(f"🤪 {quote['silly_score']:.1f}")
|
||
if quote["suspicious_score"] > 5:
|
||
scores.append(f"🤔 {quote['suspicious_score']:.1f}")
|
||
if quote["asinine_score"] > 5:
|
||
scores.append(f"🙄 {quote['asinine_score']:.1f}")
|
||
|
||
score_text = " | ".join(scores) if scores else "No significant scores"
|
||
|
||
# Truncate long quotes
|
||
quote_text = quote["quote"]
|
||
if len(quote_text) > 150:
|
||
quote_text = quote_text[:150] + "..."
|
||
|
||
timestamp = quote["timestamp"].strftime("%Y-%m-%d %H:%M")
|
||
|
||
embed.add_field(
|
||
name=f"Quote #{self.current_page * self.page_size + i} (Score: {quote['overall_score']:.1f})",
|
||
value=f'*"{quote_text}"*\n\n**Scores:** {score_text}\n**Date:** {timestamp}',
|
||
inline=False,
|
||
)
|
||
|
||
return embed
|
||
|
||
@ui.button(label="Previous", style=discord.ButtonStyle.secondary, emoji="⬅️")
|
||
async def previous_page(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
) -> None:
|
||
"""Go to previous page"""
|
||
try:
|
||
if interaction.user.id != self.user_id:
|
||
await interaction.response.send_message(
|
||
"You can only browse your own quotes.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
if self.current_page > 0:
|
||
self.current_page -= 1
|
||
self._update_buttons()
|
||
|
||
embed = self._create_page_embed()
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
else:
|
||
await interaction.response.send_message(
|
||
"You're already on the first page.", ephemeral=True
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error in previous page button: {e}")
|
||
await interaction.response.send_message(
|
||
"An error occurred while navigating.", ephemeral=True
|
||
)
|
||
|
||
@ui.button(label="Next", style=discord.ButtonStyle.secondary, emoji="➡️")
|
||
async def next_page(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
) -> None:
|
||
"""Go to next page"""
|
||
try:
|
||
if interaction.user.id != self.user_id:
|
||
await interaction.response.send_message(
|
||
"You can only browse your own quotes.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
if self.current_page < self.total_pages - 1:
|
||
self.current_page += 1
|
||
self._update_buttons()
|
||
|
||
embed = self._create_page_embed()
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
else:
|
||
await interaction.response.send_message(
|
||
"You're already on the last page.", ephemeral=True
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error in next page button: {e}")
|
||
await interaction.response.send_message(
|
||
"An error occurred while navigating.", ephemeral=True
|
||
)
|
||
|
||
@ui.select(
|
||
placeholder="Filter by category...",
|
||
options=[
|
||
discord.SelectOption(label="All Categories", value="all", emoji="📝"),
|
||
discord.SelectOption(label="Funny Quotes", value="funny", emoji="😂"),
|
||
discord.SelectOption(label="Dark Humor", value="dark", emoji="🖤"),
|
||
discord.SelectOption(label="Silly Quotes", value="silly", emoji="🤪"),
|
||
discord.SelectOption(label="Suspicious", value="suspicious", emoji="🤔"),
|
||
discord.SelectOption(label="Asinine", value="asinine", emoji="🙄"),
|
||
],
|
||
)
|
||
async def category_filter(
|
||
self, interaction: discord.Interaction, select: ui.Select
|
||
) -> None:
|
||
"""Handle category filtering"""
|
||
try:
|
||
if interaction.user.id != self.user_id:
|
||
await interaction.response.send_message(
|
||
"You can only filter your own quotes.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
category = select.values[0]
|
||
|
||
# Fetch filtered quotes
|
||
if category == "all":
|
||
query = """
|
||
SELECT quote, timestamp, funny_score, dark_score, silly_score,
|
||
suspicious_score, asinine_score, overall_score
|
||
FROM quotes
|
||
WHERE user_id = $1 AND guild_id = $2
|
||
ORDER BY overall_score DESC, timestamp DESC
|
||
"""
|
||
params = [self.user_id, self.guild_id]
|
||
else:
|
||
query = f"""
|
||
SELECT quote, timestamp, funny_score, dark_score, silly_score,
|
||
suspicious_score, asinine_score, overall_score
|
||
FROM quotes
|
||
WHERE user_id = $1 AND guild_id = $2 AND {category}_score > 5.0
|
||
ORDER BY {category}_score DESC, timestamp DESC
|
||
"""
|
||
params = [self.user_id, self.guild_id]
|
||
|
||
filtered_quotes = await self.db_manager.execute_query(
|
||
query, *params, fetch_all=True
|
||
)
|
||
|
||
# Update view with filtered results
|
||
self.quotes = filtered_quotes
|
||
self.current_page = 0
|
||
self.total_pages = max(
|
||
1, (len(filtered_quotes) + self.page_size - 1) // self.page_size
|
||
)
|
||
self._update_buttons()
|
||
|
||
embed = self._create_page_embed()
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error in category filter: {e}")
|
||
await interaction.response.send_message(
|
||
"An error occurred while filtering quotes.", ephemeral=True
|
||
)
|
||
|
||
|
||
class QuoteAnalysisModal(ui.Modal):
|
||
"""
|
||
Modal for requesting detailed quote analysis
|
||
"""
|
||
|
||
def __init__(self, quote_analyzer: QuoteAnalyzer) -> None:
|
||
super().__init__(title="Quote Analysis Request")
|
||
self.quote_analyzer = quote_analyzer
|
||
|
||
quote_text = ui.TextInput(
|
||
label="Quote to Analyze",
|
||
placeholder="Enter the quote you'd like analyzed...",
|
||
style=discord.TextStyle.paragraph,
|
||
max_length=1000,
|
||
required=True,
|
||
)
|
||
|
||
context = ui.TextInput(
|
||
label="Context (Optional)",
|
||
placeholder="Provide context about when/where this was said...",
|
||
style=discord.TextStyle.paragraph,
|
||
max_length=500,
|
||
required=False,
|
||
)
|
||
|
||
async def on_submit(self, interaction: discord.Interaction) -> None:
|
||
"""Handle modal submission"""
|
||
try:
|
||
await interaction.response.defer(ephemeral=True)
|
||
|
||
quote = self.quote_text.value
|
||
context_info = self.context.value or "No context provided"
|
||
|
||
# Perform quote analysis (this would integrate with the quote analyzer service)
|
||
embed = discord.Embed(
|
||
title="🔍 Quote Analysis", description=f'*"{quote}"*', color=0x3498DB
|
||
)
|
||
|
||
embed.add_field(
|
||
name="📊 Analysis Status",
|
||
value="Quote analysis would be performed here using the QuoteAnalyzer service.",
|
||
inline=False,
|
||
)
|
||
|
||
embed.add_field(
|
||
name="🤖 AI Processing",
|
||
value="The quote would be analyzed for humor categories, context, and scoring.",
|
||
inline=False,
|
||
)
|
||
|
||
embed.add_field(name="Context Provided", value=context_info, inline=False)
|
||
|
||
await interaction.followup.send(embed=embed)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error in quote analysis modal: {e}")
|
||
await interaction.followup.send(
|
||
"An error occurred while analyzing the quote."
|
||
)
|
||
|
||
|
||
class FeedbackView(ui.View):
|
||
"""
|
||
Interactive view for collecting user feedback on quotes
|
||
"""
|
||
|
||
def __init__(self, quote_id: int, db_manager: "DatabaseManager") -> None:
|
||
super().__init__(timeout=300)
|
||
self.quote_id = quote_id
|
||
self.db_manager = db_manager
|
||
|
||
@ui.button(label="👍", style=discord.ButtonStyle.success)
|
||
async def positive_feedback(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
) -> None:
|
||
"""Handle positive feedback"""
|
||
await self._handle_feedback(interaction, "positive", "👍")
|
||
|
||
@ui.button(label="👎", style=discord.ButtonStyle.danger)
|
||
async def negative_feedback(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
) -> None:
|
||
"""Handle negative feedback"""
|
||
await self._handle_feedback(interaction, "negative", "👎")
|
||
|
||
@ui.button(label="😂", style=discord.ButtonStyle.secondary)
|
||
async def funny_feedback(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
) -> None:
|
||
"""Handle funny feedback"""
|
||
await self._handle_feedback(interaction, "funny", "😂")
|
||
|
||
@ui.button(label="🤔", style=discord.ButtonStyle.secondary)
|
||
async def confused_feedback(
|
||
self, interaction: discord.Interaction, button: ui.Button
|
||
) -> None:
|
||
"""Handle confused feedback"""
|
||
await self._handle_feedback(interaction, "confused", "🤔")
|
||
|
||
async def _handle_feedback(
|
||
self, interaction: discord.Interaction, feedback_type: str, emoji: str
|
||
):
|
||
"""Handle feedback submission"""
|
||
try:
|
||
# Store feedback in database
|
||
await self.db_manager.execute_query(
|
||
"""
|
||
INSERT INTO quote_feedback
|
||
(quote_id, user_id, feedback_type, feedback_value, timestamp)
|
||
VALUES ($1, $2, $3, $4, NOW())
|
||
ON CONFLICT (quote_id, user_id)
|
||
DO UPDATE SET
|
||
feedback_type = EXCLUDED.feedback_type,
|
||
feedback_value = EXCLUDED.feedback_value,
|
||
timestamp = EXCLUDED.timestamp
|
||
""",
|
||
self.quote_id,
|
||
interaction.user.id,
|
||
feedback_type,
|
||
emoji,
|
||
)
|
||
|
||
# Disable all buttons after feedback
|
||
for child in self.children:
|
||
child.disabled = True
|
||
|
||
embed = discord.Embed(
|
||
title="✅ Feedback Recorded",
|
||
description=f"Thank you for your feedback! {emoji}",
|
||
color=0x00FF00,
|
||
)
|
||
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error handling feedback: {e}")
|
||
await interaction.response.send_message(
|
||
"An error occurred while recording feedback.", ephemeral=True
|
||
)
|
||
|
||
|
||
class SpeakerTaggingView(ui.View):
|
||
"""Interactive speaker tagging view for unknown quotes"""
|
||
|
||
def __init__(
|
||
self,
|
||
quote_id: int,
|
||
voice_members: list[discord.Member],
|
||
db_manager: DatabaseManager,
|
||
timeout: int = 300,
|
||
):
|
||
super().__init__(timeout=timeout)
|
||
self.quote_id = quote_id
|
||
self.db_manager = db_manager
|
||
self.tagged = False
|
||
|
||
# Add buttons for each voice channel member
|
||
for i, member in enumerate(voice_members[:5]): # Limit to 5 buttons
|
||
button = ui.Button(
|
||
label=f"Tag {member.display_name}",
|
||
style=discord.ButtonStyle.primary,
|
||
custom_id=f"tag_{member.id}",
|
||
row=i // 5, # Arrange in rows
|
||
)
|
||
button.callback = self._create_tag_callback(member.id, member.display_name)
|
||
self.add_item(button)
|
||
|
||
# Add "Unknown Speaker" option
|
||
unknown_button = ui.Button(
|
||
label="Keep as Unknown",
|
||
style=discord.ButtonStyle.gray,
|
||
custom_id="unknown",
|
||
row=1,
|
||
)
|
||
unknown_button.callback = self._keep_unknown_callback
|
||
self.add_item(unknown_button)
|
||
|
||
def _create_tag_callback(
|
||
self, user_id: int, display_name: str
|
||
) -> Callable[[discord.Interaction], Awaitable[None]]:
|
||
"""Create callback for tagging a specific user."""
|
||
|
||
async def tag_callback(interaction: discord.Interaction) -> None:
|
||
if self.tagged:
|
||
await interaction.response.send_message(
|
||
"This quote has already been tagged.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
try:
|
||
# Update quote with speaker information
|
||
success = await self.db_manager.update_quote_speaker(
|
||
self.quote_id, user_id, interaction.user.id
|
||
)
|
||
|
||
if success:
|
||
self.tagged = True
|
||
|
||
# Disable all buttons
|
||
for item in self.children:
|
||
item.disabled = True
|
||
|
||
embed = discord.Embed(
|
||
title="✅ Speaker Tagged",
|
||
description=f"Quote tagged as **{display_name}**. This helps improve speaker recognition!",
|
||
color=0x00FF00,
|
||
)
|
||
|
||
embed.add_field(
|
||
name="Thank You!",
|
||
value="Your feedback helps train the speaker recognition system.",
|
||
inline=False,
|
||
)
|
||
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
logger.info(
|
||
f"Quote {self.quote_id} tagged as user {user_id} by {interaction.user.id}"
|
||
)
|
||
else:
|
||
await interaction.response.send_message(
|
||
"❌ Failed to tag speaker. Please try again.", ephemeral=True
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error tagging speaker: {e}")
|
||
await interaction.response.send_message(
|
||
"❌ An error occurred while tagging the speaker.", ephemeral=True
|
||
)
|
||
|
||
return tag_callback
|
||
|
||
async def _keep_unknown_callback(self, interaction: discord.Interaction) -> None:
|
||
"""Callback for keeping speaker as unknown."""
|
||
if self.tagged:
|
||
await interaction.response.send_message(
|
||
"This quote has already been processed.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
try:
|
||
self.tagged = True
|
||
|
||
# Disable all buttons
|
||
for item in self.children:
|
||
item.disabled = True
|
||
|
||
embed = discord.Embed(
|
||
title="👤 Speaker Kept as Unknown",
|
||
description="The quote will remain labeled with the speaker number.",
|
||
color=0x999999,
|
||
)
|
||
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
logger.info(
|
||
f"Quote {self.quote_id} kept as unknown by {interaction.user.id}"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error keeping speaker unknown: {e}")
|
||
await interaction.response.send_message(
|
||
"❌ An error occurred while processing your choice.", ephemeral=True
|
||
)
|
||
|
||
|
||
class DataDeletionView(ui.View):
|
||
"""Confirmation view for data deletion"""
|
||
|
||
def __init__(
|
||
self,
|
||
user_id: int,
|
||
guild_id: int,
|
||
quote_count: int,
|
||
consent_manager: "ConsentManager",
|
||
timeout: int = 60,
|
||
):
|
||
super().__init__(timeout=timeout)
|
||
self.user_id = user_id
|
||
self.guild_id = guild_id
|
||
self.quote_count = quote_count
|
||
self.consent_manager = consent_manager
|
||
self.confirmed = False
|
||
|
||
@ui.button(label="Confirm Delete", style=discord.ButtonStyle.danger, emoji="🗑️")
|
||
async def confirm_delete(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Confirm and execute data deletion"""
|
||
if interaction.user.id != self.user_id:
|
||
await interaction.response.send_message(
|
||
"❌ You can only delete your own data.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
if self.confirmed:
|
||
await interaction.response.send_message(
|
||
"Data deletion has already been processed.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
try:
|
||
self.confirmed = True
|
||
|
||
# Execute data deletion
|
||
deletion_counts = await self.consent_manager.delete_user_data(
|
||
self.user_id, self.guild_id
|
||
)
|
||
|
||
# Disable buttons
|
||
for item in self.children:
|
||
item.disabled = True
|
||
|
||
if "error" not in deletion_counts:
|
||
embed = discord.Embed(
|
||
title="✅ Data Deleted Successfully",
|
||
description="Your data has been permanently removed from this server.",
|
||
color=0x00FF00,
|
||
)
|
||
|
||
embed.add_field(
|
||
name="Deleted Items",
|
||
value=f"• **{deletion_counts.get('quotes', 0)}** quotes\n"
|
||
f"• **{deletion_counts.get('consent_records', 0)}** consent records\n"
|
||
f"• **{deletion_counts.get('feedback_records', 0)}** feedback records\n"
|
||
f"• **{deletion_counts.get('speaker_profile', 0)}** speaker profiles",
|
||
inline=False,
|
||
)
|
||
|
||
embed.add_field(
|
||
name="What's Next?",
|
||
value="You can still use the bot normally. Give consent again anytime with `/give_consent`.",
|
||
inline=False,
|
||
)
|
||
else:
|
||
embed = discord.Embed(
|
||
title="❌ Deletion Failed",
|
||
description=f"An error occurred: {deletion_counts['error']}",
|
||
color=0xFF0000,
|
||
)
|
||
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error confirming data deletion: {e}")
|
||
await interaction.response.send_message(
|
||
"❌ An error occurred during data deletion.", ephemeral=True
|
||
)
|
||
|
||
@ui.button(label="Cancel", style=discord.ButtonStyle.gray, emoji="❌")
|
||
async def cancel_delete(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Cancel data deletion"""
|
||
if interaction.user.id != self.user_id:
|
||
await interaction.response.send_message(
|
||
"❌ This action is not for you.", ephemeral=True
|
||
)
|
||
return
|
||
|
||
try:
|
||
# Disable buttons
|
||
for item in self.children:
|
||
item.disabled = True
|
||
|
||
embed = discord.Embed(
|
||
title="🛡️ Data Deletion Cancelled",
|
||
description="Your data remains safe and unchanged.",
|
||
color=0x0099FF,
|
||
)
|
||
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error cancelling data deletion: {e}")
|
||
await interaction.response.send_message(
|
||
"❌ An error occurred while cancelling.", ephemeral=True
|
||
)
|
||
|
||
|
||
class HelpNavigationView(ui.View):
|
||
"""Interactive help navigation view"""
|
||
|
||
def __init__(self, timeout: int = 300):
|
||
super().__init__(timeout=timeout)
|
||
|
||
@ui.select(
|
||
placeholder="Choose a help topic...",
|
||
options=[
|
||
discord.SelectOption(
|
||
label="Getting Started", value="getting_started", emoji="🏁"
|
||
),
|
||
discord.SelectOption(
|
||
label="Privacy & Consent", value="privacy", emoji="🔒"
|
||
),
|
||
discord.SelectOption(
|
||
label="Speaker Recognition", value="speaker_rec", emoji="🎙️"
|
||
),
|
||
discord.SelectOption(label="Quote Scoring", value="scoring", emoji="📊"),
|
||
discord.SelectOption(
|
||
label="Commands Reference", value="commands", emoji="📝"
|
||
),
|
||
discord.SelectOption(
|
||
label="Troubleshooting", value="troubleshoot", emoji="🔧"
|
||
),
|
||
],
|
||
)
|
||
async def help_select(self, interaction: discord.Interaction, select: ui.Select):
|
||
"""Handle help topic selection"""
|
||
try:
|
||
topic = select.values[0]
|
||
embed = await self._get_help_embed(topic)
|
||
await interaction.response.edit_message(embed=embed, view=self)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error showing help topic: {e}")
|
||
await interaction.response.send_message(
|
||
"❌ Failed to load help information.", ephemeral=True
|
||
)
|
||
|
||
async def _get_help_embed(self, topic: str) -> discord.Embed:
|
||
"""Generate help embed for specific topic"""
|
||
help_content = {
|
||
"getting_started": {
|
||
"title": "🏁 Getting Started",
|
||
"description": "Learn how to use Quote Bot for the first time.",
|
||
"fields": [
|
||
{
|
||
"name": "📋 Setup Steps",
|
||
"value": (
|
||
"1. Join a voice channel\n"
|
||
"2. Use `/start_recording` to begin\n"
|
||
"3. Give consent when prompted\n"
|
||
"4. Chat normally - the bot will detect memorable quotes!"
|
||
),
|
||
},
|
||
{
|
||
"name": "🎯 Key Commands",
|
||
"value": (
|
||
"`/give_consent` - Allow recording\n"
|
||
"`/random_quote` - Get a random quote\n"
|
||
"`/my_quotes` - View your quotes\n"
|
||
"`/help` - Get help"
|
||
),
|
||
},
|
||
],
|
||
},
|
||
"privacy": {
|
||
"title": "🔒 Privacy & Consent",
|
||
"description": "Understanding how your data is protected.",
|
||
"fields": [
|
||
{
|
||
"name": "🛡️ Privacy Controls",
|
||
"value": (
|
||
"`/give_consent` - Allow recording\n"
|
||
"`/revoke_consent` - Stop recording (this server)\n"
|
||
"`/opt_out` - Stop recording (all servers)\n"
|
||
"`/delete_my_quotes` - Remove your data"
|
||
),
|
||
},
|
||
{
|
||
"name": "📊 What We Store",
|
||
"value": (
|
||
"• Text quotes from conversations\n"
|
||
"• Humor analysis scores\n"
|
||
"• Discord username\n"
|
||
"• NO permanent audio files"
|
||
),
|
||
},
|
||
],
|
||
},
|
||
# Add more help topics as needed
|
||
}
|
||
|
||
content = help_content.get(
|
||
topic,
|
||
{
|
||
"title": "❓ Help Topic",
|
||
"description": "Help content not available for this topic.",
|
||
"fields": [],
|
||
},
|
||
)
|
||
|
||
embed = discord.Embed(
|
||
title=content["title"], description=content["description"], color=0x0099FF
|
||
)
|
||
|
||
for field in content.get("fields", []):
|
||
embed.add_field(name=field["name"], value=field["value"], inline=False)
|
||
|
||
embed.set_footer(text="Use the dropdown above to explore other topics")
|
||
return embed
|
||
|
||
|
||
class PaginatedView(ui.View):
|
||
"""Generic paginated view for long content"""
|
||
|
||
def __init__(self, pages: list[discord.Embed], timeout: int = 300):
|
||
super().__init__(timeout=timeout)
|
||
self.pages = pages
|
||
self.current_page = 0
|
||
self.max_page = len(pages) - 1
|
||
|
||
# Disable buttons if only one page
|
||
if len(pages) <= 1:
|
||
for item in self.children:
|
||
item.disabled = True
|
||
|
||
@ui.button(emoji="⬅️", style=discord.ButtonStyle.gray)
|
||
async def previous_page(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Go to previous page"""
|
||
if self.current_page > 0:
|
||
self.current_page -= 1
|
||
await interaction.response.edit_message(
|
||
embed=self.pages[self.current_page], view=self
|
||
)
|
||
else:
|
||
await interaction.response.defer()
|
||
|
||
@ui.button(emoji="➡️", style=discord.ButtonStyle.gray)
|
||
async def next_page(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Go to next page"""
|
||
if self.current_page < self.max_page:
|
||
self.current_page += 1
|
||
await interaction.response.edit_message(
|
||
embed=self.pages[self.current_page], view=self
|
||
)
|
||
else:
|
||
await interaction.response.defer()
|
||
|
||
@ui.button(emoji="🔢", style=discord.ButtonStyle.gray)
|
||
async def page_info(self, interaction: discord.Interaction, button: ui.Button):
|
||
"""Show page information"""
|
||
await interaction.response.send_message(
|
||
f"Page {self.current_page + 1} of {self.max_page + 1}", ephemeral=True
|
||
)
|
||
|
||
|
||
class UIComponentManager:
|
||
"""
|
||
Manager class for creating and handling Discord UI components
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
bot: commands.Bot,
|
||
db_manager: "DatabaseManager",
|
||
consent_manager: "ConsentManager",
|
||
memory_manager: "MemoryManager",
|
||
quote_analyzer: QuoteAnalyzer,
|
||
):
|
||
self.bot = bot
|
||
self.db_manager = db_manager
|
||
self.consent_manager = consent_manager
|
||
self.memory_manager = memory_manager
|
||
self.quote_analyzer = quote_analyzer
|
||
|
||
async def create_consent_interface(
|
||
self, user_id: int, guild_id: int
|
||
) -> tuple[discord.Embed | None, ConsentView | None]:
|
||
"""Create consent management interface"""
|
||
try:
|
||
# Check current status
|
||
has_consent = await self.consent_manager.check_consent(user_id, guild_id)
|
||
|
||
if has_consent:
|
||
embed = discord.Embed(
|
||
title="✅ Consent Status",
|
||
description="You have granted consent for voice recording.",
|
||
color=0x00FF00,
|
||
)
|
||
embed.add_field(
|
||
name="Current Status",
|
||
value="Your voice is being recorded and may be analyzed for quotes.",
|
||
inline=False,
|
||
)
|
||
else:
|
||
embed = discord.Embed(
|
||
title="📋 Consent Management",
|
||
description="Manage your voice recording consent preferences.",
|
||
color=0x3498DB,
|
||
)
|
||
embed.add_field(
|
||
name="Privacy Notice",
|
||
value="• Recording requires explicit consent\n"
|
||
"• You can revoke consent at any time\n"
|
||
"• Data is stored securely and can be deleted",
|
||
inline=False,
|
||
)
|
||
|
||
view = ConsentView(self.consent_manager, user_id, guild_id)
|
||
return embed, view
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating consent interface: {e}")
|
||
return None, None
|
||
|
||
async def create_quote_browser(
|
||
self, user_id: int, guild_id: int
|
||
) -> tuple[discord.Embed | None, "QuoteBrowserView | None"]:
|
||
"""Create quote browsing interface"""
|
||
try:
|
||
# Fetch user's quotes
|
||
quotes = await self.db_manager.execute_query(
|
||
"""
|
||
SELECT quote, timestamp, funny_score, dark_score, silly_score,
|
||
suspicious_score, asinine_score, overall_score
|
||
FROM quotes
|
||
WHERE user_id = $1 AND guild_id = $2
|
||
ORDER BY overall_score DESC, timestamp DESC
|
||
LIMIT 50
|
||
""",
|
||
user_id,
|
||
guild_id,
|
||
fetch_all=True,
|
||
)
|
||
|
||
if not quotes:
|
||
embed = discord.Embed(
|
||
title="📝 No Quotes Found",
|
||
description="You don't have any recorded quotes yet.",
|
||
color=0x888888,
|
||
)
|
||
return embed, None
|
||
|
||
view = QuoteBrowserView(self.db_manager, user_id, guild_id, quotes)
|
||
embed = view._create_page_embed()
|
||
|
||
return embed, view
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating quote browser: {e}")
|
||
return None, None
|
||
|
||
async def create_quote_display_with_feedback(
|
||
self, quote_data: dict[str, Any]
|
||
) -> tuple[discord.Embed | None, FeedbackView | None]:
|
||
"""Create quote display with feedback collection"""
|
||
try:
|
||
embed = EmbedBuilder.create_quote_embed(quote_data)
|
||
view = FeedbackView(quote_data["id"], self.db_manager)
|
||
|
||
return embed, view
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating quote display with feedback: {e}")
|
||
return None, None
|
||
|
||
def create_quote_analysis_modal(self) -> QuoteAnalysisModal:
|
||
"""Create modal for quote analysis requests"""
|
||
return QuoteAnalysisModal(self.quote_analyzer)
|
||
|
||
async def create_personality_display(self, user_id: int) -> discord.Embed:
|
||
"""Create personality profile display"""
|
||
try:
|
||
if not self.memory_manager:
|
||
return discord.Embed(
|
||
title="❌ Feature Unavailable",
|
||
description="Memory system is not available.",
|
||
color=0xFF0000,
|
||
)
|
||
|
||
profile = await self.memory_manager.get_personality_profile(user_id)
|
||
|
||
if not profile:
|
||
return discord.Embed(
|
||
title="🧠 No Personality Profile",
|
||
description="Not enough conversation data available yet.",
|
||
color=0x888888,
|
||
)
|
||
|
||
personality_data = {
|
||
"humor_preferences": profile.humor_preferences,
|
||
"communication_style": profile.communication_style,
|
||
"topic_interests": profile.topic_interests,
|
||
"activity_periods": profile.activity_periods,
|
||
"last_updated": profile.last_updated,
|
||
}
|
||
|
||
return EmbedBuilder.create_personality_embed(personality_data)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating personality display: {e}")
|
||
return discord.Embed(
|
||
title="❌ Error",
|
||
description="Failed to load personality profile.",
|
||
color=0xFF0000,
|
||
)
|