# Kreuzberg OCR Tool ## Installation 1. **Admin Panel** → **Tools** → **Create Tool** 2. **Name**: `Kreuzberg OCR` 3. **Description**: `Extract text from documents using Kreuzberg OCR. Usage: extract_file("filename") or extract_files("file1,file2")` 4. **Coller le code ci-dessous** 5. **Save and Activate** ## Tool Code ```python import os import glob import requests import io import json import mimetypes 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 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: available = self._list_files() return "[ERROR] File not found: " + filename + "\n\nAvailable files:\n" + available mime_type, _ = mimetypes.guess_type(file_path) mime_type = mime_type or 'application/octet-stream' 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... (truncated)" return "FILE: " + os.path.basename(file_path) + "\n\n" + content else: return "[WARNING] No text extracted from " + os.path.basename(file_path) 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" 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): available = self._list_files(count) return available def search_files(self, pattern): try: search_dirs = [d for d in [self.uploads_dir, self.host_uploads_dir] if os.path.exists(d)] files = [] for directory in search_dirs: files.extend(glob.glob(os.path.join(directory, "*" + 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 _list_files(self, count=10): try: files = [] search_dirs = [d for d in [self.uploads_dir, self.host_uploads_dir] if os.path.exists(d)] for directory in search_dirs: files.extend(glob.glob(os.path.join(directory, "*"))) files = [f for f in files if os.path.isfile(f)] files.sort(key=os.path.getmtime, reverse=True) if not files: return "[INFO] No files found in uploads directory." result = "[INFO] " + str(len(files)) + " recent files:\n\n" for f in files[:count]: 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 _find_file(self, filename): filename = filename.strip().strip('"').strip("'") if os.path.exists(filename): return filename search_dirs = [d for d in [self.uploads_dir, self.host_uploads_dir] if os.path.exists(d)] for directory in search_dirs: path = os.path.join(directory, filename) if os.path.exists(path): return path for directory in search_dirs: all_files = glob.glob(os.path.join(directory, "*")) for f in all_files: basename = os.path.basename(f).lower() search_name = filename.lower() if search_name in basename or basename.startswith(search_name): return f parts = search_name.replace('.jpeg', '').replace('.jpg', '').replace('.png', '').split() for part in parts: if len(part) > 3 and part in basename: return f return None 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 return '\n'.join(cleaned_lines) def _format_size(self, size): if size < 1024: return str(size) + " B" elif size < 1024 * 1024: return str(int(size / 1024)) + " KB" else: return str(int(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 ``` extract_file("document.pdf") extract_file("WhatsApp Image 2026-04-05 at 10.20.56.jpeg") extract_files("page1.jpg,page2.jpg,page3.jpg") list_recent_files() search_files("dnd") ``` --- *Updated 11 April 2026 - Improved file search, no preprocessing (Kreuzberg handles it)* ## 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*