""" title: Image OCR Filter 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 import re 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) -> 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 self._remove_spaces_between_cjk(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" or not isinstance( message.get("content"), list ): continue for c_index, content in enumerate(message["content"]): if content.get("type") != "image_url": continue url = content.get("image_url", {}).get("url") if url: return m_index, c_index, url return None def _remove_spaces_between_cjk(self, text: str) -> str: """ 移除CJK字符之间的空格 """ pattern = ( r"(?<=[\u4E00-\u9FFF\u3400-\u4DBF\U00020000-\U0002EBEF]) " r"(?=[\u4E00-\u9FFF\u3400-\u4DBF\U00020000-\U0002EBEF])" ) return re.sub(pattern, "", text) 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) 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"<img>\n{ocr_result}\n</img>", } body["messages"] = messages except Exception as e: if self.valves.status: await __event_emitter__( { "type": "status", "data": { "description": f"识别图像出错: {e}", "done": True, }, } ) return body