""" Giphy Service for Discord Bot v2.0 Provides async interface to Giphy API with disappointment-based search phrases. Used for Easter egg features like the soak command. """ import random from typing import List, Optional from urllib.parse import quote import aiohttp from utils.logging import get_contextual_logger from config import get_config from exceptions import APIException # Disappointment tier configuration DISAPPOINTMENT_TIERS = { "tier_1": { "max_seconds": 1800, # 30 minutes "phrases": [ "extremely disappointed", "so disappointed", "are you kidding me", "seriously", "unbelievable", ], "description": "Maximum Disappointment", }, "tier_2": { "max_seconds": 7200, # 2 hours "phrases": [ "very disappointed", "can't believe you", "not happy", "shame on you", "facepalm", ], "description": "Severe Disappointment", }, "tier_3": { "max_seconds": 21600, # 6 hours "phrases": [ "disappointed", "not impressed", "shaking head", "eye roll", "really", ], "description": "Strong Disappointment", }, "tier_4": { "max_seconds": 86400, # 24 hours "phrases": [ "mildly disappointed", "not great", "could be better", "sigh", "seriously", ], "description": "Moderate Disappointment", }, "tier_5": { "max_seconds": 604800, # 7 days "phrases": ["slightly disappointed", "oh well", "shrug", "meh", "not bad"], "description": "Mild Disappointment", }, "tier_6": { "max_seconds": float("inf"), # 7+ days "phrases": [ "not disappointed", "relieved", "proud", "been worse", "fine i guess", ], "description": "Minimal Disappointment", }, "first_ever": { "phrases": [ "here we go", "oh boy", "uh oh", "getting started", "and so it begins", ], "description": "The Beginning", }, } class GiphyService: """Service for fetching GIFs from Giphy API based on disappointment tiers.""" def __init__(self): """Initialize Giphy service with configuration.""" self.config = get_config() self.api_key = self.config.giphy_api_key self.translate_url = self.config.giphy_translate_url self.logger = get_contextual_logger(f"{__name__}.GiphyService") self._session: Optional[aiohttp.ClientSession] = None def _get_session(self) -> aiohttp.ClientSession: """Return the shared aiohttp session, creating it lazily if needed.""" if self._session is None or self._session.closed: self._session = aiohttp.ClientSession() return self._session def get_tier_for_seconds(self, seconds_elapsed: Optional[int]) -> str: """ Determine disappointment tier based on elapsed time. Args: seconds_elapsed: Seconds since last soak, or None for first ever Returns: Tier key string (e.g., 'tier_1', 'first_ever') """ if seconds_elapsed is None: return "first_ever" for tier_key in ["tier_1", "tier_2", "tier_3", "tier_4", "tier_5", "tier_6"]: if seconds_elapsed <= DISAPPOINTMENT_TIERS[tier_key]["max_seconds"]: return tier_key return "tier_6" # Fallback to lowest disappointment def get_random_phrase_for_tier(self, tier_key: str) -> str: """ Get a random search phrase from the specified tier. Args: tier_key: Tier identifier (e.g., 'tier_1', 'first_ever') Returns: Random search phrase from that tier Raises: ValueError: If tier_key is invalid """ if tier_key not in DISAPPOINTMENT_TIERS: raise ValueError(f"Invalid tier key: {tier_key}") phrases = DISAPPOINTMENT_TIERS[tier_key]["phrases"] return random.choice(phrases) def get_tier_description(self, tier_key: str) -> str: """ Get the human-readable description for a tier. Args: tier_key: Tier identifier Returns: Description string Raises: ValueError: If tier_key is invalid """ if tier_key not in DISAPPOINTMENT_TIERS: raise ValueError(f"Invalid tier key: {tier_key}") return DISAPPOINTMENT_TIERS[tier_key]["description"] async def get_disappointment_gif(self, tier_key: str) -> str: """ Fetch a GIF from Giphy based on disappointment tier. Randomly selects a search phrase from the tier and queries Giphy. Filters out Trump GIFs (legacy behavior). Falls back to trying other phrases if first fails. Args: tier_key: Tier identifier (e.g., 'tier_1', 'first_ever') Returns: GIF URL string Raises: ValueError: If tier_key is invalid APIException: If all GIF fetch attempts fail """ if tier_key not in DISAPPOINTMENT_TIERS: raise ValueError(f"Invalid tier key: {tier_key}") phrases = DISAPPOINTMENT_TIERS[tier_key]["phrases"] # Shuffle phrases for variety and retry capability shuffled_phrases = random.sample(phrases, len(phrases)) session = self._get_session() for phrase in shuffled_phrases: try: url = f"{self.translate_url}?s={quote(phrase)}&api_key={quote(self.api_key)}" async with session.get( url, timeout=aiohttp.ClientTimeout(total=5) ) as resp: if resp.status == 200: data = await resp.json() # Filter out Trump GIFs (legacy behavior) gif_title = data.get("data", {}).get("title", "").lower() if "trump" in gif_title: self.logger.debug( f"Filtered out Trump GIF for phrase: {phrase}" ) continue # Get the actual GIF image URL, not the web page URL gif_url = ( data.get("data", {}) .get("images", {}) .get("original", {}) .get("url") ) if gif_url: self.logger.info( f"Successfully fetched GIF for phrase: {phrase}", gif_url=gif_url, ) return gif_url else: self.logger.warning( f"No GIF URL in response for phrase: {phrase}" ) else: self.logger.warning( f"Giphy API returned status {resp.status} for phrase: {phrase}" ) except aiohttp.ClientError as e: self.logger.error(f"HTTP error fetching GIF for phrase '{phrase}': {e}") except Exception as e: self.logger.error( f"Unexpected error fetching GIF for phrase '{phrase}': {e}" ) # All phrases failed error_msg = f"Failed to fetch any GIF for tier: {tier_key}" self.logger.error(error_msg) raise APIException(error_msg) async def get_gif( self, phrase: Optional[str] = None, phrase_options: Optional[List[str]] = None ) -> str: """ Fetch a GIF from Giphy based on a phrase or list of phrase options. Args: phrase: Specific search phrase to use phrase_options: List of phrases to randomly choose from Returns: GIF URL string Raises: ValueError: If neither phrase nor phrase_options is provided APIException: If all GIF fetch attempts fail """ if phrase is None and phrase_options is None: raise ValueError( "To get a gif, one of `phrase` or `phrase_options` must be provided" ) search_phrase = "send help" if phrase is not None: search_phrase = phrase elif phrase_options is not None: search_phrase = random.choice(phrase_options) session = self._get_session() attempts = 0 while attempts < 3: attempts += 1 try: url = f"{self.translate_url}?s={quote(search_phrase)}&api_key={quote(self.api_key)}" async with session.get( url, timeout=aiohttp.ClientTimeout(total=3) ) as resp: if resp.status != 200: self.logger.warning( f"Giphy API returned status {resp.status} for phrase: {search_phrase}" ) continue data = await resp.json() # Filter out Trump GIFs (legacy behavior) gif_title = data.get("data", {}).get("title", "").lower() if "trump" in gif_title: self.logger.debug( f"Filtered out Trump GIF for phrase: {search_phrase}" ) continue # Get the actual GIF image URL, not the web page URL gif_url = ( data.get("data", {}) .get("images", {}) .get("original", {}) .get("url") ) if gif_url: self.logger.info( f"Successfully fetched GIF for phrase: {search_phrase}", gif_url=gif_url, ) return gif_url else: self.logger.warning( f"No GIF URL in response for phrase: {search_phrase}" ) except aiohttp.ClientError as e: self.logger.error( f"HTTP error fetching GIF for phrase '{search_phrase}': {e}" ) except Exception as e: self.logger.error( f"Unexpected error fetching GIF for phrase '{search_phrase}': {e}" ) # All attempts failed error_msg = f"Failed to fetch any GIF for phrase: {search_phrase}" self.logger.error(error_msg) raise APIException(error_msg)