Effi/effi-kreuzberg/TOOL-OPENWEBUI.md

Kreuzberg OCR Tool - Enhanced Version

Installation

1. Admin Panel → Tools → Create Tool
2. Name: Kreuzberg OCR
3. Description: Extract text from documents using Kreuzberg OCR (PaddleOCR backend). Single: extract_file("file") or Multiple: extract_files("file1,file2")
4. Coller le code ci-dessous
5. Save and Activate

Tool Code

import os
import glob
import requests
import io
import json
from PIL import Image, ImageEnhance


class Tools:
    def __init__(self):
        self.kreuzberg_url = os.getenv("KREUZBERG_URL", "http://192.168.1.11:8300")
        self.uploads_dir = "/app/backend/data/uploads"
        self.host_uploads_dir = "/home/effi/effi/data/uploads"

    def _preprocess_image(self, image):
        if image.mode in ('RGBA', 'LA', 'P'):
            image = image.convert('RGBA').convert('RGB')
        elif image.mode != 'RGB':
            image = image.convert('RGB')
        image = image.convert('L')
        enhancer = ImageEnhance.Contrast(image)
        image = enhancer.enhance(2.0)
        return image

    def extract_file(self, filename):
        return self._extract_single_file(filename)

    def extract_files(self, filenames):
        filenames = filenames.strip()
        if not filenames:
            return "[ERROR] No filenames provided. Usage: extract_files(\"file1.pdf,file2.jpg\")"
        
        if filenames.startswith('['):
            try:
                file_list = json.loads(filenames)
            except json.JSONDecodeError:
                return "[ERROR] Invalid JSON array: " + filenames
        else:
            file_list = [f.strip() for f in filenames.split(',')]
        
        if not file_list:
            return "[ERROR] No files to process"
        
        if len(file_list) == 1:
            return self._extract_single_file(file_list[0])
        
        results = []
        total = len(file_list)
        
        for i, filename in enumerate(file_list, 1):
            filename = filename.strip()
            if not filename:
                continue
            if total > 1:
                results.append("--- Document " + str(i) + "/" + str(total) + ": " + filename + " ---")
            result = self._extract_single_file(filename)
            results.append(result)
        
        return "\n\n".join(results)

    def _extract_single_file(self, filename):
        try:
            file_path = self._find_file(filename)
            
            if not file_path:
                return "[ERROR] File not found: " + filename + ". Use list_recent_files() to see available files."
            
            mime_type = self._get_mime_type(file_path)
            ext = os.path.splitext(file_path)[1].lower()
            is_image = ext in ('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.tif', '.webp')
            
            if is_image:
                with open(file_path, 'rb') as f:
                    img = Image.open(f)
                    img = self._preprocess_image(img)
                    img_bytes = io.BytesIO()
                    img.save(img_bytes, format='PNG')
                    img_bytes.seek(0)
                files = {
                    'files': (os.path.basename(file_path) + '.png', img_bytes, 'image/png')
                }
            else:
                with open(file_path, 'rb') as f:
                    files = {
                        'files': (os.path.basename(file_path), f, mime_type)
                    }
            
            data = {
                'image_export_mode': 'inline',
                'ocr': 'true'
            }
            
            response = requests.post(
                self.kreuzberg_url + "/v1/convert/file",
                files=files,
                data=data,
                timeout=300
            )
            
            if response.status_code == 200:
                result = response.json()
                
                if 'document' in result:
                    content = result['document'].get('md_content', '')
                    if content and len(content.strip()) > 10:
                        content = self._clean_content(content)
                        
                        if len(content) > 15000:
                            content = content[:15000] + "\n\n... (document truncated)"
                        
                        return "FILE: " + os.path.basename(file_path) + "\n\n" + content
                    else:
                        return "[WARNING] No text extracted from " + os.path.basename(file_path) + ". File may be empty or contain only images."
                
                return "[WARNING] Unexpected response from Kreuzberg"
            
            elif response.status_code == 404:
                return "[ERROR] Kreuzberg endpoint not found at " + self.kreuzberg_url
            
            else:
                return "[ERROR] Kreuzberg error (" + str(response.status_code) + ")"
        
        except requests.Timeout:
            return "[ERROR] Timeout - file is too large or Kreuzberg is overloaded"
        
        except requests.ConnectionError:
            return "[ERROR] Cannot connect to Kreuzberg at " + self.kreuzberg_url
        
        except Exception as e:
            return "[ERROR] " + str(e)

    def list_recent_files(self, count=10):
        try:
            files = glob.glob(os.path.join(self.uploads_dir, "*"))
            
            if not files and os.path.exists(self.host_uploads_dir):
                files = glob.glob(os.path.join(self.host_uploads_dir, "*"))
            
            files = [f for f in files if os.path.isfile(f)]
            files = sorted(files, key=os.path.getmtime, reverse=True)[:count]
            
            if not files:
                return "[INFO] No files found in uploads directory."
            
            result = "[INFO] " + str(len(files)) + " recent files:\n\n"
            for f in files:
                name = os.path.basename(f)
                size = os.path.getsize(f)
                size_str = self._format_size(size)
                result += "- " + name + " (" + size_str + ")\n"
            
            result += "\nUse extract_file(\"filename\") to OCR a file."
            return result
        
        except Exception as e:
            return "[ERROR] " + str(e)

    def search_files(self, pattern):
        try:
            files = glob.glob(os.path.join(self.uploads_dir, "*" + pattern + "*"))
            
            if not files and os.path.exists(self.host_uploads_dir):
                files = glob.glob(os.path.join(self.host_uploads_dir, "*" + pattern + "*"))
            
            files = [f for f in files if os.path.isfile(f)]
            
            if not files:
                return "[INFO] No files found matching: " + pattern
            
            result = "[INFO] " + str(len(files)) + " files found:\n\n"
            for f in files[:20]:
                name = os.path.basename(f)
                size = self._format_size(os.path.getsize(f))
                result += "- " + name + " (" + size + ")\n"
            
            return result
        
        except Exception as e:
            return "[ERROR] " + str(e)

    def _find_file(self, filename):
        filename = filename.strip().strip('"').strip("'")
        
        if os.path.exists(filename):
            return filename
        
        path = os.path.join(self.uploads_dir, filename)
        if os.path.exists(path):
            return path
        
        path = os.path.join(self.host_uploads_dir, filename)
        if os.path.exists(path):
            return path
        
        for directory in [self.uploads_dir, self.host_uploads_dir]:
            if not os.path.exists(directory):
                continue
            
            pattern = os.path.join(directory, filename + "*")
            matches = glob.glob(pattern)
            if matches:
                return matches[0]
            
            pattern = os.path.join(directory, "*" + filename + "*")
            matches = glob.glob(pattern)
            if matches:
                for m in matches:
                    if filename in os.path.basename(m):
                        return m
                return matches[0]
        
        return None

    def _get_mime_type(self, file_path):
        ext = os.path.splitext(file_path)[1].lower()
        mime_types = {
            '.pdf': 'application/pdf',
            '.jpg': 'image/jpeg',
            '.jpeg': 'image/jpeg',
            '.png': 'image/png',
            '.gif': 'image/gif',
            '.tiff': 'image/tiff',
            '.tif': 'image/tiff',
            '.bmp': 'image/bmp',
            '.webp': 'image/webp',
            '.doc': 'application/msword',
            '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
            '.xls': 'application/vnd.ms-excel',
            '.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
            '.ppt': 'application/vnd.ms-powerpoint',
            '.pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
            '.epub': 'application/epub+zip',
            '.txt': 'text/plain',
            '.html': 'text/html',
            '.htm': 'text/html',
        }
        return mime_types.get(ext, 'application/octet-stream')

    def _clean_content(self, content):
        if not content:
            return ""
        
        lines = content.split('\n')
        cleaned_lines = []
        prev_empty = False
        
        for line in lines:
            stripped = line.strip()
            if not stripped:
                if not prev_empty:
                    cleaned_lines.append('')
                    prev_empty = True
            else:
                cleaned_lines.append(stripped)
                prev_empty = False
        
        result = '\n'.join(cleaned_lines)
        
        if len(result) > 500:
            short_words = sum(1 for w in result.split() if len(w) <= 3)
            total_words = len(result.split())
            if total_words > 50 and short_words / total_words > 0.3:
                result += "\n\n[WARNING] This text may contain OCR errors."
        
        return result

    def _format_size(self, size):
        if size < 1024:
            return str(size) + " B"
        elif size < 1024 * 1024:
            return str(size / 1024) + " KB"
        else:
            return str(size / (1024 * 1024)) + " MB"


def extract_file(filename):
    tools = Tools()
    return tools.extract_file(filename)

def extract_files(filenames):
    tools = Tools()
    return tools.extract_files(filenames)

def list_recent_files(count=10):
    tools = Tools()
    return tools.list_recent_files(count)

def search_files(pattern):
    tools = Tools()
    return tools.search_files(pattern)

Usage Examples

After installing and activating the tool:

Extract text from a single file:

extract_file("document.pdf")
extract_file("9af3d665-998b-4577-b893-340cb1c28385_document.pdf")
extract_file("image.jpg")

Extract from multiple files (processed one by one):

extract_files("doc1.pdf,doc2.jpg")
extract_files('["page1.jpg","page2.jpg","page3.jpg"]')

List recent files:

list_recent_files()
list_recent_files(20)

Search files:

search_files("dnd")
search_files("polz")

---

Updated 11 April 2026 - Added PaddleOCR backend + multi-file processing