9.3 KiB
9.3 KiB
Kreuzberg OCR Tool
Installation
- Admin Panel → Tools → Create Tool
- Name:
Kreuzberg OCR - Description:
Extract text from documents using Kreuzberg OCR. Usage: extract_file("filename") or extract_files("file1,file2") - Coller le code ci-dessous
- Save and Activate
Tool Code
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:
file_content = f.read()
files = {
'files': (os.path.basename(file_path), io.BytesIO(file_content), 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