fix: Refactor emoji and sticker handling to use FileData for improved MIME type management

This commit is contained in:
Slipstream 2025-05-28 16:22:39 -06:00
parent df267d7db6
commit 2821f43454
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD

View File

@ -873,53 +873,35 @@ async def get_ai_response(cog: 'GurtCog', message: discord.Message, model_name:
if emoji_name:
text_parts.append(f"[Emoji: {emoji_name}]")
if emoji_url and emoji_url.startswith('http'):
# Determine MIME type for emoji URI
is_animated_emoji = emoji_info.get("animated", False)
emoji_mime_type = "image/gif" if is_animated_emoji else "image/png"
try:
if not hasattr(cog, 'session') or not cog.session:
raise ValueError("aiohttp session not found in cog for emoji download.")
async with cog.session.get(emoji_url, timeout=5) as response:
if response.status == 200:
image_bytes = await response.read()
mime_type = response.content_type or "image/png"
supported_emoji_mimes = ["image/png", "image/jpeg", "image/webp", "image/gif"]
clean_mime_type = mime_type.split(';')[0].lower()
if clean_mime_type in supported_emoji_mimes:
parts.append(types.Part(inline_data=types.Blob(data=image_bytes, mime_type=clean_mime_type)))
print(f"Added image part for historical emoji: {emoji_name} (MIME: {clean_mime_type})")
else:
print(f"Unsupported MIME type for historical emoji {emoji_name}: {clean_mime_type}. URL: {emoji_url}")
else:
print(f"Failed to download historical emoji {emoji_name} from {emoji_url}: Status {response.status}")
parts.append(types.Part(file_data=types.FileData(mime_type=emoji_mime_type, file_uri=emoji_url)))
print(f"Added FileData part for historical emoji: {emoji_name} (MIME: {emoji_mime_type}, URL: {emoji_url})")
except Exception as e:
print(f"Error processing historical emoji image {emoji_name} from {emoji_url}: {e}")
print(f"Error creating FileData for historical emoji {emoji_name} from {emoji_url}: {e}")
text_parts.append(f"[System Note: Failed to attach emoji '{emoji_name}' as FileData]")
cached_stickers = msg.get("stickers", [])
for sticker_info in cached_stickers:
sticker_name = sticker_info.get("name")
sticker_url = sticker_info.get("url")
sticker_format_text = f" (Format: {sticker_info.get('format')})" if sticker_info.get('format') else ""
sticker_format_str = sticker_info.get("format")
sticker_format_text = f" (Format: {sticker_format_str})" if sticker_format_str else ""
if sticker_name:
text_parts.append(f"[Sticker: {sticker_name}{sticker_format_text}]")
is_downloadable_sticker = sticker_info.get("format") in ["StickerFormatType.png", "StickerFormatType.apng"]
if is_downloadable_sticker and sticker_url and sticker_url.startswith('http'):
is_image_sticker = sticker_format_str in ["StickerFormatType.png", "StickerFormatType.apng"]
if is_image_sticker and sticker_url and sticker_url.startswith('http'):
# Determine MIME type for sticker URI (PNG for both PNG and APNG)
sticker_mime_type = "image/png"
try:
if not hasattr(cog, 'session') or not cog.session:
raise ValueError("aiohttp session not found in cog for sticker download.")
async with cog.session.get(sticker_url, timeout=5) as response:
if response.status == 200:
image_bytes = await response.read()
mime_type = response.content_type or "image/png"
supported_sticker_mimes = ["image/png", "image/webp"]
clean_mime_type = mime_type.split(';')[0].lower()
if clean_mime_type in supported_sticker_mimes:
parts.append(types.Part(inline_data=types.Blob(data=image_bytes, mime_type=clean_mime_type)))
print(f"Added image part for historical sticker: {sticker_name} (MIME: {clean_mime_type})")
else:
print(f"Unsupported MIME type for historical sticker {sticker_name}: {clean_mime_type}. URL: {sticker_url}")
else:
print(f"Failed to download historical sticker {sticker_name} from {sticker_url}: Status {response.status}")
parts.append(types.Part(file_data=types.FileData(mime_type=sticker_mime_type, file_uri=sticker_url)))
print(f"Added FileData part for historical sticker: {sticker_name} (MIME: {sticker_mime_type}, URL: {sticker_url})")
except Exception as e:
print(f"Error processing historical sticker image {sticker_name} from {sticker_url}: {e}")
elif sticker_info.get("format") == "StickerFormatType.lottie":
print(f"Error creating FileData for historical sticker {sticker_name} from {sticker_url}: {e}")
text_parts.append(f"[System Note: Failed to attach sticker '{sticker_name}' as FileData]")
elif sticker_format_str == "StickerFormatType.lottie":
parts.append(types.Part(text=f"[Sticker is a Lottie animation: {sticker_name}, URL: {sticker_url}]"))
@ -1092,63 +1074,40 @@ async def get_ai_response(cog: 'GurtCog', message: discord.Message, model_name:
if emoji_name and emoji_url:
emoji_sticker_parts_to_add.append(types.Part(text=f"[Emoji: {emoji_name}]"))
print(f"Added text description for current message emoji: {emoji_name}")
# Determine MIME type for emoji URI
is_animated_emoji = emoji_info.get("animated", False)
emoji_mime_type = "image/gif" if is_animated_emoji else "image/png"
try:
if not hasattr(cog, 'session') or not cog.session:
raise ValueError("aiohttp session not found in cog for emoji download.")
async with cog.session.get(emoji_url, timeout=10) as response:
if response.status == 200:
image_bytes = await response.read()
mime_type = response.content_type or "image/png" # Default for emojis
# Gemini supports: image/png, image/jpeg, image/webp, image/heic, image/heif. GIFs are common for emojis.
supported_emoji_mimes = ["image/png", "image/jpeg", "image/webp", "image/gif"]
clean_mime_type = mime_type.split(';')[0].lower()
if clean_mime_type in supported_emoji_mimes:
image_part = types.Part(inline_data=types.Blob(data=image_bytes, mime_type=clean_mime_type))
emoji_sticker_parts_to_add.append(image_part)
print(f"Added image part for emoji: {emoji_name} (MIME: {clean_mime_type})")
else:
print(f"Unsupported MIME type for emoji {emoji_name}: {clean_mime_type}. URL: {emoji_url}")
else:
print(f"Failed to download emoji {emoji_name} from {emoji_url}: Status {response.status}")
image_part = types.Part(file_data=types.FileData(mime_type=emoji_mime_type, file_uri=emoji_url))
emoji_sticker_parts_to_add.append(image_part)
print(f"Added FileData part for current emoji: {emoji_name} (MIME: {emoji_mime_type}, URL: {emoji_url})")
except Exception as e:
print(f"Error processing emoji image {emoji_name} from {emoji_url}: {e}")
print(f"Error creating FileData for current emoji {emoji_name} from {emoji_url}: {e}")
emoji_sticker_parts_to_add.append(types.Part(text=f"[System Note: Failed to attach emoji '{emoji_name}' as FileData]"))
# Process stickers from formatted_current_message
stickers_current = formatted_current_message.get("stickers", [])
for sticker_info in stickers_current:
sticker_name = sticker_info.get("name")
sticker_url = sticker_info.get("url")
# Assuming format_message stores format as string like "StickerFormatType.png"
sticker_format_str = sticker_info.get("format")
if sticker_name and sticker_url:
emoji_sticker_parts_to_add.append(types.Part(text=f"[Sticker: {sticker_name}]"))
print(f"Added text description for current message sticker: {sticker_name}")
# Check if sticker format is PNG or APNG (which is also PNG compatible)
is_downloadable_sticker = sticker_format_str in ["StickerFormatType.png", "StickerFormatType.apng"]
is_image_sticker = sticker_format_str in ["StickerFormatType.png", "StickerFormatType.apng"]
if is_downloadable_sticker:
if is_image_sticker:
# Determine MIME type for sticker URI (PNG for both PNG and APNG)
sticker_mime_type = "image/png"
try:
if not hasattr(cog, 'session') or not cog.session:
raise ValueError("aiohttp session not found in cog for sticker download.")
async with cog.session.get(sticker_url, timeout=10) as response:
if response.status == 200:
image_bytes = await response.read()
# Stickers are typically image/png or image/webp (APNG often served as image/png)
mime_type = response.content_type or "image/png"
supported_sticker_mimes = ["image/png", "image/webp"]
clean_mime_type = mime_type.split(';')[0].lower()
if clean_mime_type in supported_sticker_mimes:
image_part = types.Part(inline_data=types.Blob(data=image_bytes, mime_type=clean_mime_type))
emoji_sticker_parts_to_add.append(image_part)
print(f"Added image part for sticker: {sticker_name} (MIME: {clean_mime_type})")
else:
print(f"Unsupported MIME type for sticker {sticker_name}: {clean_mime_type}. URL: {sticker_url}")
else:
print(f"Failed to download sticker {sticker_name} from {sticker_url}: Status {response.status}")
image_part = types.Part(file_data=types.FileData(mime_type=sticker_mime_type, file_uri=sticker_url))
emoji_sticker_parts_to_add.append(image_part)
print(f"Added FileData part for current sticker: {sticker_name} (MIME: {sticker_mime_type}, URL: {sticker_url})")
except Exception as e:
print(f"Error processing sticker image {sticker_name} from {sticker_url}: {e}")
print(f"Error creating FileData for current sticker {sticker_name} from {sticker_url}: {e}")
emoji_sticker_parts_to_add.append(types.Part(text=f"[System Note: Failed to attach sticker '{sticker_name}' as FileData]"))
elif sticker_format_str == "StickerFormatType.lottie":
lottie_info_part = types.Part(text=f"[Sticker is a Lottie animation: {sticker_name}, URL: {sticker_url}]")
emoji_sticker_parts_to_add.append(lottie_info_part)