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

fix(ocr): rewrite MistralOCRProvider to use native Mistral OCR API

Browse Source 
- PDFs are now uploaded to Mistral Files API (POST /v1/files) and
  processed via a signed document_url, resolving the 422 error caused
  by passing data:application/pdf;base64,... to an image endpoint
- Images (JPEG/PNG/GIF/WEBP/BMP/TIFF) use base64 image_url directly
- Unsupported MIME types raise a clear ValueError
- Magic-byte fallback detects PDFs with no file extension
- Switches from openai chat completions to requests HTTP calls
- Adds helper method _upload_pdf_and_get_document()
- Adds TestMistralOCRProvider with 9 unit tests covering all paths

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-24 12:13:38 +00:00
parent 2703221b9c
commit c7423834d5
2 changed files with 357 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/ocr_provider.py
+114 -27
View File
@@ -239,10 +239,15 @@ class EasyOCRProvider(OCRProvider):
class MistralOCRProvider(OCRProvider):
"""OCR via Mistral's document understanding API.
"""OCR via Mistral's dedicated OCR API (``/v1/ocr``).
Uses the ``mistral-ocr-latest`` model (or ``settings.mistral_ocr_model``)
via the OpenAI-compatible messages API.
For **PDF** files the document is first uploaded to the Mistral Files API
(``POST /v1/files``) to obtain a signed URL, then the OCR endpoint is
called with ``document_url``. For **image** files (JPEG, PNG, GIF, WEBP,
BMP, TIFF) the file is base64-encoded and passed directly as
``image_url``. Passing a PDF as a ``data:application/pdf`` data-URI to
the image path is explicitly rejected by the API and will produce a 422
error, so the two paths are kept strictly separate.
Config knobs (from :class:`~app.config.Settings`):
- ``mistral_api_key`` – Mistral API key.
@@ -251,13 +256,26 @@ class MistralOCRProvider(OCRProvider):
name = "mistral"
# MIME types that may be sent as base64 image_url payloads
_IMAGE_MIME_TYPES: frozenset = frozenset(
{
"image/jpeg",
"image/png",
"image/gif",
"image/webp",
"image/bmp",
"image/tiff",
}
)
def process(self, file_path: str) -> OCRResult:
import base64
import mimetypes
try:
import openai
import requests as req
except ImportError as exc:
raise RuntimeError("openai package is required for the Mistral OCR provider.") from exc
raise RuntimeError("requests package is required for the Mistral OCR provider.") from exc
api_key = getattr(settings, "mistral_api_key", None)
if not api_key:
@@ -265,36 +283,105 @@ class MistralOCRProvider(OCRProvider):
model = getattr(settings, "mistral_ocr_model", None) or "mistral-ocr-latest"
base_url = "https://api.mistral.ai/v1"
auth_headers: Dict[str, str] = {"Authorization": f"Bearer {api_key}"}
logger.info(f"[MistralOCR] Processing {os.path.basename(file_path)} with {model}")
with open(file_path, "rb") as f:
pdf_b64 = base64.b64encode(f.read()).decode("utf-8")
# Determine MIME type from extension, falling back to magic bytes.
mime_type, _ = mimetypes.guess_type(file_path)
if mime_type is None:
with open(file_path, "rb") as fh:
magic = fh.read(5)
if magic.startswith(b"%PDF-"):
mime_type = "application/pdf"
else:
raise ValueError(
f"[MistralOCR] Cannot determine file type for '{os.path.basename(file_path)}'. "
"Supported types: PDF, JPEG, PNG, GIF, WEBP, BMP, TIFF."
)
client = openai.OpenAI(api_key=api_key, base_url=base_url)
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": f"data:application/pdf;base64,{pdf_b64}"},
},
{
"type": "text",
"text": "Extract all text from this document. Return only the extracted text, preserving structure.",
},
],
}
],
document: Dict[str, Any]
if mime_type == "application/pdf":
document = self._upload_pdf_and_get_document(file_path, base_url, auth_headers, req)
elif mime_type in self._IMAGE_MIME_TYPES:
with open(file_path, "rb") as fh:
img_b64 = base64.b64encode(fh.read()).decode("utf-8")
document = {
"type": "image_url",
"image_url": f"data:{mime_type};base64,{img_b64}",
}
else:
raise ValueError(
f"[MistralOCR] Unsupported file type '{mime_type}' for "
f"'{os.path.basename(file_path)}'. "
"Supported types: PDF, JPEG, PNG, GIF, WEBP, BMP, TIFF."
)
ocr_payload: Dict[str, Any] = {"model": model, "document": document}
resp = req.post(
f"{base_url}/ocr",
headers={**auth_headers, "Content-Type": "application/json"},
json=ocr_payload,
timeout=300,
)
extracted_text = response.choices[0].message.content or ""
logger.info(f"[MistralOCR] Extracted {len(extracted_text)} chars")
resp.raise_for_status()
ocr_data = resp.json()
pages = ocr_data.get("pages", [])
extracted_text = "\n\n".join(page.get("markdown", "") for page in pages).strip()
logger.info(f"[MistralOCR] Extracted {len(extracted_text)} chars from {len(pages)} page(s)")
return OCRResult(provider="mistral", text=extracted_text)
def _upload_pdf_and_get_document(
self,
file_path: str,
base_url: str,
auth_headers: Dict[str, str],
req: Any,
) -> Dict[str, Any]:
"""Upload *file_path* to the Mistral Files API and return an OCR document dict.
Args:
file_path: Local path to the PDF file.
base_url: Mistral API base URL.
auth_headers: Dict containing the ``Authorization`` header.
req: The ``requests`` module (injected to allow mocking in tests).
Returns:
A document dict suitable for the ``/v1/ocr`` payload, e.g.
``{"type": "document_url", "document_url": "https://..."}``.
Raises:
requests.HTTPError: If the Files API upload or URL retrieval fails.
"""
logger.info(f"[MistralOCR] Uploading '{os.path.basename(file_path)}' to Mistral Files API")
with open(file_path, "rb") as fh:
upload_resp = req.post(
f"{base_url}/files",
headers=auth_headers,
files={"file": (os.path.basename(file_path), fh, "application/pdf")},
data={"purpose": "ocr"},
timeout=300,
)
upload_resp.raise_for_status()
file_id = upload_resp.json()["id"]
logger.info(f"[MistralOCR] Uploaded file id={file_id}; fetching signed URL")
url_resp = req.get(
f"{base_url}/files/{file_id}/url",
headers=auth_headers,
params={"expiry": 24},
timeout=30,
)
url_resp.raise_for_status()
signed_url: str = url_resp.json()["url"]
return {"type": "document_url", "document_url": signed_url}
class GoogleDocAIOCRProvider(OCRProvider):
"""OCR via Google Cloud Document AI.
tests/test_ocr_processing.py
+243
View File
@@ -777,3 +777,246 @@ startxref
# Verify rotation was applied despite string keys
assert result["status"] == "rotated"
assert "applied_rotations" in result
@pytest.mark.unit
class TestMistralOCRProvider:
"""Tests for MistralOCRProvider – Mistral native OCR API integration."""
def _make_provider(self):
from app.utils.ocr_provider import MistralOCRProvider
return MistralOCRProvider()
# ------------------------------------------------------------------
# Helpers: build mock response objects
# ------------------------------------------------------------------
def _mock_response(self, json_data: dict, status_code: int = 200):
mock_resp = Mock()
mock_resp.status_code = status_code
mock_resp.json.return_value = json_data
mock_resp.raise_for_status = Mock()
return mock_resp
def _mock_error_response(self, status_code: int = 422):
from requests import HTTPError
mock_resp = Mock()
mock_resp.status_code = status_code
mock_resp.raise_for_status.side_effect = HTTPError(response=mock_resp)
return mock_resp
# ------------------------------------------------------------------
# Missing API key
# ------------------------------------------------------------------
def test_missing_api_key_raises(self, tmp_path):
"""ValueError is raised when MISTRAL_API_KEY is not configured."""
provider = self._make_provider()
dummy_pdf = tmp_path / "doc.pdf"
dummy_pdf.write_bytes(b"%PDF-1.4\n%%EOF")
with patch("app.utils.ocr_provider.settings") as mock_settings:
mock_settings.mistral_api_key = None
mock_settings.mistral_ocr_model = None
with pytest.raises(ValueError, match="MISTRAL_API_KEY must be set"):
provider.process(str(dummy_pdf))
# ------------------------------------------------------------------
# PDF workflow
# ------------------------------------------------------------------
def test_pdf_uses_document_url_workflow(self, tmp_path):
"""PDF files are uploaded then processed via document_url."""
provider = self._make_provider()
pdf_file = tmp_path / "sample.pdf"
pdf_file.write_bytes(b"%PDF-1.4\n%%EOF")
upload_resp = self._mock_response({"id": "file-abc123"})
signed_url_resp = self._mock_response({"url": "https://signed.example.com/doc"})
ocr_resp = self._mock_response({"pages": [{"markdown": "Hello PDF World"}]})
with (
patch("app.utils.ocr_provider.settings") as mock_settings,
patch("requests.post", side_effect=[upload_resp, ocr_resp]) as mock_post,
patch("requests.get", return_value=signed_url_resp) as mock_get,
):
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
result = provider.process(str(pdf_file))
assert result.provider == "mistral"
assert result.text == "Hello PDF World"
# First POST should be the file upload
upload_call = mock_post.call_args_list[0]
assert "/files" in upload_call[0][0]
assert upload_call[1]["data"] == {"purpose": "ocr"}
# GET should fetch the signed URL
get_call = mock_get.call_args_list[0]
assert "file-abc123" in get_call[0][0]
# Second POST should be the OCR call
ocr_call = mock_post.call_args_list[1]
assert "/ocr" in ocr_call[0][0]
ocr_json = ocr_call[1]["json"]
assert ocr_json["document"]["type"] == "document_url"
assert ocr_json["document"]["document_url"] == "https://signed.example.com/doc"
def test_pdf_multi_page_text_joined(self, tmp_path):
"""Text from multiple pages is joined with double newlines."""
provider = self._make_provider()
pdf_file = tmp_path / "multi.pdf"
pdf_file.write_bytes(b"%PDF-1.4\n%%EOF")
upload_resp = self._mock_response({"id": "file-xyz"})
signed_url_resp = self._mock_response({"url": "https://signed.example.com/multi"})
ocr_resp = self._mock_response({"pages": [{"markdown": "Page one text"}, {"markdown": "Page two text"}]})
with (
patch("app.utils.ocr_provider.settings") as mock_settings,
patch("requests.post", side_effect=[upload_resp, ocr_resp]),
patch("requests.get", return_value=signed_url_resp),
):
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
result = provider.process(str(pdf_file))
assert "Page one text" in result.text
assert "Page two text" in result.text
# ------------------------------------------------------------------
# Image workflow
# ------------------------------------------------------------------
def test_image_uses_image_url_workflow(self, tmp_path):
"""Image files are base64-encoded and passed as image_url (not document_url)."""
provider = self._make_provider()
img_file = tmp_path / "photo.jpg"
img_file.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 20) # minimal JPEG bytes
ocr_resp = self._mock_response({"pages": [{"markdown": "Image text here"}]})
with (
patch("app.utils.ocr_provider.settings") as mock_settings,
patch("requests.post", return_value=ocr_resp) as mock_post,
patch("requests.get") as mock_get,
):
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
result = provider.process(str(img_file))
assert result.provider == "mistral"
assert result.text == "Image text here"
# Should only make a single POST (no file upload step)
assert mock_post.call_count == 1
assert mock_get.call_count == 0
ocr_call = mock_post.call_args_list[0]
assert "/ocr" in ocr_call[0][0]
ocr_json = ocr_call[1]["json"]
assert ocr_json["document"]["type"] == "image_url"
assert ocr_json["document"]["image_url"].startswith("data:image/jpeg;base64,")
# ------------------------------------------------------------------
# Unsupported file type
# ------------------------------------------------------------------
def test_unsupported_mime_type_raises(self, tmp_path):
"""ValueError is raised for unsupported file types."""
provider = self._make_provider()
txt_file = tmp_path / "document.txt"
txt_file.write_text("hello world")
with patch("app.utils.ocr_provider.settings") as mock_settings:
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
with pytest.raises(ValueError, match="Unsupported file type"):
provider.process(str(txt_file))
def test_unknown_extension_pdf_magic_bytes_detected(self, tmp_path):
"""Files without extension are identified as PDF via magic bytes."""
provider = self._make_provider()
no_ext_file = tmp_path / "nodotfile"
no_ext_file.write_bytes(b"%PDF-1.4\n%%EOF")
upload_resp = self._mock_response({"id": "file-magic"})
signed_url_resp = self._mock_response({"url": "https://signed.example.com/magic"})
ocr_resp = self._mock_response({"pages": [{"markdown": "Magic PDF"}]})
with (
patch("app.utils.ocr_provider.settings") as mock_settings,
patch("requests.post", side_effect=[upload_resp, ocr_resp]),
patch("requests.get", return_value=signed_url_resp),
):
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
result = provider.process(str(no_ext_file))
assert result.text == "Magic PDF"
def test_unknown_extension_non_pdf_magic_bytes_raises(self, tmp_path):
"""Files without extension that aren't PDFs raise ValueError."""
provider = self._make_provider()
unknown_file = tmp_path / "unknownfile"
unknown_file.write_bytes(b"\x00\x01\x02\x03\x04")
with patch("app.utils.ocr_provider.settings") as mock_settings:
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
with pytest.raises(ValueError, match="Cannot determine file type"):
provider.process(str(unknown_file))
# ------------------------------------------------------------------
# Upload / API error propagation
# ------------------------------------------------------------------
def test_upload_failure_raises_http_error(self, tmp_path):
"""HTTPError from the Files API upload is propagated to the caller."""
from requests import HTTPError
provider = self._make_provider()
pdf_file = tmp_path / "bad.pdf"
pdf_file.write_bytes(b"%PDF-1.4\n%%EOF")
with (
patch("app.utils.ocr_provider.settings") as mock_settings,
patch("requests.post", return_value=self._mock_error_response(status_code=401)),
):
mock_settings.mistral_api_key = "bad-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
with pytest.raises(HTTPError):
provider.process(str(pdf_file))
def test_ocr_api_failure_raises_http_error(self, tmp_path):
"""HTTPError from the /ocr endpoint is propagated to the caller."""
from requests import HTTPError
provider = self._make_provider()
pdf_file = tmp_path / "bad.pdf"
pdf_file.write_bytes(b"%PDF-1.4\n%%EOF")
upload_resp = self._mock_response({"id": "file-err"})
signed_url_resp = self._mock_response({"url": "https://signed.example.com/err"})
with (
patch("app.utils.ocr_provider.settings") as mock_settings,
patch("requests.post", side_effect=[upload_resp, self._mock_error_response(status_code=422)]),
patch("requests.get", return_value=signed_url_resp),
):
mock_settings.mistral_api_key = "test-key"
mock_settings.mistral_ocr_model = "mistral-ocr-latest"
with pytest.raises(HTTPError):
provider.process(str(pdf_file))