""" title: Image URL fetcher description: Filter function which detects image url(s) in chat messages and fetches/attaches the image(s) to the chat message author: artificialidiot1337 version: 0.1 """ from pydantic import BaseModel, Field from typing import Optional import logging import re import urllib.request import base64 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def extract_image_urls(text): image_extensions = ( ".png", ".jpg", ".jpeg", ".bmp", ".webp", ".tiff", ".svg", ) url_pattern = re.compile(r"https?://\S+") urls = url_pattern.findall(text) return [url for url in urls if url.lower().endswith(image_extensions)] def image_url_to_base64(url): req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req) as response: image_data = response.read() return base64.b64encode(image_data).decode("utf-8") class Filter: class Valves(BaseModel): enabled: bool = Field( default=True, description="Whether to automatically fetch image URLs and attach them to messages.", ) pass class UserValves(BaseModel): enabled: bool = Field( default=True, description="Whether to automatically fetch image URLs and attach them to your messages.", ) pass def __init__(self): self.valves = self.Valves() pass def inlet(self, body: dict, __user__: Optional[dict] = None) -> dict: if __user__.get("role", "admin") in ["user", "admin"]: if not __user__["valves"].enabled: return body messages = body.get("messages", []) if len(messages) == 0: return body current_message = messages[-1] message_content = current_message["content"] if isinstance(message_content, str): message_content = [{"type": "text", "text": message_content}] image_urls = [] for content in message_content: if content["type"] != "text": continue image_urls += extract_image_urls(content["text"]) if len(image_urls) > 0: for image_url in image_urls: logger.info(f"fetching detected image url: {image_url}") message_content.append( { "type": "image_url", "image_url": {"url": image_url_to_base64(image_url)}, } ) current_message["content"] = message_content return body def outlet(self, body: dict, __user__: Optional[dict] = None) -> dict: return body