feat: Implement Tenor GIF search functionality and integrate with tool mapping

This commit is contained in:
Slipstream 2025-05-28 14:51:30 -06:00
parent f0de735d13
commit bb1c34f0c5
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD
2 changed files with 32 additions and 34 deletions

View File

@ -1956,35 +1956,3 @@ async def get_internal_ai_json_response(
if __name__ == "__main__":
print(_preprocess_schema_for_vertex(RESPONSE_SCHEMA['schema']))
async def search_tenor_gifs(cog: 'GurtCog', query: str, limit: int = 5) -> List[str]:
"""Searches Tenor for GIFs and returns a list of URLs."""
if not cog.TENOR_API_KEY:
print("Tenor API key not configured.")
return []
url = "https://tenor.googleapis.com/v2/search"
params = {
"q": query,
"key": cog.TENOR_API_KEY,
"limit": limit,
"media_filter": "gif", # Ensure we get GIFs
"contentfilter": "medium" # Adjust content filter as needed (off, low, medium, high)
}
try:
if not cog.session:
cog.session = aiohttp.ClientSession() # Ensure session exists
async with cog.session.get(url, params=params, timeout=10) as response:
if response.status == 200:
data = await response.json()
gif_urls = [result["media_formats"]["gif"]["url"] for result in data.get("results", []) if "media_formats" in result and "gif" in result["media_formats"] and "url" in result["media_formats"]["gif"]]
return gif_urls
else:
print(f"Error searching Tenor: {response.status} - {await response.text()}")
return []
except Exception as e:
print(f"Exception during Tenor API call: {e}")
return []

View File

@ -30,7 +30,6 @@ from .config import (
# Add these:
TAVILY_DEFAULT_SEARCH_DEPTH, TAVILY_DEFAULT_MAX_RESULTS, TAVILY_DISABLE_ADVANCED
)
from .api import search_tenor_gifs # Import search_tenor_gifs
# Assume these helpers will be moved or are accessible via cog
# We might need to pass 'cog' to these tool functions if they rely on cog state heavily
@ -2958,7 +2957,6 @@ async def tool_search_tenor_gifs(cog: commands.Cog, query: str, limit: int = 1)
# Ensure limit is within a reasonable range, e.g., 1-10
limit = max(1, min(limit, 10))
try:
# search_tenor_gifs is imported from .api
gif_urls = await search_tenor_gifs(cog, query, limit=limit)
if gif_urls:
return {
@ -2991,5 +2989,37 @@ async def tool_search_tenor_gifs(cog: commands.Cog, query: str, limit: int = 1)
"timestamp": datetime.datetime.now().isoformat()
}
# --- Tenor GIF Search Tool Implementation ---
async def search_tenor_gifs(cog: commands.Cog, query: str, limit: int = 5) -> List[str]:
"""Searches Tenor for GIFs and returns a list of URLs."""
if not cog.TENOR_API_KEY:
print("Tenor API key not configured.")
return []
url = "https://tenor.googleapis.com/v2/search"
params = {
"q": query,
"key": cog.TENOR_API_KEY,
"limit": limit,
"media_filter": "gif", # Ensure we get GIFs
"contentfilter": "medium" # Adjust content filter as needed (off, low, medium, high)
}
try:
if not cog.session:
cog.session = aiohttp.ClientSession() # Ensure session exists
async with cog.session.get(url, params=params, timeout=10) as response:
if response.status == 200:
data = await response.json()
gif_urls = [result["media_formats"]["gif"]["url"] for result in data.get("results", []) if "media_formats" in result and "gif" in result["media_formats"] and "url" in result["media_formats"]["gif"]]
return gif_urls
else:
print(f"Error searching Tenor: {response.status} - {await response.text()}")
return []
except Exception as e:
print(f"Exception during Tenor API call: {e}")
return []
# Update TOOL_MAPPING to include the new Tenor GIF tool
TOOL_MAPPING["search_tenor_gifs"] = tool_search_tenor_gifs