Merge pull request #391 from christianlouis/copilot/ensure-tesseract-easyocr-languages
feat(ocr): auto-install Tesseract/EasyOCR language data from settings
This commit is contained in:
Christian Krakau-Louis
@@ -21,11 +21,13 @@ COPY --from=builder /usr/local/bin /usr/local/bin
|
||||
# ghostscript – required by ocrmypdf for PDF/PS operations
|
||||
# poppler-utils – provides pdfinfo/pdftoppm used by pdf2image
|
||||
# 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 \
|
||||
tesseract-ocr \
|
||||
ghostscript \
|
||||
poppler-utils \
|
||||
unpaper \
|
||||
wget \
|
||||
&& apt-get clean && rm -rf /var/lib/apt/lists/*
|
||||
|
||||
# Copy application code
|
||||
|
||||
@@ -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")
|
||||
async def get_key_history(key: str, request: Request, db: DbSession, admin: AdminUser):
|
||||
"""
|
||||
|
||||
@@ -69,6 +69,11 @@ async def lifespan(app: FastAPI):
|
||||
finally:
|
||||
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
|
||||
from app.utils.config_validator import dump_all_settings
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -297,6 +297,19 @@ class TesseractOCRProvider(OCRProvider):
|
||||
|
||||
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})")
|
||||
pages = convert_from_path(file_path, dpi=300)
|
||||
texts: List[str] = []
|
||||
@@ -342,6 +355,9 @@ class EasyOCRProvider(OCRProvider):
|
||||
gpu = getattr(settings, "easyocr_gpu", False)
|
||||
|
||||
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)
|
||||
pages = convert_from_path(file_path, dpi=300)
|
||||
texts: List[str] = []
|
||||
|
||||
@@ -71,6 +71,17 @@ def notify_settings_updated() -> None:
|
||||
except Exception as 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:
|
||||
"""
|
||||
@@ -96,6 +107,13 @@ def register_settings_reload_signal() -> None:
|
||||
reload_settings_from_db(settings)
|
||||
_last_seen_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:
|
||||
logger.debug(f"Settings version check skipped: {exc}")
|
||||
|
||||
|
||||
@@ -485,7 +485,9 @@ For providers that do **not** embed a text layer, DocuElevate automatically runs
|
||||
|
||||
#### 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** |
|
||||
|------------------------|-----------------------------------------------------------------------------------|-------------|
|
||||
@@ -497,10 +499,17 @@ OCR_PROVIDERS=tesseract
|
||||
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)
|
||||
|
||||
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** |
|
||||
|-----------------------|------------------------------------------------------------------------|-------------|
|
||||
| `EASYOCR_LANGUAGES` | Comma-separated EasyOCR language codes, e.g. `en,de,fr`. | `en,de` |
|
||||
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user