gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #568 from christianlouis/copilot/add-multi-language-ocr-support

Browse Source 
feat(ocr): per-pipeline language override for multi-language OCR
This commit is contained in:
Christian Krakau-Louis
2026-03-08 23:11:41 +01:00
committed by GitHub
parent 2e241dd5f3 0b291995b9
commit 6376b6d73b
9 changed files with 816 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/pipelines.py
+42 -1
View File
@@ -51,7 +51,48 @@ PIPELINE_STEP_TYPES: dict[str, dict[str, Any]] = {
"type": "boolean",
"default": False,
"description": "Always use cloud OCR even if the PDF already has embedded text.",
}
},
"ocr_language": {
"type": "select",
"default": "auto",
"description": (
"Language(s) used for OCR text extraction. Applies to Tesseract and EasyOCR "
"providers; Azure and Mistral perform auto-detection by default. "
"Use Tesseract codes such as 'eng', 'deu', or 'eng+deu' for multi-language "
"documents. 'auto' falls back to the global system setting."
),
"options": [
{"value": "auto", "label": "Auto (use system default)"},
{"value": "ara", "label": "Arabic"},
{"value": "chi_sim", "label": "Chinese (Simplified)"},
{"value": "chi_tra", "label": "Chinese (Traditional)"},
{"value": "ces", "label": "Czech"},
{"value": "dan", "label": "Danish"},
{"value": "nld", "label": "Dutch"},
{"value": "eng", "label": "English"},
{"value": "fin", "label": "Finnish"},
{"value": "fra", "label": "French"},
{"value": "deu", "label": "German"},
{"value": "ell", "label": "Greek"},
{"value": "heb", "label": "Hebrew"},
{"value": "hin", "label": "Hindi"},
{"value": "hun", "label": "Hungarian"},
{"value": "ita", "label": "Italian"},
{"value": "jpn", "label": "Japanese"},
{"value": "kor", "label": "Korean"},
{"value": "nor", "label": "Norwegian"},
{"value": "pol", "label": "Polish"},
{"value": "por", "label": "Portuguese"},
{"value": "ron", "label": "Romanian"},
{"value": "rus", "label": "Russian"},
{"value": "spa", "label": "Spanish"},
{"value": "swe", "label": "Swedish"},
{"value": "tha", "label": "Thai"},
{"value": "tur", "label": "Turkish"},
{"value": "ukr", "label": "Ukrainian"},
{"value": "vie", "label": "Vietnamese"},
],
},
},
},
"extract_metadata": {
app/tasks/process_document.py
+83 -4
View File
@@ -1,10 +1,14 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
import logging
import mimetypes
import os
import shutil
import uuid
from typing import TYPE_CHECKING
import pypdf # Upgraded from PyPDF2 to fix CVE-2023-36464
from pypdf.errors import PdfReadError
@@ -12,7 +16,7 @@ from pypdf.errors import PdfReadError
from app.celery_app import celery
from app.config import settings
from app.database import SessionLocal
from app.models import FileRecord
from app.models import FileRecord, Pipeline, PipelineStep
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
from app.tasks.process_with_ocr import process_with_ocr
from app.tasks.retry_config import BaseTaskWithRetry
@@ -20,9 +24,75 @@ from app.utils import get_unique_filepath_with_counter, hash_file, log_task_prog
from app.utils.step_manager import initialize_file_steps
from app.utils.text_quality import check_text_quality, detect_pdf_text_source
if TYPE_CHECKING:
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
def _get_pipeline_ocr_language(db: "Session", file_record: FileRecord, owner_id: str | None) -> str | None:
"""Look up the OCR language override from the file's pipeline OCR step config.
Resolution order:
1. Explicit pipeline assigned to the file (``file_record.pipeline_id``).
2. User's own default pipeline (``owner_id``, ``is_default=True``).
3. System default pipeline (``owner_id=NULL``, ``is_default=True``).
Returns the ``ocr_language`` value from the pipeline's OCR step config, or
``None`` when no override is configured.
"""
pipeline = None
if file_record.pipeline_id:
pipeline = db.query(Pipeline).filter(Pipeline.id == file_record.pipeline_id).first()
if pipeline is None and owner_id:
pipeline = (
db.query(Pipeline)
.filter(
Pipeline.owner_id == owner_id,
Pipeline.is_default.is_(True),
Pipeline.is_active.is_(True),
)
.first()
)
if pipeline is None:
pipeline = (
db.query(Pipeline)
.filter(
Pipeline.owner_id.is_(None),
Pipeline.is_default.is_(True),
Pipeline.is_active.is_(True),
)
.first()
)
if pipeline is None:
return None
ocr_step = (
db.query(PipelineStep)
.filter(
PipelineStep.pipeline_id == pipeline.id,
PipelineStep.step_type == "ocr",
PipelineStep.enabled.is_(True),
)
.first()
)
if ocr_step is None or not ocr_step.config:
return None
try:
step_config = json.loads(ocr_step.config)
lang = step_config.get("ocr_language")
# "auto" is treated as no override
return lang if lang and lang != "auto" else None
except Exception:
return None
@celery.task(base=BaseTaskWithRetry, bind=True)
def process_document(
self,
@@ -109,6 +179,7 @@ def process_document(
)
# Acquire DB session in the task
ocr_language: str | None = None # Pipeline OCR language override resolved inside DB session
with SessionLocal() as db:
# When file_id is provided, we are reprocessing an existing file.
# Skip the duplicate check and reuse the existing record.
@@ -305,6 +376,14 @@ def process_document(
new_record.local_filename = new_local_path
db.commit()
# Look up pipeline OCR language override before the session closes.
# This reads the OCR step config from the file's assigned pipeline (or
# the user/system default pipeline) so the language is available when
# dispatching process_with_ocr below.
ocr_language = _get_pipeline_ocr_language(db, new_record, owner_id)
if ocr_language:
logger.info(f"[{task_id}] Pipeline OCR language override: {ocr_language!r}")
# Store file_id before session closes to avoid DetachedInstanceError
file_id = new_record.id
@@ -334,7 +413,7 @@ def process_document(
"Queued for forced OCR processing",
file_id=file_id,
)
process_with_ocr.delay(new_filename, file_id)
process_with_ocr.delay(new_filename, file_id, language=ocr_language)
return {"file": new_local_path, "status": "Queued for forced OCR", "file_id": file_id}
# If the file is not a PDF, skip embedded text check and convert to PDF first
@@ -491,7 +570,7 @@ def process_document(
"Queued for OCR (text quality too low)",
file_id=file_id,
)
process_with_ocr.delay(new_filename, file_id, extracted_text)
process_with_ocr.delay(new_filename, file_id, extracted_text, language=ocr_language)
return {
"file": new_local_path,
"status": "Queued for OCR (poor embedded text quality)",
@@ -564,5 +643,5 @@ def process_document(
"Queued for OCR processing",
file_id=file_id,
)
process_with_ocr.delay(new_filename, file_id)
process_with_ocr.delay(new_filename, file_id, language=ocr_language)
return {"file": new_local_path, "status": "Queued for OCR", "file_id": file_id}
app/tasks/process_with_ocr.py
+19 -5
View File
@@ -17,7 +17,6 @@ task with a multi-engine OCR pipeline that:
import logging
import os
from typing import Optional
from app.celery_app import celery
from app.config import settings
@@ -33,7 +32,13 @@ logger = logging.getLogger(__name__)
@celery.task(base=OcrTaskWithRetry, bind=True)
def process_with_ocr(self, filename: str, file_id: Optional[int] = None, original_text: Optional[str] = None):
def process_with_ocr(
self,
filename: str,
file_id: int | None = None,
original_text: str | None = None,
language: str | None = None,
):
"""Run the configured OCR providers on *filename* and continue the pipeline.
When multiple OCR providers are configured the results are merged using the
@@ -47,6 +52,10 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
filename: Base name of the file inside ``<workdir>/tmp/``.
file_id: Optional database record ID passed through to downstream tasks.
original_text: Optional original embedded text for head-to-head comparison.
language: Optional Tesseract-style language code(s) (e.g. ``"eng+deu"``)
to override the global OCR language settings for this specific run.
Pass ``None`` or ``"auto"`` to use the global settings. This
enables per-pipeline language configuration.
"""
task_id = self.request.id
log_task_progress(
@@ -62,7 +71,7 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
if not os.path.exists(tmp_file_path):
raise FileNotFoundError(f"Local file not found: {tmp_file_path}")
providers = get_ocr_providers()
providers = get_ocr_providers(language=language)
provider_names = [p.name for p in providers]
logger.info(f"[{task_id}] Running {len(providers)} OCR provider(s): {provider_names}")
@@ -122,7 +131,12 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
# PDF with ocrmypdf to embed an invisible text layer so the output is
# selectable/searchable in PDF viewers.
if searchable_pdf_path is None:
lang = getattr(settings, "tesseract_language", None) or "eng"
# Use the per-call language override; fall back to global setting
embed_lang = (
language
if language and language != "auto"
else (getattr(settings, "tesseract_language", None) or "eng")
)
log_task_progress(
task_id,
"embed_text_layer",
@@ -130,7 +144,7 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
"Embedding searchable text layer into PDF",
file_id=file_id,
)
embedded = embed_text_layer(tmp_file_path, tmp_file_path, language=lang)
embedded = embed_text_layer(tmp_file_path, tmp_file_path, language=embed_lang)
if embedded:
searchable_pdf_path = tmp_file_path
log_task_progress(
app/utils/ocr_provider.py
+141 -5
View File
@@ -192,6 +192,95 @@ class OCRResult:
)
# ---------------------------------------------------------------------------
# Multi-language support
# ---------------------------------------------------------------------------
#: Canonical list of supported OCR languages for pipeline configuration.
#: Keys are display names; values are Tesseract language code(s).
#: Tesseract codes are used as the canonical format because they are the most
#: widely applicable across self-hosted providers (Tesseract + ocrmypdf).
#: "auto" falls back to the global ``tesseract_language`` / ``easyocr_languages``
#: settings (i.e. no per-call override).
OCR_LANGUAGES: Dict[str, str] = {
"Auto (use system default)": "auto",
"Arabic": "ara",
"Chinese (Simplified)": "chi_sim",
"Chinese (Traditional)": "chi_tra",
"Czech": "ces",
"Danish": "dan",
"Dutch": "nld",
"English": "eng",
"Finnish": "fin",
"French": "fra",
"German": "deu",
"Greek": "ell",
"Hebrew": "heb",
"Hindi": "hin",
"Hungarian": "hun",
"Italian": "ita",
"Japanese": "jpn",
"Korean": "kor",
"Norwegian": "nor",
"Polish": "pol",
"Portuguese": "por",
"Romanian": "ron",
"Russian": "rus",
"Spanish": "spa",
"Swedish": "swe",
"Thai": "tha",
"Turkish": "tur",
"Ukrainian": "ukr",
"Vietnamese": "vie",
}
#: Mapping from Tesseract language codes to EasyOCR language codes.
#: Used when ``TesseractOCRProvider``-style codes are specified but EasyOCR is
#: the active provider. Codes not present in this map are passed through as-is
#: (EasyOCR accepts its own ISO 639-1 codes such as ``"en"`` or ``"de"``).
TESSERACT_TO_EASYOCR: Dict[str, str] = {
"ara": "ar",
"ces": "cs",
"chi_sim": "ch_sim",
"chi_tra": "ch_tra",
"dan": "da",
"deu": "de",
"ell": "el",
"eng": "en",
"fin": "fi",
"fra": "fr",
"heb": "he",
"hin": "hi",
"hun": "hu",
"ita": "it",
"jpn": "ja",
"kor": "ko",
"nld": "nl",
"nor": "no",
"pol": "pl",
"por": "pt",
"ron": "ro",
"rus": "ru",
"spa": "es",
"swe": "sv",
"tha": "th",
"tur": "tr",
"ukr": "uk",
"vie": "vi",
}
def _tesseract_codes_to_easyocr(tesseract_lang: str) -> List[str]:
"""Convert a Tesseract language string (e.g. ``"eng+deu"``) to a list of
EasyOCR language codes (e.g. ``["en", "de"]``).
Unknown codes are passed through unchanged, so native EasyOCR codes such
as ``"en"`` also work transparently.
"""
codes = [part.strip() for part in tesseract_lang.split("+") if part.strip()]
return [TESSERACT_TO_EASYOCR.get(code, code) for code in codes]
class OCRProvider(ABC):
"""Abstract base class for OCR providers.
@@ -290,10 +379,24 @@ class TesseractOCRProvider(OCRProvider):
- ``tesseract_cmd`` – path to the ``tesseract`` binary (optional).
- ``tesseract_language`` – Tesseract language code(s), e.g. ``"eng"`` or
``"eng+deu"`` (default: ``"eng"``).
The optional *language* constructor argument overrides the global
``tesseract_language`` setting for this specific provider instance, enabling
per-pipeline language configuration.
"""
name = "tesseract"
def __init__(self, language: Optional[str] = None) -> None:
"""Initialise the Tesseract provider.
Args:
language: Optional Tesseract language code(s) to use instead of the
global ``tesseract_language`` setting (e.g. ``"eng+deu"``).
Pass ``None`` or ``"auto"`` to use the global setting.
"""
self._language_override: Optional[str] = language if language and language != "auto" else None
def process(self, file_path: str) -> OCRResult:
try:
import pytesseract
@@ -308,7 +411,7 @@ class TesseractOCRProvider(OCRProvider):
if tesseract_cmd:
pytesseract.pytesseract.tesseract_cmd = tesseract_cmd
lang = getattr(settings, "tesseract_language", None) or "eng"
lang = self._language_override or getattr(settings, "tesseract_language", None) or "eng"
# Ensure language data files are present; attempt download if missing.
from app.utils.ocr_language_manager import ensure_tesseract_languages # noqa: PLC0415
@@ -349,10 +452,26 @@ class EasyOCRProvider(OCRProvider):
- ``easyocr_languages`` – comma-separated list of language codes
(default: ``"en"``).
- ``easyocr_gpu`` – whether to use GPU acceleration (default: ``False``).
The optional *language* constructor argument accepts a Tesseract-style
language string (e.g. ``"eng+deu"``) which is automatically translated to
EasyOCR codes (e.g. ``["en", "de"]``), overriding the global
``easyocr_languages`` setting for this provider instance.
"""
name = "easyocr"
def __init__(self, language: Optional[str] = None) -> None:
"""Initialise the EasyOCR provider.
Args:
language: Optional Tesseract-style language code(s) (e.g. ``"eng+deu"``)
or a comma-separated EasyOCR language list (e.g. ``"en,de"``).
Pass ``None`` or ``"auto"`` to use the global ``easyocr_languages``
setting.
"""
self._language_override: Optional[str] = language if language and language != "auto" else None
def process(self, file_path: str) -> OCRResult:
try:
import easyocr
@@ -363,8 +482,12 @@ class EasyOCRProvider(OCRProvider):
"Install them with: pip install easyocr pdf2image"
) from exc
lang_str = getattr(settings, "easyocr_languages", None) or "en"
langs = [lang.strip() for lang in lang_str.split(",") if lang.strip()]
if self._language_override:
# Convert Tesseract-style codes to EasyOCR codes
langs = _tesseract_codes_to_easyocr(self._language_override)
else:
lang_str = getattr(settings, "easyocr_languages", None) or "en"
langs = [lang.strip() for lang in lang_str.split(",") if lang.strip()]
gpu = getattr(settings, "easyocr_gpu", False)
logger.info(f"[EasyOCR] Processing {os.path.basename(file_path)} (langs={langs}, gpu={gpu})")
@@ -679,23 +802,36 @@ KNOWN_OCR_PROVIDERS: List[str] = sorted(_PROVIDER_MAP.keys())
MAX_OCR_TEXT_FOR_AI_MERGE = 4000
def get_ocr_providers() -> List[OCRProvider]:
def get_ocr_providers(language: Optional[str] = None) -> List[OCRProvider]:
"""Return a list of configured OCR provider instances.
Reads ``settings.ocr_providers`` (comma-separated provider names) and
returns one instantiated provider per entry. Falls back to ``["azure"]``
when the setting is absent.
Args:
language: Optional Tesseract-style language code(s) (e.g. ``"eng+deu"``)
to override the global language settings for providers that support
per-call language configuration (Tesseract and EasyOCR). Pass
``None`` or ``"auto"`` to use the global settings.
"""
raw = getattr(settings, "ocr_providers", None) or "azure"
provider_names = [name.strip().lower() for name in raw.split(",") if name.strip()]
# Normalise "auto" to None so providers fall back to global settings
effective_language = language if language and language != "auto" else None
providers: List[OCRProvider] = []
for name in provider_names:
cls = _PROVIDER_MAP.get(name)
if cls is None:
logger.warning(f"Unknown OCR provider '{name}' in OCR_PROVIDERS – skipping.")
continue
providers.append(cls())
# Pass language override to providers that support per-call language config
if effective_language is not None and name in ("tesseract", "easyocr"):
providers.append(cls(language=effective_language))
else:
providers.append(cls())
logger.debug(f"Registered OCR provider: {name}")
if not providers:
docs/API.md
+29 -3
View File
@@ -1788,12 +1788,27 @@ Returns the catalogue of built-in step types.
"label": "OCR Processing",
"description": "Extract text using Azure Document Intelligence or local Tesseract.",
"config_schema": {
"force_cloud_ocr": { "type": "boolean", "default": false }
"force_cloud_ocr": { "type": "boolean", "default": false },
"ocr_language": {
"type": "select",
"default": "auto",
"description": "Language(s) for OCR. Overrides the global setting for Tesseract/EasyOCR. Azure/Mistral auto-detect.",
"options": [
{ "value": "auto", "label": "Auto (use system default)" },
{ "value": "eng", "label": "English" },
{ "value": "deu", "label": "German" },
{ "value": "fra", "label": "French" },
{ "value": "spa", "label": "Spanish" },
"..."
]
}
}
}
}
```
The `ocr_language` field accepts Tesseract language codes (e.g. `"eng"`, `"deu"`, `"eng+deu"` for multi-language) or `"auto"` to fall back to the global system setting. The full list of 28 supported language codes is returned by the step-types endpoint.
### List pipelines
```bash
@@ -1886,12 +1901,23 @@ Content-Type: application/json
{
"step_type": "ocr",
"label": "Cloud OCR",
"config": { "force_cloud_ocr": true },
"label": "German OCR",
"config": { "force_cloud_ocr": false, "ocr_language": "deu" },
"enabled": true
}
```
Multi-language (Tesseract `+`-separated codes):
```bash
{
"step_type": "ocr",
"config": { "ocr_language": "eng+deu" }
}
```
Use `"ocr_language": "auto"` (or omit the field) to fall back to the global system language setting.
### Update step
```bash
docs/UserGuide.md
+38 -1
View File
@@ -519,13 +519,50 @@ Processing pipelines let you define exactly what happens to your documents when
|-----------|-------------|
| `convert_to_pdf` | Convert non-PDF files to PDF using Gotenberg |
| `check_duplicates` | Detect duplicate files by content hash |
| `ocr` | Extract text with Azure Document Intelligence or local Tesseract |
| `ocr` | Extract text with OCR (supports multi-language configuration, see below) |
| `extract_metadata` | Extract structured metadata (type, sender, tags) with AI |
| `embed_metadata` | Write extracted metadata into the PDF document properties |
| `compute_embedding` | Compute semantic embeddings for similarity search |
| `send_to_destinations` | Upload the processed document to all configured storage destinations |
| `classify` | Classify the document type with AI |
#### OCR step options
The `ocr` step supports two optional configuration fields:
| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `force_cloud_ocr` | boolean | `false` | Always run cloud OCR even if the PDF already has embedded text |
| `ocr_language` | string | `"auto"` | Language(s) to use for OCR text extraction (see below) |
**`ocr_language` — per-pipeline language override**
This option enables manual language control per pipeline, overriding the global Tesseract/EasyOCR language settings for all documents processed by that pipeline. The following values are supported (28 languages total):
| Value | Language | Value | Language |
|-------|----------|-------|----------|
| `auto` | Auto (use system default) | `jpn` | Japanese |
| `ara` | Arabic | `kor` | Korean |
| `chi_sim` | Chinese (Simplified) | `nor` | Norwegian |
| `chi_tra` | Chinese (Traditional) | `pol` | Polish |
| `ces` | Czech | `por` | Portuguese |
| `dan` | Danish | `ron` | Romanian |
| `nld` | Dutch | `rus` | Russian |
| `eng` | English | `spa` | Spanish |
| `fin` | Finnish | `swe` | Swedish |
| `fra` | French | `tha` | Thai |
| `deu` | German | `tur` | Turkish |
| `ell` | Greek | `ukr` | Ukrainian |
| `heb` | Hebrew | `vie` | Vietnamese |
| `hin` | Hindi | | |
| `hun` | Hungarian | | |
| `ita` | Italian | | |
> **Notes:**
> - The language override applies to **Tesseract** and **EasyOCR** providers. **Azure Document Intelligence** and **Mistral OCR** perform automatic language detection regardless of this setting.
> - For multi-language documents with Tesseract, combine codes with `+`, e.g. `eng+deu`.
> - Setting `ocr_language` to `auto` or leaving it unset uses the global `TESSERACT_LANGUAGE` / `EASYOCR_LANGUAGES` environment variables.
### Assigning a pipeline to a file
You can assign (or change) the pipeline for an individual document via the file detail page or the API:
frontend/templates/pipelines.html
+47
View File
@@ -404,6 +404,53 @@
</label>
</template>
<!-- ocr_language (only shown for ocr step) -->
<template x-if="stepModal.form.step_type === 'ocr'">
<div>
<label for="ocrLanguage" class="block text-sm font-medium text-gray-700 dark:text-gray-300 mb-1">
OCR Language
</label>
<select
id="ocrLanguage"
x-model="stepModal.form.config.ocr_language"
class="w-full border border-gray-300 dark:border-gray-600 rounded-md px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-blue-400 dark:bg-gray-700 dark:text-white"
>
<option value="auto">Auto (use system default)</option>
<option value="ara">Arabic</option>
<option value="chi_sim">Chinese (Simplified)</option>
<option value="chi_tra">Chinese (Traditional)</option>
<option value="ces">Czech</option>
<option value="dan">Danish</option>
<option value="nld">Dutch</option>
<option value="eng">English</option>
<option value="fin">Finnish</option>
<option value="fra">French</option>
<option value="deu">German</option>
<option value="ell">Greek</option>
<option value="heb">Hebrew</option>
<option value="hin">Hindi</option>
<option value="hun">Hungarian</option>
<option value="ita">Italian</option>
<option value="jpn">Japanese</option>
<option value="kor">Korean</option>
<option value="nor">Norwegian</option>
<option value="pol">Polish</option>
<option value="por">Portuguese</option>
<option value="ron">Romanian</option>
<option value="rus">Russian</option>
<option value="spa">Spanish</option>
<option value="swe">Swedish</option>
<option value="tha">Thai</option>
<option value="tur">Turkish</option>
<option value="ukr">Ukrainian</option>
<option value="vie">Vietnamese</option>
</select>
<p class="mt-1 text-xs text-gray-400">
Overrides the global language setting for Tesseract/EasyOCR. Azure and Mistral auto-detect the language.
</p>
</div>
</template>
<!-- Enabled -->
<label class="inline-flex items-center gap-2 cursor-pointer">
<input
tests/test_ocr_provider_coverage.py
+238
View File
@@ -1057,3 +1057,241 @@ class TestMergeOCRResults:
ms.openai_model = "gpt-4"
text, _, _ = merge_ocr_results([r1, r2], "doc.pdf")
assert text == "this is the longer text from tesseract engine"
# ---------------------------------------------------------------------------
# Multi-language OCR support
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestOCRLanguageConstants:
"""Tests for the OCR_LANGUAGES constant and TESSERACT_TO_EASYOCR mapping."""
def test_ocr_languages_has_20_plus_entries(self):
"""OCR_LANGUAGES contains at least 20 language options (excluding 'auto')."""
from app.utils.ocr_provider import OCR_LANGUAGES
language_entries = {k: v for k, v in OCR_LANGUAGES.items() if v != "auto"}
assert len(language_entries) >= 20, f"Expected ≥20 languages, got {len(language_entries)}"
def test_ocr_languages_includes_auto(self):
"""OCR_LANGUAGES includes 'auto' as the first option."""
from app.utils.ocr_provider import OCR_LANGUAGES
assert "auto" in OCR_LANGUAGES.values()
def test_ocr_languages_common_languages(self):
"""OCR_LANGUAGES includes the most common European and Asian languages."""
from app.utils.ocr_provider import OCR_LANGUAGES
expected_codes = {"eng", "deu", "fra", "spa", "ita", "por", "rus", "chi_sim", "jpn", "kor"}
all_codes = set(OCR_LANGUAGES.values())
missing = expected_codes - all_codes
assert not missing, f"Missing expected language codes: {missing}"
def test_tesseract_to_easyocr_mapping(self):
"""TESSERACT_TO_EASYOCR maps common Tesseract codes to EasyOCR codes."""
from app.utils.ocr_provider import TESSERACT_TO_EASYOCR
assert TESSERACT_TO_EASYOCR["eng"] == "en"
assert TESSERACT_TO_EASYOCR["deu"] == "de"
assert TESSERACT_TO_EASYOCR["fra"] == "fr"
assert TESSERACT_TO_EASYOCR["chi_sim"] == "ch_sim"
def test_tesseract_codes_to_easyocr_single(self):
"""_tesseract_codes_to_easyocr converts a single Tesseract code."""
from app.utils.ocr_provider import _tesseract_codes_to_easyocr
result = _tesseract_codes_to_easyocr("eng")
assert result == ["en"]
def test_tesseract_codes_to_easyocr_multi(self):
"""_tesseract_codes_to_easyocr splits '+'-separated Tesseract codes."""
from app.utils.ocr_provider import _tesseract_codes_to_easyocr
result = _tesseract_codes_to_easyocr("eng+deu")
assert result == ["en", "de"]
def test_tesseract_codes_to_easyocr_passthrough_unknown(self):
"""_tesseract_codes_to_easyocr passes through codes not in the mapping."""
from app.utils.ocr_provider import _tesseract_codes_to_easyocr
# EasyOCR-native codes are passed through unchanged
result = _tesseract_codes_to_easyocr("en")
assert result == ["en"]
@pytest.mark.unit
class TestTesseractLanguageOverride:
"""Tests for per-call language override in TesseractOCRProvider."""
def test_language_override_used_in_process(self, tmp_path):
"""Language override is used instead of global setting."""
pdf = _make_pdf(tmp_path)
provider = TesseractOCRProvider(language="deu")
mock_pytesseract = Mock()
mock_pytesseract.image_to_string.return_value = "Deutsches Text"
mock_pytesseract.pytesseract = Mock()
mock_pdf2image = Mock()
mock_pdf2image.convert_from_path.return_value = [Mock()]
with (
patch.dict(
sys.modules,
{"pytesseract": mock_pytesseract, "pdf2image": mock_pdf2image},
),
patch("app.utils.ocr_provider.settings") as ms,
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=[]),
):
ms.tesseract_cmd = None
ms.tesseract_language = "eng" # global setting; should be overridden
result = provider.process(pdf)
# Ensure image_to_string was called with the override language ("deu"), not global "eng"
mock_pytesseract.image_to_string.assert_called_once()
call_kwargs = mock_pytesseract.image_to_string.call_args
assert call_kwargs[1].get("lang") == "deu" or (call_kwargs[0] and call_kwargs[0][1] == "deu")
assert result.provider == "tesseract"
def test_auto_language_falls_back_to_global(self, tmp_path):
"""'auto' language override falls back to global tesseract_language setting."""
pdf = _make_pdf(tmp_path)
provider = TesseractOCRProvider(language="auto")
mock_pytesseract = Mock()
mock_pytesseract.image_to_string.return_value = ""
mock_pytesseract.pytesseract = Mock()
mock_pdf2image = Mock()
mock_pdf2image.convert_from_path.return_value = [Mock()]
with (
patch.dict(
sys.modules,
{"pytesseract": mock_pytesseract, "pdf2image": mock_pdf2image},
),
patch("app.utils.ocr_provider.settings") as ms,
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=[]),
):
ms.tesseract_cmd = None
ms.tesseract_language = "fra"
provider.process(pdf)
# Should use global setting "fra" since "auto" means no override
mock_pytesseract.image_to_string.assert_called_once()
call_kwargs = mock_pytesseract.image_to_string.call_args
lang_used = call_kwargs[1].get("lang") if call_kwargs[1] else call_kwargs[0][1]
assert lang_used == "fra"
def test_none_language_falls_back_to_global(self, tmp_path):
"""None language override falls back to global setting."""
pdf = _make_pdf(tmp_path)
provider = TesseractOCRProvider(language=None)
assert provider._language_override is None
@pytest.mark.unit
class TestEasyOCRLanguageOverride:
"""Tests for per-call language override in EasyOCRProvider."""
def test_language_override_converted_and_used(self, tmp_path):
"""Tesseract-style language override is converted to EasyOCR codes."""
pdf = _make_pdf(tmp_path)
provider = EasyOCRProvider(language="deu")
mock_reader = Mock()
mock_reader.readtext.return_value = ["Deutsches Text"]
mock_easyocr = Mock()
mock_easyocr.Reader.return_value = mock_reader
mock_pdf2image = Mock()
mock_pdf2image.convert_from_path.return_value = [Mock()]
with (
patch.dict(
sys.modules,
{"easyocr": mock_easyocr, "pdf2image": mock_pdf2image},
),
patch("app.utils.ocr_provider.settings") as ms,
):
ms.easyocr_languages = "en" # global; should be overridden
ms.easyocr_gpu = False
provider.process(pdf)
# Should call Reader with ["de"] (converted from "deu"), not global ["en"]
mock_easyocr.Reader.assert_called_once()
langs_arg = mock_easyocr.Reader.call_args[0][0]
assert langs_arg == ["de"]
def test_auto_language_uses_global_setting(self, tmp_path):
"""'auto' language override falls back to global easyocr_languages setting."""
pdf = _make_pdf(tmp_path)
provider = EasyOCRProvider(language="auto")
mock_reader = Mock()
mock_reader.readtext.return_value = []
mock_easyocr = Mock()
mock_easyocr.Reader.return_value = mock_reader
mock_pdf2image = Mock()
mock_pdf2image.convert_from_path.return_value = [Mock()]
with (
patch.dict(
sys.modules,
{"easyocr": mock_easyocr, "pdf2image": mock_pdf2image},
),
patch("app.utils.ocr_provider.settings") as ms,
):
ms.easyocr_languages = "fr,es"
ms.easyocr_gpu = False
provider.process(pdf)
langs_arg = mock_easyocr.Reader.call_args[0][0]
assert langs_arg == ["fr", "es"]
@pytest.mark.unit
class TestGetOCRProvidersWithLanguage:
"""Tests for get_ocr_providers(language=...) factory."""
def test_language_passed_to_tesseract_provider(self):
"""Language override is passed to TesseractOCRProvider."""
with patch("app.utils.ocr_provider.settings") as ms:
ms.ocr_providers = "tesseract"
providers = get_ocr_providers(language="deu")
assert len(providers) == 1
assert isinstance(providers[0], TesseractOCRProvider)
assert providers[0]._language_override == "deu"
def test_language_passed_to_easyocr_provider(self):
"""Language override is passed to EasyOCRProvider."""
with patch("app.utils.ocr_provider.settings") as ms:
ms.ocr_providers = "easyocr"
providers = get_ocr_providers(language="fra")
assert len(providers) == 1
assert isinstance(providers[0], EasyOCRProvider)
assert providers[0]._language_override == "fra"
def test_language_not_passed_to_azure(self):
"""Language override is NOT passed to AzureOCRProvider (it auto-detects)."""
with patch("app.utils.ocr_provider.settings") as ms:
ms.ocr_providers = "azure"
providers = get_ocr_providers(language="deu")
assert len(providers) == 1
assert isinstance(providers[0], AzureOCRProvider)
# AzureOCRProvider has no _language_override attribute
assert not hasattr(providers[0], "_language_override")
def test_auto_language_not_passed_as_override(self):
"""'auto' language is treated as no override for Tesseract."""
with patch("app.utils.ocr_provider.settings") as ms:
ms.ocr_providers = "tesseract"
providers = get_ocr_providers(language="auto")
assert providers[0]._language_override is None
def test_none_language_no_override(self):
"""None language results in no override."""
with patch("app.utils.ocr_provider.settings") as ms:
ms.ocr_providers = "tesseract"
providers = get_ocr_providers(language=None)
assert providers[0]._language_override is None
tests/test_process_document.py
+179
View File
@@ -842,3 +842,182 @@ startxref
mock_init_steps.assert_called_once()
called_file_id = mock_init_steps.call_args[0][1]
assert called_file_id == result["file_id"]
# ---------------------------------------------------------------------------
# _get_pipeline_ocr_language helper
# ---------------------------------------------------------------------------
@pytest.mark.unit
@pytest.mark.requires_db
def test_get_pipeline_ocr_language_returns_none_when_no_pipeline(db_session):
"""Returns None when no pipeline exists in the database."""
from app.tasks.process_document import _get_pipeline_ocr_language
# FileRecord with no pipeline_id
file_record = FileRecord(
filehash="abc123",
original_filename="test.pdf",
local_filename="/tmp/test.pdf",
file_size=1024,
mime_type="application/pdf",
is_duplicate=False,
)
db_session.add(file_record)
db_session.commit()
result = _get_pipeline_ocr_language(db_session, file_record, owner_id=None)
assert result is None
@pytest.mark.unit
@pytest.mark.requires_db
def test_get_pipeline_ocr_language_returns_language_from_system_default(db_session):
"""Returns ocr_language from the system default pipeline's OCR step config."""
import json
from app.models import Pipeline, PipelineStep
from app.tasks.process_document import _get_pipeline_ocr_language
# Create system default pipeline with OCR step configured to "deu"
pipeline = Pipeline(
owner_id=None,
name="System Default",
is_default=True,
is_active=True,
)
db_session.add(pipeline)
db_session.commit()
ocr_step = PipelineStep(
pipeline_id=pipeline.id,
position=0,
step_type="ocr",
config=json.dumps({"force_cloud_ocr": False, "ocr_language": "deu"}),
enabled=True,
)
db_session.add(ocr_step)
db_session.commit()
file_record = FileRecord(
filehash="def456",
original_filename="doc.pdf",
local_filename="/tmp/doc.pdf",
file_size=512,
mime_type="application/pdf",
is_duplicate=False,
)
db_session.add(file_record)
db_session.commit()
result = _get_pipeline_ocr_language(db_session, file_record, owner_id=None)
assert result == "deu"
@pytest.mark.unit
@pytest.mark.requires_db
def test_get_pipeline_ocr_language_auto_returns_none(db_session):
"""Returns None when ocr_language is 'auto' (should use global settings)."""
import json
from app.models import Pipeline, PipelineStep
from app.tasks.process_document import _get_pipeline_ocr_language
pipeline = Pipeline(
owner_id=None,
name="Auto Lang Pipeline",
is_default=True,
is_active=True,
)
db_session.add(pipeline)
db_session.commit()
ocr_step = PipelineStep(
pipeline_id=pipeline.id,
position=0,
step_type="ocr",
config=json.dumps({"ocr_language": "auto"}),
enabled=True,
)
db_session.add(ocr_step)
db_session.commit()
file_record = FileRecord(
filehash="ghi789",
original_filename="auto.pdf",
local_filename="/tmp/auto.pdf",
file_size=128,
mime_type="application/pdf",
is_duplicate=False,
)
db_session.add(file_record)
db_session.commit()
result = _get_pipeline_ocr_language(db_session, file_record, owner_id=None)
assert result is None
@pytest.mark.unit
@pytest.mark.requires_db
def test_get_pipeline_ocr_language_explicit_pipeline_takes_priority(db_session):
"""Explicit pipeline_id on file takes priority over system default pipeline."""
import json
from app.models import Pipeline, PipelineStep
from app.tasks.process_document import _get_pipeline_ocr_language
# System default pipeline with "eng"
sys_pipeline = Pipeline(
owner_id=None,
name="System Default",
is_default=True,
is_active=True,
)
db_session.add(sys_pipeline)
db_session.commit()
sys_step = PipelineStep(
pipeline_id=sys_pipeline.id,
position=0,
step_type="ocr",
config=json.dumps({"ocr_language": "eng"}),
enabled=True,
)
db_session.add(sys_step)
db_session.commit()
# Explicit pipeline with "fra"
explicit_pipeline = Pipeline(
owner_id="user1",
name="French Pipeline",
is_default=False,
is_active=True,
)
db_session.add(explicit_pipeline)
db_session.commit()
explicit_step = PipelineStep(
pipeline_id=explicit_pipeline.id,
position=0,
step_type="ocr",
config=json.dumps({"ocr_language": "fra"}),
enabled=True,
)
db_session.add(explicit_step)
db_session.commit()
file_record = FileRecord(
filehash="jkl012",
original_filename="french.pdf",
local_filename="/tmp/french.pdf",
file_size=256,
mime_type="application/pdf",
is_duplicate=False,
pipeline_id=explicit_pipeline.id,
)
db_session.add(file_record)
db_session.commit()
result = _get_pipeline_ocr_language(db_session, file_record, owner_id="user1")
assert result == "fra"