Folzo
e9d6de3513
- Remove complex docstrings that caused parsing issues - Use simple string concatenation instead of f-strings with special chars - Remove duplicate code - Keep functionality intact: single file, multi-file, list, search
337 lines
11 KiB
Markdown
337 lines
11 KiB
Markdown
# 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
|
|
|
|
```python
|
|
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* |