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 (PDF, images, scanned pages) using Kreuzberg OCR. Call `extract_file("filename")` on attached files.`
4. **Coller le code ci-dessous**
5. **Save and Activate**

## Tool Code

```python
"""
Kreuzberg OCR Tool for OpenWebUI
Extracts text from scanned documents, images, and PDFs using Kreuzberg OCR.
Supports: PDF, JPEG, PNG, TIFF, DOCX, PPTX, XLSX, EPUB, etc.
"""

import os
import glob
import requests
import hashlib
import io
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: Image.Image) -> Image.Image:
        """Preprocess image for better OCR quality (grayscale + contrast)"""
        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: str) -> str:
        """
        Extract text from a file using Kreuzberg OCR.
        
        Args:
            filename: The filename shown in the chat (e.g., "document.pdf", "image.jpg")
                     or the full name like "UUID_name.pdf"
        
        Returns:
            Extracted text content with markdown formatting
        """
        try:
            file_path = self._find_file(filename)
            
            if not file_path:
                return f"❌ File not found: {filename}\n\nTip: 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(
                f"{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:
                        # Clean up the content
                        content = self._clean_content(content)
                        
                        # Truncate if too long (LLM context limit)
                        if len(content) > 15000:
                            content = content[:15000] + "\n\n... (document truncated - too long for context)"
                        
                        return f"📄 **{os.path.basename(file_path)}**\n\n{content}"
                    else:
                        return f"⚠️ No text extracted from {os.path.basename(file_path)}\n\nThe file may be empty, protected, or contain only images without selectable text."
                
                return f"⚠️ Unexpected response format from Kreuzberg: {str(result)[:500]}"
            
            elif response.status_code == 404:
                return f"❌ Kreuzberg endpoint not found. Is the server running at {self.kreuzberg_url}?"
            
            else:
                return f"❌ Kreuzberg error ({response.status_code}): {response.text[:500]}"
        
        except requests.Timeout:
            return f"⏱️ Timeout - File is too large or Kreuzberg is overloaded. Try a smaller file."
        
        except requests.ConnectionError:
            return f"❌ Cannot connect to Kreuzberg at {self.kreuzberg_url}. Is the server running?"
        
        except Exception as e:
            return f"❌ Error: {str(e)}"

    def list_recent_files(self, count: int = 10) -> str:
        """List recent uploaded files"""
        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 "📂 No files found in uploads directory."
            
            result = f"📂 **{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 += f"- `{name}` ({size_str})\n"
            
            result += "\n💡 Use `extract_file(\"filename\")` to OCR a file."
            return result
        
        except Exception as e:
            return f"❌ Error listing files: {str(e)}"

    def search_files(self, pattern: str) -> str:
        """Search for files matching a pattern"""
        try:
            # Search in uploads
            files = glob.glob(os.path.join(self.uploads_dir, f"*{pattern}*"))
            
            if not files and os.path.exists(self.host_uploads_dir):
                files = glob.glob(os.path.join(self.host_uploads_dir, f"*{pattern}*"))
            
            files = [f for f in files if os.path.isfile(f)]
            
            if not files:
                return f"🔍 No files found matching: {pattern}"
            
            result = f"🔍 **{len(files)} files found:**\n\n"
            for f in files[:20]:  # Limit to 20
                name = os.path.basename(f)
                size = self._format_size(os.path.getsize(f))
                result += f"- `{name}` ({size})\n"
            
            return result
        
        except Exception as e:
            return f"❌ Error searching files: {str(e)}"

    def _find_file(self, filename: str) -> str:
        """Find file in uploads directory"""
        # Clean up filename
        filename = filename.strip().strip('"').strip("'")
        
        # Try exact path
        if os.path.exists(filename):
            return filename
        
        # Try in uploads directory with exact name
        path = os.path.join(self.uploads_dir, filename)
        if os.path.exists(path):
            return path
        
        # Try host mount path
        path = os.path.join(self.host_uploads_dir, filename)
        if os.path.exists(path):
            return path
        
        # Try glob patterns
        for directory in [self.uploads_dir, self.host_uploads_dir]:
            if not os.path.exists(directory):
                continue
            
            # UUID prefix match
            pattern = os.path.join(directory, f"{filename}*")
            matches = glob.glob(pattern)
            if matches:
                return matches[0]
            
            # Filename contains match
            pattern = os.path.join(directory, f"*{filename}*")
            matches = glob.glob(pattern)
            if matches:
                # Prefer exact 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: str) -> str:
        """Determine MIME type from file extension"""
        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: str) -> str:
        """Clean up extracted content"""
        if not content:
            return ""
        
        # Remove excessive whitespace
        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
        
        # Join and limit consecutive empty lines
        result = '\n'.join(cleaned_lines)
        
        # Remove very long lines (>200 chars wrap)
        if len(result) > 500:
            # Check if content looks garbled (lots of short fragments)
            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 due to complex formatting or handwriting."
        
        return result

    def _format_size(self, size: int) -> str:
        """Format file size"""
        if size < 1024:
            return f"{size} B"
        elif size < 1024 * 1024:
            return f"{size / 1024:.1f} KB"
        else:
            return f"{size / (1024 * 1024):.1f} MB"


# OpenWebUI Tool Functions
def extract_file(filename: str) -> str:
    """
    Extract text from a file using Kreuzberg OCR.
    
    Args:
        filename: The filename or UUID from the chat (e.g., "document.pdf", "abc123_document.pdf")
    
    Returns:
        Extracted text with markdown formatting
    """
    tools = Tools()
    return tools.extract_file(filename)

def list_recent_files(count: int = 10) -> str:
    """List recent uploaded files"""
    tools = Tools()
    return tools.list_recent_files(count)

def search_files(pattern: str) -> str:
    """Search for files by name pattern"""
    tools = Tools()
    return tools.search_files(pattern)
```

## Usage Examples

After installing and activating the tool:

### Extract text from a file:
```
extract_file("document.pdf")
extract_file("9af3d665-998b-4577-b893-340cb1c28385_document.pdf")
extract_file("image.jpg")
```

### List recent files:
```
list_recent_files()
list_recent_files(20)
```

### Search files:
```
search_files("dnd")
search_files("polz")
```

---

*Updated 11 April 2026 - Added image preprocessing (grayscale + contrast) for better OCR quality*