feat(ocr): auto-install Tesseract/EasyOCR languages from settings

- Add app/utils/ocr_language_manager.py: detects tessdata dir, downloads
  missing .traineddata files via wget/curl from tessdata_fast GitHub repo,
  pre-downloads EasyOCR models, exposes async background-thread helper
- TesseractOCRProvider.process() calls ensure_tesseract_languages() before
  running pytesseract; raises clear error if languages remain unavailable
- EasyOCRProvider.process() logs informational message when models download
- app/main.py: calls ensure_ocr_languages_async() at startup
- app/utils/settings_sync.py: triggers language re-check after every
  settings reload so UI changes take effect without container restart
- app/api/settings.py: adds POST /api/settings/install-ocr-languages
  endpoint for on-demand language installation from the admin UI
- Dockerfile: adds wget for runtime tessdata downloads
- docs/ConfigurationGuide.md: documents automatic language download
- tests/test_ocr_language_manager.py: 29 unit tests

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-24 16:28:14 +00:00
parent fb6988db01
commit 8a2a4dc2b3
8 changed files with 884 additions and 1 deletions
+2
View File
@@ -21,11 +21,13 @@ COPY --from=builder /usr/local/bin /usr/local/bin
# ghostscript required by ocrmypdf for PDF/PS operations # ghostscript required by ocrmypdf for PDF/PS operations
# poppler-utils provides pdfinfo/pdftoppm used by pdf2image # poppler-utils provides pdfinfo/pdftoppm used by pdf2image
# unpaper optional deskewing pre-processor used by ocrmypdf # unpaper optional deskewing pre-processor used by ocrmypdf
# wget used by ocr_language_manager to download tessdata files
RUN apt-get update && apt-get install -y --no-install-recommends \ RUN apt-get update && apt-get install -y --no-install-recommends \
tesseract-ocr \ tesseract-ocr \
ghostscript \ ghostscript \
poppler-utils \ poppler-utils \
unpaper \ unpaper \
wget \
&& apt-get clean && rm -rf /var/lib/apt/lists/* && apt-get clean && rm -rf /var/lib/apt/lists/*
# Copy application code # Copy application code
+41
View File
@@ -405,6 +405,47 @@ async def bulk_update_settings(updates: list[SettingUpdate], request: Request, d
} }
@router.post("/install-ocr-languages")
async def install_ocr_languages(request: Request, admin: AdminUser):
"""
Trigger on-demand installation of Tesseract language data files and
EasyOCR model downloads for the languages currently configured in the
application settings.
This endpoint is useful after changing ``tesseract_language`` or
``easyocr_languages`` so that the required data is available without
restarting the container. The download runs synchronously and may take
a few seconds (or minutes for large EasyOCR models).
Returns a summary of which languages are now available and which could
not be installed.
Admin only.
"""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings # noqa: PLC0415
try:
result = ensure_ocr_languages_from_settings()
tesseract_missing = result.get("tesseract_missing", [])
easyocr_failed = result.get("easyocr_failed", [])
all_ok = not tesseract_missing and not easyocr_failed
return {
"success": all_ok,
"tesseract_missing": tesseract_missing,
"easyocr_failed": easyocr_failed,
"message": (
"All configured OCR languages are available."
if all_ok
else f"Some languages could not be installed: tesseract={tesseract_missing}, easyocr={easyocr_failed}"
),
}
except Exception as e:
logger.error(f"Error during OCR language installation: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to install OCR language data",
)
@router.get("/{key}/history") @router.get("/{key}/history")
async def get_key_history(key: str, request: Request, db: DbSession, admin: AdminUser): async def get_key_history(key: str, request: Request, db: DbSession, admin: AdminUser):
""" """
+5
View File
@@ -69,6 +69,11 @@ async def lifespan(app: FastAPI):
finally: finally:
db.close() db.close()
# Ensure OCR language data is available (background download, non-blocking)
from app.utils.ocr_language_manager import ensure_ocr_languages_async
ensure_ocr_languages_async()
# Force settings dump to log for troubleshooting # Force settings dump to log for troubleshooting
from app.utils.config_validator import dump_all_settings from app.utils.config_validator import dump_all_settings
+341
View File
@@ -0,0 +1,341 @@
"""OCR language manager for DocuElevate.
Ensures that Tesseract language data files (``.traineddata``) and EasyOCR
model files are present for every language code configured in the
application settings.
**Tesseract** language data is downloaded on demand from the
``tessdata_fast`` GitHub repository
(``https://github.com/tesseract-ocr/tessdata_fast``). The data files are
written to the tessdata directory discovered at runtime (respects the
``TESSDATA_PREFIX`` environment variable and falls back to common system
paths).
**EasyOCR** models are downloaded via the library's built-in mechanism
instantiating ``easyocr.Reader([lang])`` triggers the download if the model
files are absent from ``~/.EasyOCR/model/``.
Both functions are idempotent: they skip languages whose data is already
present.
Typical call sites:
* Application startup (``app/main.py`` lifespan) runs in a background
thread so it does not delay HTTP server readiness.
* Celery worker startup (``app/celery_worker.py``) scheduled shortly
after the worker comes online.
* Settings reload (``app/utils/settings_sync.py``) triggered whenever the
``tesseract_language`` or ``easyocr_languages`` settings change.
* OCR provider ``process()`` methods last-chance check before actually
running OCR so a clear error is raised rather than a cryptic pytesseract
or easyocr exception.
"""
import logging
import os
import shutil
import subprocess
import threading
from typing import Optional
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Public constants
# ---------------------------------------------------------------------------
#: URL template for downloading Tesseract tessdata_fast language data files.
#: ``{lang}`` is replaced with the ISO 639-2 Tesseract language code
#: (e.g. ``eng``, ``deu``, ``fra``).
TESSDATA_FAST_BASE_URL = "https://github.com/tesseract-ocr/tessdata_fast/raw/main"
# ---------------------------------------------------------------------------
# Tesseract helpers
# ---------------------------------------------------------------------------
def get_tessdata_dir() -> Optional[str]:
"""Return the tessdata directory that Tesseract will use at runtime.
Resolution order:
1. ``TESSDATA_PREFIX`` environment variable (if it points to an existing
directory).
2. Common Debian/Ubuntu system paths (``/usr/share/tesseract-ocr/*/tessdata``).
3. ``/usr/share/tessdata`` and ``/usr/local/share/tessdata`` as fallback.
Returns:
Absolute path to the tessdata directory, or ``None`` when none of the
candidate paths exist.
"""
# 1. Honour explicit TESSDATA_PREFIX
tessdata_prefix = os.environ.get("TESSDATA_PREFIX")
if tessdata_prefix:
if os.path.isdir(tessdata_prefix):
return tessdata_prefix
logger.debug(f"TESSDATA_PREFIX={tessdata_prefix!r} is set but not a directory; ignoring")
# 2. Common Debian/Ubuntu APT install paths (ordered by preference)
candidates = [
"/usr/share/tesseract-ocr/5/tessdata",
"/usr/share/tesseract-ocr/4.00/tessdata",
"/usr/share/tessdata",
"/usr/local/share/tessdata",
]
for path in candidates:
if os.path.isdir(path):
return path
return None
def is_tesseract_language_available(lang_code: str) -> bool:
"""Return ``True`` if the ``<lang_code>.traineddata`` file exists.
Args:
lang_code: Tesseract language code, e.g. ``"eng"`` or ``"deu"``.
"""
tessdata_dir = get_tessdata_dir()
if not tessdata_dir:
return False
return os.path.isfile(os.path.join(tessdata_dir, f"{lang_code}.traineddata"))
def download_tesseract_language(lang_code: str) -> bool:
"""Download a Tesseract ``.traineddata`` file from the tessdata_fast repo.
Uses ``wget`` if available, otherwise falls back to ``curl``. The file
is written directly into the tessdata directory so Tesseract can find it
without any additional configuration.
Args:
lang_code: Tesseract language code, e.g. ``"eng"`` or ``"deu"``.
Returns:
``True`` on success, ``False`` when the download fails or neither
``wget`` nor ``curl`` is available.
"""
tessdata_dir = get_tessdata_dir()
if not tessdata_dir:
logger.warning(
"No tessdata directory found; cannot download language data for '%s'. "
"Set TESSDATA_PREFIX to a writable directory or install the Tesseract "
"language pack manually.",
lang_code,
)
return False
target_path = os.path.join(tessdata_dir, f"{lang_code}.traineddata")
url = f"{TESSDATA_FAST_BASE_URL}/{lang_code}.traineddata"
wget_bin = shutil.which("wget")
curl_bin = shutil.which("curl")
if wget_bin:
cmd = [wget_bin, "-q", "--show-progress", "-O", target_path, url]
elif curl_bin:
cmd = [curl_bin, "-fsSL", "-o", target_path, url]
else:
logger.warning(
"Neither wget nor curl is available; cannot download Tesseract language data for '%s'. "
"Install wget or curl, or add the language data manually to %s.",
lang_code,
tessdata_dir,
)
return False
logger.info("Downloading Tesseract language data for '%s' from %s", lang_code, url)
try:
proc = subprocess.run( # noqa: S603 # args are trusted paths/URLs
cmd,
capture_output=True,
text=True,
timeout=180,
check=False,
)
except subprocess.TimeoutExpired:
logger.warning("Timed out downloading Tesseract language data for '%s'", lang_code)
# Remove partial download to avoid a corrupt tessdata file
if os.path.exists(target_path):
try:
os.remove(target_path)
except OSError:
pass
return False
if proc.returncode != 0:
stderr_snippet = (proc.stderr or "").strip()[:300]
logger.warning(
"Failed to download Tesseract language data for '%s' (exit %d): %s",
lang_code,
proc.returncode,
stderr_snippet,
)
if os.path.exists(target_path):
try:
os.remove(target_path)
except OSError:
pass
return False
logger.info("Tesseract language data for '%s' downloaded successfully to %s", lang_code, target_path)
return True
def ensure_tesseract_languages(lang_str: str) -> list[str]:
"""Ensure Tesseract language data files are available for all languages in *lang_str*.
Language codes are separated by ``+`` (Tesseract convention), e.g.
``"eng+deu+fra"``. For each code the function checks whether the
``.traineddata`` file already exists; if not it attempts to download it
from the tessdata_fast repository.
Args:
lang_str: Tesseract-style language string, e.g. ``"eng"`` or ``"eng+deu"``.
Returns:
A list of language codes that are still unavailable after the download
attempt. An empty list means all languages are ready.
"""
lang_codes = [code.strip() for code in lang_str.split("+") if code.strip()]
still_missing: list[str] = []
for lang_code in lang_codes:
if is_tesseract_language_available(lang_code):
logger.debug("Tesseract language '%s' is already available", lang_code)
continue
logger.info("Tesseract language '%s' not found locally; attempting download", lang_code)
if not download_tesseract_language(lang_code):
still_missing.append(lang_code)
if still_missing:
logger.warning(
"The following Tesseract language(s) could not be installed: %s. "
"OCR quality may be degraded or processing may fail.",
", ".join(still_missing),
)
return still_missing
# ---------------------------------------------------------------------------
# EasyOCR helpers
# ---------------------------------------------------------------------------
def ensure_easyocr_models(lang_list: list[str]) -> list[str]:
"""Pre-download EasyOCR model files for *lang_list*.
Instantiates a temporary ``easyocr.Reader`` for each language to trigger
the built-in model-download mechanism. Models are cached in
``~/.EasyOCR/model/`` and reused on subsequent calls, so this function is
safe to call repeatedly.
This function is a no-op when ``easyocr`` is not installed.
Args:
lang_list: List of EasyOCR language codes, e.g. ``["en", "de"]``.
Returns:
A list of language codes whose models could not be downloaded. An
empty list means all models are ready.
"""
try:
import easyocr # noqa: PLC0415
except ImportError:
logger.debug("easyocr is not installed; skipping model pre-download")
return []
failed: list[str] = []
for lang in lang_list:
try:
logger.info("Pre-downloading EasyOCR model for '%s'", lang)
easyocr.Reader([lang], gpu=False, verbose=False)
logger.info("EasyOCR model for '%s' is ready", lang)
except Exception as exc: # noqa: BLE001
logger.warning("Failed to pre-download EasyOCR model for '%s': %s", lang, exc)
failed.append(lang)
if failed:
logger.warning(
"The following EasyOCR language model(s) could not be downloaded: %s. "
"OCR processing for these languages may fail.",
", ".join(failed),
)
return failed
# ---------------------------------------------------------------------------
# Combined entry point
# ---------------------------------------------------------------------------
def ensure_ocr_languages_from_settings() -> dict[str, list[str]]:
"""Ensure OCR language data is available for all configured languages.
Reads ``ocr_providers``, ``tesseract_language``, and ``easyocr_languages``
from the application settings and ensures the required language data is
present. Missing Tesseract tessdata files are downloaded automatically;
missing EasyOCR models are downloaded via the library's built-in mechanism.
This function is idempotent calling it multiple times is safe.
Returns:
A dict with keys:
* ``"tesseract_missing"`` Tesseract language codes that could not
be installed.
* ``"easyocr_failed"`` EasyOCR language codes whose models could
not be downloaded.
"""
# Import inside function to avoid circular imports at module load time
from app.config import settings # noqa: PLC0415
result: dict[str, list[str]] = {"tesseract_missing": [], "easyocr_failed": []}
providers_raw = getattr(settings, "ocr_providers", None) or "azure"
active_providers = {p.strip().lower() for p in providers_raw.split(",") if p.strip()}
if "tesseract" in active_providers:
lang_str = getattr(settings, "tesseract_language", None) or "eng"
logger.info("Ensuring Tesseract language data for configured languages: %s", lang_str)
result["tesseract_missing"] = ensure_tesseract_languages(lang_str)
if "easyocr" in active_providers:
lang_raw = getattr(settings, "easyocr_languages", None) or "en"
lang_list = [stripped for lang in lang_raw.split(",") if (stripped := lang.strip())]
logger.info("Ensuring EasyOCR models for configured languages: %s", lang_list)
result["easyocr_failed"] = ensure_easyocr_models(lang_list)
return result
def ensure_ocr_languages_async() -> None:
"""Run :func:`ensure_ocr_languages_from_settings` in a background thread.
This is the preferred startup call so that language downloads do not block
the HTTP server or Celery worker from becoming ready.
"""
thread = threading.Thread(
target=_run_ensure_languages,
name="ocr-language-manager",
daemon=True,
)
thread.start()
def _run_ensure_languages() -> None:
"""Internal target function for the background thread."""
try:
result = ensure_ocr_languages_from_settings()
missing = result.get("tesseract_missing", []) + result.get("easyocr_failed", [])
if missing:
logger.warning(
"OCR language setup incomplete the following language(s) are still unavailable: %s",
", ".join(missing),
)
else:
logger.info("OCR language setup complete all configured languages are available")
except Exception as exc: # noqa: BLE001
logger.error("Error during OCR language setup: %s", exc)
+16
View File
@@ -297,6 +297,19 @@ class TesseractOCRProvider(OCRProvider):
lang = getattr(settings, "tesseract_language", None) or "eng" lang = 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
missing = ensure_tesseract_languages(lang)
if missing:
raise RuntimeError(
f"[TesseractOCR] Required language data files are not available and could not be "
f"downloaded: {', '.join(missing)}. "
"Install the corresponding tesseract-ocr language packages "
"(e.g. apt-get install tesseract-ocr-deu) or ensure internet access so DocuElevate "
"can download them automatically."
)
logger.info(f"[TesseractOCR] Processing {os.path.basename(file_path)} (lang={lang})") logger.info(f"[TesseractOCR] Processing {os.path.basename(file_path)} (lang={lang})")
pages = convert_from_path(file_path, dpi=300) pages = convert_from_path(file_path, dpi=300)
texts: List[str] = [] texts: List[str] = []
@@ -342,6 +355,9 @@ class EasyOCRProvider(OCRProvider):
gpu = getattr(settings, "easyocr_gpu", False) gpu = getattr(settings, "easyocr_gpu", False)
logger.info(f"[EasyOCR] Processing {os.path.basename(file_path)} (langs={langs}, gpu={gpu})") logger.info(f"[EasyOCR] Processing {os.path.basename(file_path)} (langs={langs}, gpu={gpu})")
# easyocr.Reader automatically downloads missing models on first use;
# log a clear message so operators know a download may be in progress.
logger.info(f"[EasyOCR] Initialising reader for langs={langs} (models will be downloaded if absent)")
reader = easyocr.Reader(langs, gpu=gpu) reader = easyocr.Reader(langs, gpu=gpu)
pages = convert_from_path(file_path, dpi=300) pages = convert_from_path(file_path, dpi=300)
texts: List[str] = [] texts: List[str] = []
+18
View File
@@ -71,6 +71,17 @@ def notify_settings_updated() -> None:
except Exception as exc: except Exception as exc:
logger.warning(f"Could not reload in-process settings: {exc}") logger.warning(f"Could not reload in-process settings: {exc}")
# Re-check OCR language availability in the background whenever settings
# are updated. This ensures that if a user changes tesseract_language or
# easyocr_languages via the UI, the new language data is downloaded without
# requiring a container restart.
try:
from app.utils.ocr_language_manager import ensure_ocr_languages_async
ensure_ocr_languages_async()
except Exception as exc:
logger.warning(f"Could not schedule OCR language check: {exc}")
def register_settings_reload_signal() -> None: def register_settings_reload_signal() -> None:
""" """
@@ -96,6 +107,13 @@ def register_settings_reload_signal() -> None:
reload_settings_from_db(settings) reload_settings_from_db(settings)
_last_seen_version = current_version _last_seen_version = current_version
logger.info(f"Worker settings reloaded (version={current_version})") logger.info(f"Worker settings reloaded (version={current_version})")
# Ensure OCR language data is up to date after a settings reload.
try:
from app.utils.ocr_language_manager import ensure_ocr_languages_async
ensure_ocr_languages_async()
except Exception as lang_exc:
logger.warning(f"Could not schedule OCR language check on worker: {lang_exc}")
except Exception as exc: except Exception as exc:
logger.debug(f"Settings version check skipped: {exc}") logger.debug(f"Settings version check skipped: {exc}")
+10 -1
View File
@@ -485,7 +485,9 @@ For providers that do **not** embed a text layer, DocuElevate automatically runs
#### Tesseract (self-hosted) #### Tesseract (self-hosted)
Requires `tesseract-ocr` to be installed in the Docker image or on the host. The default Docker image ships with Tesseract. Requires `tesseract-ocr` to be installed in the Docker image or on the host. The default Docker image ships with Tesseract (English language data only).
**Automatic language data download**: DocuElevate automatically downloads missing Tesseract `.traineddata` files at startup using `wget` from the [tessdata_fast](https://github.com/tesseract-ocr/tessdata_fast) repository. No manual installation is required — simply set `TESSERACT_LANGUAGE` to the desired language codes and the data files are fetched on first start. The container must have outbound internet access for this to work.
| **Variable** | **Description** | **Default** | | **Variable** | **Description** | **Default** |
|------------------------|-----------------------------------------------------------------------------------|-------------| |------------------------|-----------------------------------------------------------------------------------|-------------|
@@ -497,10 +499,17 @@ OCR_PROVIDERS=tesseract
TESSERACT_LANGUAGE=eng+deu TESSERACT_LANGUAGE=eng+deu
``` ```
> **Language codes**: Use ISO 639-2 codes separated by `+`, e.g. `eng+deu+fra` for English + German + French.
> All codes supported by Tesseract are available. See the [tessdata repository](https://github.com/tesseract-ocr/tessdata_fast) for the full list.
> **No internet access?** Set `TESSDATA_PREFIX` to a writable directory and pre-populate it with the required `.traineddata` files. Alternatively, build a custom Docker image that installs the needed language packages via `apt-get install tesseract-ocr-<lang>`.
#### EasyOCR (self-hosted) #### EasyOCR (self-hosted)
Requires the `easyocr` Python package. Install it separately as it is not included in the base requirements. Requires the `easyocr` Python package. Install it separately as it is not included in the base requirements.
**Automatic model download**: EasyOCR model files are downloaded automatically on first use (or at startup) to `~/.EasyOCR/model/`. The container must have outbound internet access. Model download can take several minutes depending on the language.
| **Variable** | **Description** | **Default** | | **Variable** | **Description** | **Default** |
|-----------------------|------------------------------------------------------------------------|-------------| |-----------------------|------------------------------------------------------------------------|-------------|
| `EASYOCR_LANGUAGES` | Comma-separated EasyOCR language codes, e.g. `en,de,fr`. | `en,de` | | `EASYOCR_LANGUAGES` | Comma-separated EasyOCR language codes, e.g. `en,de,fr`. | `en,de` |
+451
View File
@@ -0,0 +1,451 @@
"""Unit tests for app.utils.ocr_language_manager.
Tests cover tessdata-directory detection, language availability checks,
download success/failure paths, and the combined orchestration function.
All external calls (filesystem, subprocess, easyocr, app.config) are mocked.
"""
import subprocess
from unittest.mock import MagicMock, Mock, patch
import pytest
@pytest.mark.unit
class TestGetTessdataDir:
"""Tests for get_tessdata_dir()."""
def test_returns_tessdata_prefix_when_valid(self, tmp_path):
"""TESSDATA_PREFIX env var pointing to an existing dir is used first."""
from app.utils.ocr_language_manager import get_tessdata_dir
with patch.dict("os.environ", {"TESSDATA_PREFIX": str(tmp_path)}):
result = get_tessdata_dir()
assert result == str(tmp_path)
def test_ignores_tessdata_prefix_when_not_a_dir(self, tmp_path):
"""TESSDATA_PREFIX pointing to a non-directory path is ignored."""
non_dir = str(tmp_path / "nonexistent")
from app.utils.ocr_language_manager import get_tessdata_dir
with (
patch.dict("os.environ", {"TESSDATA_PREFIX": non_dir}),
patch("os.path.isdir", side_effect=lambda p: False),
):
result = get_tessdata_dir()
assert result is None
def test_falls_back_to_system_path(self, tmp_path):
"""Returns first existing system candidate path when TESSDATA_PREFIX is absent."""
from app.utils.ocr_language_manager import get_tessdata_dir
def isdir_side_effect(path: str) -> bool:
return path == "/usr/share/tesseract-ocr/5/tessdata"
with (
patch.dict("os.environ", {}, clear=False),
patch("os.environ.get", side_effect=lambda k, d=None: None if k == "TESSDATA_PREFIX" else d),
patch("os.path.isdir", side_effect=isdir_side_effect),
):
result = get_tessdata_dir()
assert result == "/usr/share/tesseract-ocr/5/tessdata"
def test_returns_none_when_no_path_found(self):
"""Returns None when TESSDATA_PREFIX is absent and no system paths exist."""
from app.utils.ocr_language_manager import get_tessdata_dir
with (
patch("os.environ.get", return_value=None),
patch("os.path.isdir", return_value=False),
):
result = get_tessdata_dir()
assert result is None
@pytest.mark.unit
class TestIsTesseractLanguageAvailable:
"""Tests for is_tesseract_language_available()."""
def test_returns_true_when_file_exists(self, tmp_path):
"""Returns True when <lang>.traineddata is present in tessdata dir."""
(tmp_path / "eng.traineddata").write_bytes(b"fake")
from app.utils.ocr_language_manager import is_tesseract_language_available
with patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)):
assert is_tesseract_language_available("eng") is True
def test_returns_false_when_file_missing(self, tmp_path):
"""Returns False when the traineddata file is absent."""
from app.utils.ocr_language_manager import is_tesseract_language_available
with patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)):
assert is_tesseract_language_available("deu") is False
def test_returns_false_when_no_tessdata_dir(self):
"""Returns False when no tessdata directory is found."""
from app.utils.ocr_language_manager import is_tesseract_language_available
with patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=None):
assert is_tesseract_language_available("eng") is False
@pytest.mark.unit
class TestDownloadTesseractLanguage:
"""Tests for download_tesseract_language()."""
def test_returns_false_when_no_tessdata_dir(self):
"""Returns False when no tessdata directory is found."""
from app.utils.ocr_language_manager import download_tesseract_language
with patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=None):
result = download_tesseract_language("deu")
assert result is False
def test_returns_false_when_no_downloader(self, tmp_path):
"""Returns False when neither wget nor curl is on PATH."""
from app.utils.ocr_language_manager import download_tesseract_language
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("shutil.which", return_value=None),
):
result = download_tesseract_language("deu")
assert result is False
def test_uses_wget_when_available(self, tmp_path):
"""Uses wget when available and returns True on success."""
from app.utils.ocr_language_manager import download_tesseract_language
mock_proc = Mock()
mock_proc.returncode = 0
def which_side_effect(cmd):
if cmd == "wget":
return "/usr/bin/wget"
return None
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("shutil.which", side_effect=which_side_effect),
patch("subprocess.run", return_value=mock_proc) as mock_run,
):
result = download_tesseract_language("deu")
assert result is True
called_cmd = mock_run.call_args[0][0]
assert called_cmd[0] == "/usr/bin/wget"
assert any("deu.traineddata" in arg for arg in called_cmd)
def test_falls_back_to_curl(self, tmp_path):
"""Falls back to curl when wget is unavailable and returns True on success."""
from app.utils.ocr_language_manager import download_tesseract_language
mock_proc = Mock()
mock_proc.returncode = 0
def which_side_effect(cmd):
if cmd == "curl":
return "/usr/bin/curl"
return None
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("shutil.which", side_effect=which_side_effect),
patch("subprocess.run", return_value=mock_proc) as mock_run,
):
result = download_tesseract_language("fra")
assert result is True
called_cmd = mock_run.call_args[0][0]
assert called_cmd[0] == "/usr/bin/curl"
def test_returns_false_on_nonzero_exit(self, tmp_path):
"""Returns False and removes partial file when download exits non-zero."""
target = tmp_path / "deu.traineddata"
target.write_bytes(b"partial")
from app.utils.ocr_language_manager import download_tesseract_language
mock_proc = Mock()
mock_proc.returncode = 1
mock_proc.stderr = "404 Not Found"
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("shutil.which", return_value="/usr/bin/wget"),
patch("subprocess.run", return_value=mock_proc),
):
result = download_tesseract_language("deu")
assert result is False
assert not target.exists()
def test_returns_false_on_timeout(self, tmp_path):
"""Returns False and removes partial file when subprocess times out."""
target = tmp_path / "deu.traineddata"
target.write_bytes(b"partial")
from app.utils.ocr_language_manager import download_tesseract_language
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("shutil.which", return_value="/usr/bin/wget"),
patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="wget", timeout=180)),
):
result = download_tesseract_language("deu")
assert result is False
assert not target.exists()
@pytest.mark.unit
class TestEnsureTesseractLanguages:
"""Tests for ensure_tesseract_languages()."""
def test_skips_already_available_languages(self, tmp_path):
"""No download is attempted for languages that already have traineddata files."""
(tmp_path / "eng.traineddata").write_bytes(b"fake")
(tmp_path / "deu.traineddata").write_bytes(b"fake")
from app.utils.ocr_language_manager import ensure_tesseract_languages
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("app.utils.ocr_language_manager.download_tesseract_language") as mock_dl,
):
missing = ensure_tesseract_languages("eng+deu")
assert missing == []
mock_dl.assert_not_called()
def test_downloads_missing_language(self, tmp_path):
"""Attempts to download a missing language and returns empty list on success."""
(tmp_path / "eng.traineddata").write_bytes(b"fake")
from app.utils.ocr_language_manager import ensure_tesseract_languages
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("app.utils.ocr_language_manager.download_tesseract_language", return_value=True) as mock_dl,
):
missing = ensure_tesseract_languages("eng+deu")
assert missing == []
mock_dl.assert_called_once_with("deu")
def test_returns_still_missing_on_download_failure(self, tmp_path):
"""Returns failing language codes when download fails."""
from app.utils.ocr_language_manager import ensure_tesseract_languages
with (
patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)),
patch("app.utils.ocr_language_manager.download_tesseract_language", return_value=False),
):
missing = ensure_tesseract_languages("eng+fra")
assert set(missing) == {"eng", "fra"}
def test_single_language_no_plus(self, tmp_path):
"""Handles single language code without '+' separator."""
(tmp_path / "eng.traineddata").write_bytes(b"fake")
from app.utils.ocr_language_manager import ensure_tesseract_languages
with patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)):
missing = ensure_tesseract_languages("eng")
assert missing == []
def test_ignores_empty_tokens(self, tmp_path):
"""Handles malformed lang strings with extra '+' separators."""
(tmp_path / "eng.traineddata").write_bytes(b"fake")
from app.utils.ocr_language_manager import ensure_tesseract_languages
with patch("app.utils.ocr_language_manager.get_tessdata_dir", return_value=str(tmp_path)):
missing = ensure_tesseract_languages("+eng+")
assert missing == []
@pytest.mark.unit
class TestEnsureEasyOCRModels:
"""Tests for ensure_easyocr_models()."""
def test_no_op_when_easyocr_not_installed(self):
"""Returns empty list silently when easyocr cannot be imported."""
from app.utils.ocr_language_manager import ensure_easyocr_models
with patch("builtins.__import__", side_effect=ImportError("No module named 'easyocr'")):
result = ensure_easyocr_models(["en"])
# The function should handle import error gracefully
assert isinstance(result, list)
def test_returns_empty_list_on_success(self):
"""Returns empty list when all models download successfully."""
mock_reader_cls = MagicMock()
from app.utils.ocr_language_manager import ensure_easyocr_models
with patch.dict("sys.modules", {"easyocr": MagicMock(Reader=mock_reader_cls)}):
result = ensure_easyocr_models(["en", "de"])
assert result == []
def test_returns_failed_langs_on_error(self):
"""Returns failing language codes when easyocr.Reader raises an exception."""
mock_easyocr = MagicMock()
mock_easyocr.Reader.side_effect = RuntimeError("download failed")
from app.utils.ocr_language_manager import ensure_easyocr_models
with patch.dict("sys.modules", {"easyocr": mock_easyocr}):
result = ensure_easyocr_models(["fr"])
assert "fr" in result
def test_skips_successfully_downloads_partially(self):
"""Returns only the failing codes when some languages succeed and others fail."""
call_count = {"n": 0}
def reader_side_effect(langs, gpu=False, verbose=False):
call_count["n"] += 1
if langs == ["ja"]:
raise RuntimeError("model download failed")
mock_easyocr = MagicMock()
mock_easyocr.Reader.side_effect = reader_side_effect
from app.utils.ocr_language_manager import ensure_easyocr_models
with patch.dict("sys.modules", {"easyocr": mock_easyocr}):
result = ensure_easyocr_models(["en", "ja"])
assert result == ["ja"]
assert call_count["n"] == 2
@pytest.mark.unit
class TestEnsureOCRLanguagesFromSettings:
"""Tests for ensure_ocr_languages_from_settings()."""
def _mock_settings(self, providers="tesseract", tesseract_lang="eng", easyocr_langs="en"):
mock_settings = MagicMock()
mock_settings.ocr_providers = providers
mock_settings.tesseract_language = tesseract_lang
mock_settings.easyocr_languages = easyocr_langs
return mock_settings
def test_runs_tesseract_check_when_provider_active(self):
"""Calls ensure_tesseract_languages when 'tesseract' is in ocr_providers."""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings
mock_settings = self._mock_settings(providers="tesseract", tesseract_lang="eng+deu")
with (
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=[]) as mock_tess,
patch("app.utils.ocr_language_manager.ensure_easyocr_models", return_value=[]) as mock_easy,
patch("app.config.settings", mock_settings),
):
result = ensure_ocr_languages_from_settings()
mock_tess.assert_called_once_with("eng+deu")
mock_easy.assert_not_called()
assert result == {"tesseract_missing": [], "easyocr_failed": []}
def test_runs_easyocr_check_when_provider_active(self):
"""Calls ensure_easyocr_models when 'easyocr' is in ocr_providers."""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings
mock_settings = self._mock_settings(providers="easyocr", easyocr_langs="en,de")
with (
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=[]) as mock_tess,
patch("app.utils.ocr_language_manager.ensure_easyocr_models", return_value=[]) as mock_easy,
patch("app.config.settings", mock_settings),
):
result = ensure_ocr_languages_from_settings()
mock_tess.assert_not_called()
mock_easy.assert_called_once_with(["en", "de"])
assert result["easyocr_failed"] == []
def test_skips_both_when_only_azure(self):
"""Neither Tesseract nor EasyOCR checks run when only Azure is configured."""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings
mock_settings = self._mock_settings(providers="azure")
with (
patch("app.utils.ocr_language_manager.ensure_tesseract_languages") as mock_tess,
patch("app.utils.ocr_language_manager.ensure_easyocr_models") as mock_easy,
patch("app.config.settings", mock_settings),
):
result = ensure_ocr_languages_from_settings()
mock_tess.assert_not_called()
mock_easy.assert_not_called()
assert result == {"tesseract_missing": [], "easyocr_failed": []}
def test_runs_both_when_both_providers_active(self):
"""Both checks run when both 'tesseract' and 'easyocr' are in ocr_providers."""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings
mock_settings = self._mock_settings(providers="tesseract,easyocr")
with (
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=[]) as mock_tess,
patch("app.utils.ocr_language_manager.ensure_easyocr_models", return_value=[]) as mock_easy,
patch("app.config.settings", mock_settings),
):
ensure_ocr_languages_from_settings()
mock_tess.assert_called_once()
mock_easy.assert_called_once()
def test_returns_missing_and_failed_on_partial_failure(self):
"""Returns correct missing/failed lists on partial failures."""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings
mock_settings = self._mock_settings(providers="tesseract,easyocr")
with (
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=["deu"]),
patch("app.utils.ocr_language_manager.ensure_easyocr_models", return_value=["ja"]),
patch("app.config.settings", mock_settings),
):
result = ensure_ocr_languages_from_settings()
assert result["tesseract_missing"] == ["deu"]
assert result["easyocr_failed"] == ["ja"]
def test_uses_defaults_when_settings_are_none(self):
"""Falls back to 'eng' / 'en' when language settings are None."""
from app.utils.ocr_language_manager import ensure_ocr_languages_from_settings
mock_settings = MagicMock()
mock_settings.ocr_providers = "tesseract,easyocr"
mock_settings.tesseract_language = None
mock_settings.easyocr_languages = None
with (
patch("app.utils.ocr_language_manager.ensure_tesseract_languages", return_value=[]) as mock_tess,
patch("app.utils.ocr_language_manager.ensure_easyocr_models", return_value=[]) as mock_easy,
patch("app.config.settings", mock_settings),
):
ensure_ocr_languages_from_settings()
mock_tess.assert_called_once_with("eng")
mock_easy.assert_called_once_with(["en"])
@pytest.mark.unit
class TestEnsureOCRLanguagesAsync:
"""Tests for ensure_ocr_languages_async() background thread helper."""
def test_starts_daemon_thread(self):
"""ensure_ocr_languages_async starts a daemon thread."""
import threading
from app.utils.ocr_language_manager import ensure_ocr_languages_async
started_threads: list[threading.Thread] = []
original_start = threading.Thread.start
def track_start(self_thread):
started_threads.append(self_thread)
original_start(self_thread)
with (
patch.object(threading.Thread, "start", track_start),
patch("app.utils.ocr_language_manager.ensure_ocr_languages_from_settings", return_value={}),
):
ensure_ocr_languages_async()
assert any(t.name == "ocr-language-manager" for t in started_threads)
assert all(t.daemon for t in started_threads if t.name == "ocr-language-manager")