Files
disbord/cogs/quotes_cog.py
Travis Vasceannie 3acb779569 chore: remove .env.example and add new files for project structure
- 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.
2025-08-27 23:00:19 -04:00

736 lines
28 KiB
Python

"""
Quotes Cog for Discord Voice Chat Quote Bot
Handles quote management, search, analysis, and display functionality
with sophisticated AI integration and dimensional score analysis.
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
import discord
from discord import app_commands
from discord.ext import commands
from core.database import DatabaseManager
from services.quotes.quote_analyzer import QuoteAnalyzer
from services.quotes.quote_explanation import (ExplanationDepth,
QuoteExplanationService)
from ui.utils import (EmbedBuilder, EmbedStyles, StatusIndicators, UIFormatter,
ValidationHelper)
if TYPE_CHECKING:
from main import QuoteBot
logger = logging.getLogger(__name__)
class QuotesCog(commands.Cog):
"""
Quote management and AI-powered analysis operations.
Commands:
- /quotes - Search and display quotes with dimensional scores
- /quote_stats - Show comprehensive quote statistics
- /my_quotes - Show your quotes with analysis
- /top_quotes - Show highest-rated quotes
- /random_quote - Get a random quote with analysis
- /explain_quote - Get detailed AI explanation of quote analysis
- /legendary_quotes - Show quotes above realtime threshold (8.5+)
- /search_by_category - Search quotes by dimensional score categories
"""
# Quote score thresholds from CLAUDE.md
REALTIME_THRESHOLD: float = 8.5
ROTATION_THRESHOLD: float = 6.0
DAILY_THRESHOLD: float = 3.0
def __init__(self, bot: "QuoteBot") -> None:
self.bot = bot
# Validate required bot attributes
required_attrs = ["db_manager", "quote_analyzer"]
for attr in required_attrs:
if not hasattr(bot, attr) or not getattr(bot, attr):
raise RuntimeError(f"Bot {attr} is not initialized")
self.db_manager: DatabaseManager = bot.db_manager # type: ignore[assignment]
self.quote_analyzer: QuoteAnalyzer = bot.quote_analyzer # type: ignore[assignment]
# Initialize QuoteExplanationService
self.explanation_service: QuoteExplanationService | None = None
self._initialize_explanation_service()
def _initialize_explanation_service(self) -> None:
"""Initialize the quote explanation service."""
try:
if hasattr(self.bot, "ai_manager") and self.bot.ai_manager:
self.explanation_service = QuoteExplanationService(
self.bot, self.db_manager, self.bot.ai_manager
)
logger.info("QuoteExplanationService initialized successfully")
else:
logger.warning(
"AI manager not available, explanation features disabled"
)
except Exception as e:
logger.error(f"Failed to initialize QuoteExplanationService: {e}")
self.explanation_service = None
@app_commands.command(name="quotes", description="Search and display quotes")
@app_commands.describe(
search="Search term to find quotes",
user="Filter quotes by specific user",
limit="Number of quotes to display (1-10, default 5)",
)
async def quotes(
self,
interaction: discord.Interaction,
search: str | None = None,
user: discord.Member | None = None,
limit: int | None = 5,
) -> None:
"""Search and display quotes with filters"""
await interaction.response.defer()
try:
# Validate limit
limit = max(1, min(limit or 5, 10))
# Build search parameters
search_params = {
"guild_id": interaction.guild_id,
"search_term": search,
"user_id": user.id if user else None,
"limit": limit,
}
# Get quotes from database with dimensional scores
quotes = await self.db_manager.search_quotes(**search_params)
if not quotes:
embed = EmbedBuilder.create_info_embed(
"No Quotes Found", "No quotes match your search criteria."
)
await interaction.followup.send(embed=embed)
return
# Create enhanced embed with dimensional scores
embed = await self._create_quotes_embed(
"Quote Results", f"Found {len(quotes)} quote(s)", quotes
)
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in quotes command: {e}")
embed = EmbedBuilder.create_error_embed(
"Quote Search Error",
"Failed to retrieve quotes.",
details=str(e) if logger.isEnabledFor(logging.DEBUG) else None,
)
await interaction.followup.send(embed=embed, ephemeral=True)
async def _create_quotes_embed(
self, title: str, description: str, quotes: list[dict[str, Any]]
) -> discord.Embed:
"""Create enhanced embed with dimensional scores for quotes."""
# Determine embed color based on highest score in results
max_score = max(
(quote.get("overall_score", 0.0) for quote in quotes), default=0.0
)
if max_score >= self.REALTIME_THRESHOLD:
color = EmbedStyles.FUNNY # Gold for legendary
elif max_score >= self.ROTATION_THRESHOLD:
color = EmbedStyles.SUCCESS # Green for good
elif max_score >= self.DAILY_THRESHOLD:
color = EmbedStyles.WARNING # Orange for decent
else:
color = EmbedStyles.INFO # Blue for low
embed = discord.Embed(title=title, description=description, color=color)
for i, quote in enumerate(quotes, 1):
speaker_name = quote.get("speaker_name", "Unknown") or "Unknown"
quote_text = quote.get("text", "No text") or "No text"
overall_score = quote.get("overall_score", 0.0) or 0.0
timestamp = quote.get("timestamp", datetime.now(timezone.utc))
# Truncate long quotes
display_text = ValidationHelper.sanitize_user_input(
UIFormatter.truncate_text(quote_text, 150)
)
# Create dimensional scores display
dimensional_scores = self._format_dimensional_scores(quote)
score_bar = UIFormatter.format_score_bar(overall_score)
field_value = (
f'*"{display_text}"*\n'
f"{score_bar} **{overall_score:.1f}/10**\n"
f"{dimensional_scores}\n"
f"<t:{int(timestamp.timestamp())}:R>"
)
embed.add_field(
name=f"{i}. {speaker_name}",
value=field_value,
inline=False,
)
return embed
def _format_dimensional_scores(self, quote: dict[str, Any]) -> str:
"""Format dimensional scores with emojis and bars."""
score_categories = [
("funny_score", "funny", StatusIndicators.FUNNY),
("dark_score", "dark", StatusIndicators.DARK),
("silly_score", "silly", StatusIndicators.SILLY),
("suspicious_score", "suspicious", StatusIndicators.SUSPICIOUS),
("asinine_score", "asinine", StatusIndicators.ASININE),
]
formatted_scores = []
for score_key, _, emoji in score_categories:
score = quote.get(score_key, 0.0) or 0.0
if score > 1.0: # Only show meaningful scores
formatted_scores.append(f"{emoji}{score:.1f}")
return " ".join(formatted_scores) if formatted_scores else "📊 General"
@app_commands.command(
name="quote_stats", description="Show quote statistics for the server"
)
async def quote_stats(self, interaction: discord.Interaction) -> None:
"""Display quote statistics for the current server"""
await interaction.response.defer()
try:
guild_id = interaction.guild_id
if guild_id is None:
embed = EmbedBuilder.create_error_embed(
"Error", "This command must be used in a server."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
stats = await self.db_manager.get_quote_stats(guild_id)
guild_name = interaction.guild.name if interaction.guild else "Unknown"
embed = EmbedBuilder.create_info_embed(
"Quote Statistics", f"Stats for {guild_name}"
)
embed.add_field(
name="Total Quotes",
value=str(stats.get("total_quotes", 0)),
inline=True,
)
embed.add_field(
name="Total Speakers",
value=str(stats.get("unique_speakers", 0)),
inline=True,
)
embed.add_field(
name="Average Score",
value=f"{stats.get('avg_score', 0.0):.1f}",
inline=True,
)
embed.add_field(
name="Highest Score",
value=f"{stats.get('max_score', 0.0):.1f}",
inline=True,
)
embed.add_field(
name="This Week",
value=str(stats.get("quotes_this_week", 0)),
inline=True,
)
embed.add_field(
name="This Month",
value=str(stats.get("quotes_this_month", 0)),
inline=True,
)
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in quote_stats command: {e}")
embed = EmbedBuilder.create_error_embed(
"Statistics Error", "Failed to retrieve quote statistics."
)
await interaction.followup.send(embed=embed, ephemeral=True)
@app_commands.command(name="my_quotes", description="Show your quotes")
@app_commands.describe(limit="Number of quotes to display (1-10, default 5)")
async def my_quotes(
self, interaction: discord.Interaction, limit: int | None = 5
) -> None:
"""Show quotes from the command user"""
# Convert interaction.user to Member if in guild context
user_member = None
if interaction.guild and isinstance(interaction.user, discord.Member):
user_member = interaction.user
elif interaction.guild:
# Try to get member from guild
user_member = interaction.guild.get_member(interaction.user.id)
if not user_member:
embed = EmbedBuilder.create_error_embed(
"User Not Found", "Unable to find user in this server context."
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return
# Call the quotes functionality directly
# Extract the quotes search logic into a reusable method
await interaction.response.defer()
try:
# Validate limit
limit = max(1, min(limit or 5, 10))
# Build search parameters
search_params = {
"guild_id": interaction.guild_id,
"search_term": None,
"user_id": user_member.id,
"limit": limit,
}
# Get quotes from database with dimensional scores
quotes = await self.db_manager.search_quotes(**search_params)
if not quotes:
embed = EmbedBuilder.create_info_embed(
"No Quotes Found", f"No quotes found for {user_member.mention}."
)
await interaction.followup.send(embed=embed)
return
# Create enhanced embed with dimensional scores
embed = await self._create_quotes_embed(
f"Quotes for {user_member.display_name}",
f"Found {len(quotes)} quote(s)",
quotes,
)
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in my_quotes command: {e}")
embed = EmbedBuilder.create_error_embed(
"Quote Search Error",
"Failed to retrieve quotes.",
details=str(e) if logger.isEnabledFor(logging.DEBUG) else None,
)
await interaction.followup.send(embed=embed, ephemeral=True)
@app_commands.command(name="top_quotes", description="Show highest-rated quotes")
@app_commands.describe(limit="Number of quotes to display (1-10, default 5)")
async def top_quotes(
self, interaction: discord.Interaction, limit: int | None = 5
) -> None:
"""Show top-rated quotes from the server"""
await interaction.response.defer()
try:
guild_id = interaction.guild_id
if guild_id is None:
embed = EmbedBuilder.create_error_embed(
"Error", "This command must be used in a server."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
limit = max(1, min(limit or 5, 10))
quotes = await self.db_manager.get_top_quotes(guild_id, limit)
if not quotes:
embed = EmbedBuilder.create_info_embed(
"No Quotes", "No quotes found in this server."
)
await interaction.followup.send(embed=embed)
return
# Use enhanced embed with dimensional scores
embed = await self._create_quotes_embed(
"Top Quotes",
f"Highest-rated quotes from {interaction.guild.name}",
quotes,
)
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in top_quotes command: {e}")
embed = EmbedBuilder.create_error_embed(
"Top Quotes Error", "Failed to retrieve top quotes."
)
await interaction.followup.send(embed=embed, ephemeral=True)
@app_commands.command(name="random_quote", description="Get a random quote")
async def random_quote(self, interaction: discord.Interaction) -> None:
"""Get a random quote from the server"""
await interaction.response.defer()
try:
guild_id = interaction.guild_id
if guild_id is None:
embed = EmbedBuilder.create_error_embed(
"Error", "This command must be used in a server."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
quote = await self.db_manager.get_random_quote(guild_id)
if not quote:
embed = EmbedBuilder.create_info_embed(
"No Quotes", "No quotes found in this server."
)
await interaction.followup.send(embed=embed)
return
# Use enhanced embed for single quote display
embed = await self._create_quotes_embed(
"Random Quote", "Here's a random quote for you!", [quote]
)
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in random_quote command: {e}")
embed = EmbedBuilder.create_error_embed(
"Random Quote Error", "Failed to retrieve random quote."
)
await interaction.followup.send(embed=embed, ephemeral=True)
@app_commands.command(
name="explain_quote",
description="Get detailed AI analysis explanation for a quote",
)
@app_commands.describe(
quote_id="Quote ID to explain (from quote display)",
search="Search for a quote to explain",
depth="Level of detail (basic, detailed, comprehensive)",
)
async def explain_quote(
self,
interaction: discord.Interaction,
quote_id: int | None = None,
search: str | None = None,
depth: str = "detailed",
) -> None:
"""Provide detailed AI explanation of quote analysis."""
await interaction.response.defer()
try:
if not self.explanation_service:
embed = EmbedBuilder.create_warning_embed(
"Feature Unavailable",
"Quote explanation service is not available.",
warning=(
"AI analysis features require proper service " "initialization."
),
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
# Validate depth parameter
try:
explanation_depth = ExplanationDepth(depth.lower())
except ValueError:
embed = EmbedBuilder.create_error_embed(
"Invalid Depth",
"Depth must be 'basic', 'detailed', or 'comprehensive'.",
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
# Find quote by ID or search
guild_id = interaction.guild_id
if guild_id is None:
embed = EmbedBuilder.create_error_embed(
"Error", "This command must be used in a server."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
target_quote_id = await self._resolve_quote_id(guild_id, quote_id, search)
if not target_quote_id:
embed = EmbedBuilder.create_error_embed(
"Quote Not Found",
"Could not find the specified quote.",
details=("Try providing a valid quote ID or search term."),
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
# Initialize explanation service if needed
if not self.explanation_service._initialized:
await self.explanation_service.initialize()
# Generate explanation
explanation = await self.explanation_service.generate_explanation(
target_quote_id, explanation_depth
)
if not explanation:
embed = EmbedBuilder.create_error_embed(
"Analysis Failed", "Failed to generate quote explanation."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
# Create explanation embed and view
embed = await self.explanation_service.create_explanation_embed(explanation)
view = await self.explanation_service.create_explanation_view(explanation)
await interaction.followup.send(embed=embed, view=view)
except Exception as e:
logger.error(f"Error in explain_quote command: {e}")
embed = EmbedBuilder.create_error_embed(
"Explanation Error",
"Failed to generate quote explanation.",
details=str(e) if logger.isEnabledFor(logging.DEBUG) else None,
)
await interaction.followup.send(embed=embed, ephemeral=True)
@app_commands.command(
name="legendary_quotes",
description=f"Show legendary quotes (score >= {REALTIME_THRESHOLD})",
)
@app_commands.describe(limit="Number of quotes to display (1-10, default 5)")
async def legendary_quotes(
self, interaction: discord.Interaction, limit: int | None = 5
) -> None:
"""Show quotes above the realtime threshold for legendary content."""
await interaction.response.defer()
try:
guild_id = interaction.guild_id
if guild_id is None:
embed = EmbedBuilder.create_error_embed(
"Error", "This command must be used in a server."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
limit = max(1, min(limit or 5, 10))
# Get quotes above realtime threshold
quotes = await self.db_manager.get_quotes_by_score(
guild_id, self.REALTIME_THRESHOLD, limit
)
if not quotes:
embed = EmbedBuilder.create_info_embed(
"No Legendary Quotes",
f"No quotes found with score >= {self.REALTIME_THRESHOLD:.1f} in this server.",
)
await interaction.followup.send(embed=embed)
return
# Create enhanced embed with golden styling for legendary quotes
embed = discord.Embed(
title="🏆 Legendary Quotes",
description=(
f"Top {len(quotes)} legendary quotes "
f"(score >= {self.REALTIME_THRESHOLD:.1f})"
),
color=EmbedStyles.FUNNY, # Gold color
)
for i, quote in enumerate(quotes, 1):
speaker_name = quote.get("speaker_name", "Unknown") or "Unknown"
quote_text = quote.get("text", "No text") or "No text"
overall_score = quote.get("overall_score", 0.0) or 0.0
timestamp = quote.get("timestamp", datetime.now(timezone.utc))
# Enhanced display for legendary quotes
display_text = ValidationHelper.sanitize_user_input(
UIFormatter.truncate_text(quote_text, 180)
)
dimensional_scores = self._format_dimensional_scores(quote)
score_bar = UIFormatter.format_score_bar(overall_score)
field_value = (
f'*"{display_text}"*\n'
f"🌟 {score_bar} **{overall_score:.2f}/10** 🌟\n"
f"{dimensional_scores}\n"
f"<t:{int(timestamp.timestamp())}:F>"
)
embed.add_field(
name=f"#{i} {speaker_name}",
value=field_value,
inline=False,
)
embed.set_footer(text=f"Realtime threshold: {self.REALTIME_THRESHOLD}")
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in legendary_quotes command: {e}")
embed = EmbedBuilder.create_error_embed(
"Legendary Quotes Error", "Failed to retrieve legendary quotes."
)
await interaction.followup.send(embed=embed, ephemeral=True)
@app_commands.command(
name="search_by_category",
description="Search quotes by dimensional score categories",
)
@app_commands.describe(
category="Score category (funny, dark, silly, suspicious, asinine)",
min_score="Minimum score for the category (0.0-10.0)",
limit="Number of quotes to display (1-10, default 5)",
)
async def search_by_category(
self,
interaction: discord.Interaction,
category: str,
min_score: float = 5.0,
limit: int | None = 5,
) -> None:
"""Search quotes by specific dimensional score categories."""
await interaction.response.defer()
try:
# Validate category
valid_categories = ["funny", "dark", "silly", "suspicious", "asinine"]
category = category.lower()
if category not in valid_categories:
embed = EmbedBuilder.create_error_embed(
"Invalid Category",
f"Category must be one of: {', '.join(valid_categories)}",
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
# Validate score range
min_score = max(0.0, min(min_score, 10.0))
limit = max(1, min(limit or 5, 10))
# Build query for category search
score_column = f"{category}_score"
quotes = await self.db_manager.execute_query(
f"""
SELECT q.*, u.username as speaker_name
FROM quotes q
LEFT JOIN user_consent u ON q.user_id = u.user_id AND q.guild_id = u.guild_id
WHERE q.guild_id = $1 AND q.{score_column} >= $2
ORDER BY q.{score_column} DESC
LIMIT $3
""",
interaction.guild_id,
min_score,
limit,
)
if not quotes:
embed = EmbedBuilder.create_info_embed(
"No Matches Found",
f"No quotes found with {category} score >= {min_score:.1f}",
)
await interaction.followup.send(embed=embed)
return
# Get category emoji and color
category_emoji = StatusIndicators.get_score_emoji(category)
category_colors = {
"funny": EmbedStyles.FUNNY,
"dark": EmbedStyles.DARK,
"silly": EmbedStyles.SILLY,
"suspicious": EmbedStyles.SUSPICIOUS,
"asinine": EmbedStyles.ASININE,
}
embed = discord.Embed(
title=f"{category_emoji} {category.title()} Quotes",
description=(
f"Top {len(quotes)} quotes with {category} score >= {min_score:.1f}"
),
color=category_colors.get(category, EmbedStyles.INFO),
)
for i, quote in enumerate(quotes, 1):
speaker_name = quote.get("speaker_name", "Unknown") or "Unknown"
quote_text = quote.get("text", "No text") or "No text"
category_score = quote.get(score_column, 0.0) or 0.0
overall_score = quote.get("overall_score", 0.0) or 0.0
timestamp = quote.get("timestamp", datetime.now(timezone.utc))
display_text = ValidationHelper.sanitize_user_input(
UIFormatter.truncate_text(quote_text, 150)
)
dimensional_scores = self._format_dimensional_scores(quote)
category_bar = UIFormatter.format_score_bar(category_score)
field_value = (
f'*"{display_text}"*\n'
f"{category_emoji} {category_bar} **{category_score:.1f}/10**\n"
f"📊 Overall: **{overall_score:.1f}/10**\n"
f"{dimensional_scores}\n"
f"<t:{int(timestamp.timestamp())}:R>"
)
embed.add_field(
name=f"{i}. {speaker_name}",
value=field_value,
inline=False,
)
embed.set_footer(text=f"Filtered by {category} score >= {min_score:.1f}")
await interaction.followup.send(embed=embed)
except Exception as e:
logger.error(f"Error in search_by_category command: {e}")
embed = EmbedBuilder.create_error_embed(
"Category Search Error", "Failed to search quotes by category."
)
await interaction.followup.send(embed=embed, ephemeral=True)
async def _resolve_quote_id(
self, guild_id: int, quote_id: int | None, search: str | None
) -> int | None:
"""Resolve quote ID from direct ID or search term."""
try:
if quote_id:
# Verify quote exists in this guild
quote = await self.db_manager.execute_query(
"SELECT id FROM quotes WHERE id = $1 AND guild_id = $2",
quote_id,
guild_id,
fetch_one=True,
)
return quote["id"] if quote else None
elif search:
# Find first matching quote
quotes = await self.db_manager.search_quotes(
guild_id=guild_id, search_term=search, limit=1
)
return quotes[0]["id"] if quotes else None
return None
except Exception as e:
logger.error(f"Error resolving quote ID: {e}")
return None
async def setup(bot: "QuoteBot") -> None:
"""Setup function for the cog."""
await bot.add_cog(QuotesCog(bot))