""" title: Image OCR Filter (Use Tika Server) author: EntropyYue author_url: https://github.com/EntropyYue funding_url: https://openwebui.com/f/xiaopa233/image_ocr_filter version: 0.1 """ from pydantic import BaseModel, Field from typing import Callable, Awaitable, Any, Optional, Tuple, List, Dict import httpx import base64 class Filter: class Valves(BaseModel): priority: int = Field( default=0, description="Priority level for filtering operations." ) TIKA_URL: str = Field(default="http://tika:9998", description="Tika Server URL") LANGUAGE: str = Field(default="eng", description="tesseract格式的语言名称") status: bool = Field(default=False) def __init__(self): self.valves = self.Valves() async def _perform_ocr( self, image_base64: str, event_emitter: Callable[[Any], Awaitable[None]] ) -> str: """ 执行OCR操作，通过Tika Server识别图像中的文字。 """ image_bytes = self._extract_image_bytes(image_base64) headers = { "Accept": "text/plain", "Content-Type": "application/octet-stream", "X-Tika-OCRLanguage": self.valves.LANGUAGE, } async with httpx.AsyncClient() as client: response = await client.put( f"{self.valves.TIKA_URL}/tika", headers=headers, content=image_bytes ) if response.status_code != 200: raise Exception( f"Tika Server 错误: 状态码 {response.status_code}，响应: {response.text}" ) ocr_result = response.text.strip() return ocr_result def _extract_image_bytes(self, image_base64: str) -> bytes: """ 从base64编码的图像字符串中提取图像数据（字节）。 期望格式："data:image/{format};base64,{data}" """ try: _, data = image_base64.split(",", 1) return base64.b64decode(data) except Exception as e: raise ValueError("无效的base64图像数据") from e def _find_image_in_messages( self, messages: List[Dict] ) -> Optional[Tuple[int, int, str]]: """ 在消息列表中查找图片信息，返回 (消息索引, 内容索引, 图片base64字符串)。 如果未找到，则返回 None。 """ for m_index, message in enumerate(messages): if message.get("role") == "user" and isinstance( message.get("content"), list ): for c_index, content in enumerate(message["content"]): if content.get("type") == "image_url": image_info = content.get("image_url", {}) if "url" in image_info: return m_index, c_index, image_info["url"] return None async def inlet( self, body: dict, __event_emitter__: Callable[[Any], Awaitable[None]], __user__: Optional[dict] = None, __model__: Optional[dict] = None, ) -> dict: messages = body.get("messages", []) image_info = self._find_image_in_messages(messages) if not image_info: return body message_index, content_index, image_base64 = image_info try: ocr_result = await self._perform_ocr(image_base64, __event_emitter__) if not ocr_result: ocr_result = ( "图像识别成功，但是返回为空，可能是图像不包含文本或文本不可辨认" ) if self.valves.status: await __event_emitter__( { "type": "status", "data": { "description": "识别图像成功", "done": True, }, } ) messages[message_index]["content"][content_index] = { "type": "text", "text": f"用户上传了一张图片，以下是图片转文字的结果：\n\n{ocr_result}", } body["messages"] = messages except Exception as e: if self.valves.status: await __event_emitter__( { "type": "status", "data": { "description": f"识别图像出错: {e}", "done": True, }, } ) return body