""" Giphy GIF Replacer Filter for Open WebUI Author: Alban GitHub: https://github.com/reise69 Installation instructions: https://github.com/reise69/open-webui-function-gif-replacer Version: 1.0.0 Description: Add some fun in your chat. A comprehensive filter that replaces /gif "query" commands with random GIFs from Giphy. Features: - Random GIF selection - Configurable max results - Debug mode Requirements: - Giphy API key required_open_webui_version: 0.3.30 """ import os import re import random import requests from pydantic import BaseModel, Field from typing import Optional, Callable, Any, Awaitable class Filter: """ A filter to replace text commands with random GIFs from Giphy """ class Valves(BaseModel): """ Configuration settings for the filter """ GIPHY_API_KEY: str = Field( default="", description="Giphy API key for searching GIFs" ) DEBUG_MODE: bool = Field( default=False, description="Enable detailed debug logging" ) PRIORITY: int = Field( default=5, description="Filter processing priority" ) MAX_GIF_RESULTS: int = Field( default=10, description="Maximum number of GIFs to retrieve per search" ) class UserValves(BaseModel): """ User-configurable settings """ ENABLE_GIF_REPLACE: bool = Field( default=True, description="Enable GIF replacement functionality" ) RANDOM_GIF: bool = Field( default=True, description="Select a random GIF from search results" ) def __init__(self): """ Initialize filter with default configurations """ self.valves = self.Valves( GIPHY_API_KEY=os.getenv("GIPHY_API_KEY", "") ) self.user_valves = self.UserValves() self.gif_cache = {} # Cache to store search results def _find_gif_commands(self, content: str) -> list: """ Find all /gif "query" commands in the content Args: content (str): Text to search for GIF commands Returns: list: Found GIF query commands """ pattern = r'/gif\s*"([^"]*)"' return re.findall(pattern, content) def _get_gif_url(self, query: str) -> str: """ Retrieve a GIF URL from Giphy with randomization strategy Args: query (str): Search term for GIF Returns: str: URL of a selected GIF """ if not self.valves.GIPHY_API_KEY: return f"Error: Giphy API key missing for query: {query}" # Use cache if available if query in self.gif_cache: gifs = self.gif_cache[query] else: params = { "api_key": self.valves.GIPHY_API_KEY, "q": query, "limit": self.valves.MAX_GIF_RESULTS, "rating": "g", } try: response = requests.get( "https://api.giphy.com/v1/gifs/search", params=params ) data = response.json() if data.get("data"): gifs = [ gif["images"]["fixed_width"]["url"] for gif in data["data"] if gif.get("images", {}).get("fixed_width", {}).get("url") ] self.gif_cache[query] = gifs else: return f"No GIF found for: {query}" except Exception as e: return f"GIF search error: {e}" if not gifs: return f"No GIF found for: {query}" if self.user_valves.RANDOM_GIF: # Random selection return random.choice(gifs) else: # Sequential rotation gif_url = gifs[0] self.gif_cache[query] = gifs[1:] + [gif_url] return gif_url def inlet(self, body: dict, __user__: Optional[dict] = None) -> dict: """ Pre-processing method (currently a pass-through) Args: body (dict): Incoming message body __user__ (dict, optional): User information Returns: dict: Processed message body """ if not self.user_valves.ENABLE_GIF_REPLACE: return body return body async def outlet( self, body: dict, __user__: Optional[dict] = None, __event_emitter__: Optional[Callable[[Any], Awaitable[None]]] = None, ) -> dict: """ Post-processing method to replace GIF commands Args: body (dict): Message body __user__ (dict, optional): User information __event_emitter__ (Callable, optional): Event emission function Returns: dict: Processed message body """ if not self.user_valves.ENABLE_GIF_REPLACE: return body messages = body.get("messages", []) if not messages: return body last_message = messages[-1] # Process text content if isinstance(last_message.get("content"), str): content = last_message["content"] gif_commands = self._find_gif_commands(content) for query in gif_commands: gif_url = self._get_gif_url(query) content = content.replace(f'/gif "{query}"', f"") last_message["content"] = content # Process list content elif isinstance(last_message.get("content"), list): for item in last_message["content"]: if item.get("type") == "text": text = item["text"] gif_commands = self._find_gif_commands(text) for query in gif_commands: gif_url = self._get_gif_url(query) text = text.replace(f'/gif "{query}"', f"") item["text"] = text # Emit debug status if self.valves.DEBUG_MODE and __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "GIF Filter: Commands processed", "done": True, }, } ) return body