diff --git a/docs/usage.md b/docs/usage.md index db7f221a1..66e6205a9 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -867,6 +867,9 @@ processed in the background. Zone OCR overwrites the fields it fills. A zone that finds no value leaves its field unchanged. +Each template's field updates are saved together. If a database write fails, that template's updates +are rolled back; the remaining templates can still run. + Managing OCR templates requires the OCR template permissions. Running zone OCR on documents requires permission to change them, and choosing a template requires permission to view it. diff --git a/src/documents/tests/test_zone_ocr.py b/src/documents/tests/test_zone_ocr.py index 7e0f7db9e..9e9a5e135 100644 --- a/src/documents/tests/test_zone_ocr.py +++ b/src/documents/tests/test_zone_ocr.py @@ -1,5 +1,6 @@ """Tests for the zone-based OCR extraction engine.""" +import subprocess import tempfile from datetime import date from pathlib import Path @@ -24,7 +25,9 @@ from documents.zone_ocr import _convert_value from documents.zone_ocr import _crop_zone from documents.zone_ocr import _detect_mime from documents.zone_ocr import _extract_zone +from documents.zone_ocr import _prepare_field_update from documents.zone_ocr import _process_template +from documents.zone_ocr import _render_pages from documents.zone_ocr import _resolve_doc_path from documents.zone_ocr import extract_zone_preview from documents.zone_ocr import get_templates_for_document @@ -110,6 +113,39 @@ class TestZoneExtraction(SimpleTestCase): self.assertIsNone(result.raw_text) self.assertIsNone(result.value) + @patch("documents.zone_ocr.subprocess.run") + def test_reader_failures_return_no_value(self, run): + for error in [ + subprocess.TimeoutExpired("tesseract", 30), + subprocess.CalledProcessError(1, "tesseract", stderr="failure"), + subprocess.CalledProcessError(1, "tesseract"), + FileNotFoundError("tesseract"), + ]: + with self.subTest(error=type(error).__name__): + run.side_effect = error + result = _extract_zone(self.page, self.zone, 100, 50, self.tmp_path) + self.assertIsNone(result.raw_text) + self.assertIsNone(result.value) + + @patch("zxingcpp.read_barcodes", side_effect=RuntimeError("reader failed")) + def test_barcode_failure_returns_no_value(self, read_barcodes): + self.zone.transform = "qr_code" + result = _extract_zone(self.page, self.zone, 100, 50, self.tmp_path) + self.assertIsNone(result.raw_text) + self.assertIsNone(result.value) + + @patch("documents.zone_ocr.subprocess.run") + def test_preview_tiny_crop_returns_empty_result(self, run): + self.zone.width = 0 + self.assertEqual( + extract_zone_preview(self.page, self.zone, 100, 50, 1), + { + "raw_text": None, + "value": None, + }, + ) + run.assert_not_called() + @patch("documents.zone_ocr.subprocess.run") def test_date_transform_uses_zone_format(self, run): self.zone.transform = "date" @@ -193,7 +229,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): self.template.combine_formats = {"title": "{Second} / {First}"} run.side_effect = [MagicMock(stdout="one"), MagicMock(stdout="two")] with patch.object(self.document, "save", wraps=self.document.save) as save: - _process_template(self.document, self.page, self.template, zones) + _process_template( + self.document, + {0: self.page}, + self.template, + zones, + self.dirs.scratch_dir, + ) self.document.refresh_from_db() self.assertEqual(self.document.title, "two / one") save.assert_called_once_with(update_fields=["title"]) @@ -216,7 +258,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): "no", ] ] - _process_template(self.document, self.page, self.template, zones) + _process_template( + self.document, + {0: self.page}, + self.template, + zones, + self.dirs.scratch_dir, + ) self.document.refresh_from_db() self.assertEqual(self.document.title, "New title") self.assertEqual(self.document.archive_serial_number, 42) @@ -249,7 +297,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): "text", ] ] - _process_template(self.document, self.page, self.template, zones) + _process_template( + self.document, + {0: self.page}, + self.template, + zones, + self.dirs.scratch_dir, + ) self.document.refresh_from_db() self.assertEqual(self.document.title, "Valid") self.assertIsNone(self.document.archive_serial_number) @@ -272,7 +326,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): with patch.object(self.document, "save", side_effect=fail_after_save): with self.assertRaisesMessage(RuntimeError, "save failed"): - _process_template(self.document, self.page, self.template, zones) + _process_template( + self.document, + {0: self.page}, + self.template, + zones, + self.dirs.scratch_dir, + ) self.assertEqual(self.document.title, "Original") self.document.refresh_from_db() self.assertEqual(self.document.title, "Original") @@ -303,6 +363,102 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): self.document.refresh_from_db() self.assertEqual(self.document.title, "Second") + @patch("documents.zone_ocr.subprocess.run") + def test_document_run_renders_each_page_once_across_templates(self, run): + self.document.page_count = 2 + self.zone("First", page=1) + self.zone("Last", page=-1) + first = self.template + self.template = OcrTemplate.objects.create( + name="Second", + source_width=100, + source_height=50, + ) + self.zone("Last again", page=2) + pdf = self.dirs.scratch_dir / "document.pdf" + pdf.write_bytes(b"%PDF-1.4") + rendered = [] + + def execute(args, **kwargs): + if args[0] == "pdftoppm": + output = Path(args[-1]).with_suffix(".png") + with Image.new("RGB", (100, 50), "white") as image: + image.save(output) + rendered.append(output) + return subprocess.CompletedProcess(args, 0) + return subprocess.CompletedProcess(args, 0, stdout="extracted") + + run.side_effect = execute + results = run_zone_extraction( + self.document, + pdf, + templates=[first, self.template], + ) + render_calls = [ + call for call in run.call_args_list if call.args[0][0] == "pdftoppm" + ] + self.assertEqual(len(render_calls), 2) + self.assertEqual( + [call.args[0][call.args[0].index("-f") + 1] for call in render_calls], + ["1", "2"], + ) + self.assertEqual([result.reason for result in results], [None, None, None]) + self.assertFalse(any(path.exists() for path in rendered)) + + @patch("documents.zone_ocr.subprocess.run") + def test_zone_dimensions_override_template_dimensions(self, run): + zone = self.zone("Title", zone_source_width=200, zone_source_height=100) + + def read_crop(args, **kwargs): + with Image.open(args[1]) as crop: + self.assertEqual(crop.size, (20, 10)) + return subprocess.CompletedProcess(args, 0, stdout="extracted") + + run.side_effect = read_crop + _process_template( + self.document, + {0: self.page}, + self.template, + [zone], + self.dirs.scratch_dir, + ) + self.assertEqual(self.document.title, "extracted") + + def test_missing_page_returns_skipped_result(self): + zone = self.zone("Last", page=-1) + results = _process_template( + self.document, + {}, + self.template, + [zone], + self.dirs.scratch_dir, + ) + self.assertEqual(results[0].reason, "missing_page") + self.assertIsNone(results[0].value) + + def test_missing_document_file_returns_skipped_results(self): + self.zone("Title") + results = run_zone_extraction(self.document, None, templates=[self.template]) + self.assertEqual(len(results), 1) + self.assertEqual(results[0].reason, "missing_file") + self.assertIsNone(results[0].value) + + @patch("documents.zone_ocr.subprocess.run") + def test_empty_combination_cannot_clear_a_field(self, run): + zone = self.zone("Title") + self.template.combine_formats = {"title": "{Unknown}"} + run.return_value.stdout = "extracted" + results = _process_template( + self.document, + {0: self.page}, + self.template, + [zone], + self.dirs.scratch_dir, + ) + self.assertEqual(results[0].reason, "invalid_value") + self.assertIsNone(results[0].value) + self.assertEqual(self.document.title, "Original") + @patch("documents.zone_ocr.subprocess.run") def test_skipped_results_never_report_existing_document_values(self, run): zones = [ @@ -314,7 +470,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): run.side_effect = [ MagicMock(stdout=value) for value in ["", "rejected", "abc", "orphan"] ] - results = _process_template(self.document, self.page, self.template, zones) + results = _process_template( + self.document, + {0: self.page}, + self.template, + zones, + self.dirs.scratch_dir, + ) self.assertEqual( [result.reason for result in results], [ @@ -362,7 +524,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): zone = self.zone("Title") self.template.combine_formats = {"title": "Invoice {Title}"} run.return_value.stdout = "" - results = _process_template(self.document, self.page, self.template, [zone]) + results = _process_template( + self.document, + {0: self.page}, + self.template, + [zone], + self.dirs.scratch_dir, + ) self.assertEqual(results[0].reason, "no_value") self.assertIsNone(results[0].value) self.assertEqual(self.document.title, "Original") @@ -383,7 +551,13 @@ class TestSavedZoneExtraction(DirectoriesMixin, TestCase): preview = extract_zone_preview(self.page, zone, 100, 50, 1) self.assertEqual(preview["raw_text"], raw_text) self.assertEqual(preview["value"], raw_text.upper()) - _process_template(self.document, self.page, self.template, [zone]) + _process_template( + self.document, + {0: self.page}, + self.template, + [zone], + self.dirs.scratch_dir, + ) self.document.refresh_from_db() self.assertEqual(self.document.title, expected_title) @@ -564,6 +738,24 @@ class TestApplyTransform(TestCase): def test_unknown_transform_strips(self): self.assertEqual(_apply_transform(" hello ", "unknown"), "hello") + def test_strip_punctuation(self): + self.assertEqual( + _apply_transform(" -- Account #123. -- ", "strip_punctuation"), + "Account #123", + ) + + def test_invalid_iso_date_is_not_transformed(self): + self.assertEqual(_apply_transform("2026-13-32", "date"), "2026-13-32") + + @patch("documents.zone_ocr.dateparser.parse", side_effect=OverflowError) + def test_unparseable_date_preserves_text(self, parse): + self.assertEqual(_apply_transform("unparsable", "date"), "unparsable") + + def test_unknown_target_cannot_produce_a_field_update(self): + self.assertIsNone( + _prepare_field_update(OcrTemplateZone(target="unknown"), "value"), + ) + class TestConvertValue(TestCase): """Tests for the _convert_value function.""" @@ -697,6 +889,95 @@ class TestConvertValue(TestCase): def test_empty_string_returns_none(self): self.assertIsNone(_convert_value("", CustomField.FieldDataType.STRING)) + def test_numeric_types_reject_malformed_numbers(self): + for data_type in [ + CustomField.FieldDataType.FLOAT, + CustomField.FieldDataType.MONETARY, + ]: + for value in ["1.2.3", "--12", "12-34", "", ".", "-", "not numeric"]: + with self.subTest(data_type=data_type, value=value): + self.assertIsNone(_convert_value(value, data_type)) + + def test_float_overflow_is_not_stored_as_infinity(self): + self.assertIsNone(_convert_value("9" * 400, CustomField.FieldDataType.FLOAT)) + + def test_monetary_conversion_preserves_decimal_precision(self): + value = "12345678901234567890,12345678901234567890" + self.assertEqual( + _convert_value(value, CustomField.FieldDataType.MONETARY), + value.replace(",", "."), + ) + + +class TestRenderPages(SimpleTestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.tmp_path = Path(directory.name) + self.pdf = self.tmp_path / "document.pdf" + self.pdf.write_bytes(b"%PDF-1.4") + + @patch("documents.zone_ocr.subprocess.run") + def test_renders_requested_pages_in_order_and_skips_out_of_range(self, run): + def render(args, **kwargs): + Path(args[-1]).with_suffix(".png").write_bytes(b"rendered") + + run.side_effect = render + pages = _render_pages(self.pdf, {-1, 0, 2, 3}, self.tmp_path, 3) + self.assertEqual(list(pages), [0, 2]) + self.assertEqual(run.call_count, 2) + for page, call in zip([1, 3], run.call_args_list, strict=True): + args = call.args[0] + self.assertIn("-singlefile", args) + self.assertEqual(args[args.index("-f") + 1], str(page)) + self.assertEqual(args[args.index("-l") + 1], str(page)) + self.assertEqual(call.kwargs["timeout"], 60) + self.assertTrue(call.kwargs["check"]) + + @patch("documents.zone_ocr.subprocess.run") + def test_missing_render_output_is_not_a_page(self, run): + self.assertEqual(_render_pages(self.pdf, {0}, self.tmp_path, None), {}) + + @patch("documents.zone_ocr.subprocess.run") + def test_render_failure_does_not_prevent_other_pages(self, run): + for error in [ + subprocess.TimeoutExpired("pdftoppm", 60), + subprocess.CalledProcessError(1, "pdftoppm", stderr=b"failure"), + subprocess.CalledProcessError(1, "pdftoppm"), + ]: + with self.subTest(error=type(error).__name__): + + def render(args, **kwargs): + if args[args.index("-f") + 1] == "1": + raise error + Path(args[-1]).with_suffix(".png").write_bytes(b"rendered") + + run.side_effect = render + self.assertEqual( + list(_render_pages(self.pdf, {0, 1}, self.tmp_path, 2)), + [1], + ) + + @patch( + "documents.zone_ocr.subprocess.run", + side_effect=FileNotFoundError("pdftoppm"), + ) + def test_missing_renderer_stops_without_retrying_every_page(self, run): + self.assertEqual(_render_pages(self.pdf, {0, 1}, self.tmp_path, 2), {}) + run.assert_called_once() + + @patch("documents.zone_ocr.subprocess.run") + def test_image_is_used_directly_only_for_first_page(self, run): + image_path = self.tmp_path / "image.png" + with Image.new("RGB", (10, 10)) as image: + image.save(image_path) + self.assertEqual( + _render_pages(image_path, {0}, self.tmp_path, 1), + {0: image_path}, + ) + self.assertEqual(_render_pages(image_path, {1}, self.tmp_path, 1), {}) + run.assert_not_called() + class TestDetectMime(TestCase): """Tests for _detect_mime.""" @@ -727,6 +1008,21 @@ class TestResolveDocPath(TestCase): result = _resolve_doc_path(doc, None) self.assertIsNone(result) + def test_prefers_archive_and_falls_back_to_source(self): + with tempfile.TemporaryDirectory() as directory: + source = Path(directory) / "source.pdf" + archive = Path(directory) / "archive.pdf" + source.touch() + archive.touch() + document = MagicMock( + has_archive_version=True, + archive_path=archive, + source_path=source, + ) + self.assertEqual(_resolve_doc_path(document, None), archive) + archive.unlink() + self.assertEqual(_resolve_doc_path(document, None), source) + def test_returns_original_file_as_fallback(self): doc = MagicMock() doc.has_archive_version = False diff --git a/src/documents/zone_ocr.py b/src/documents/zone_ocr.py index 320518965..8ca1e49c0 100644 --- a/src/documents/zone_ocr.py +++ b/src/documents/zone_ocr.py @@ -1,15 +1,7 @@ -""" -Zone-based OCR extraction engine. - -After a document is consumed, this module checks if the document's type has -an active OCR template. If so, it renders the relevant pages as images, -crops each zone, runs Tesseract OCR on the crop, applies transforms, -and writes the results to the mapped custom fields. -""" - from __future__ import annotations import logging +import math import re import string import subprocess @@ -17,8 +9,13 @@ import tempfile from dataclasses import dataclass from datetime import date from datetime import datetime +from decimal import Decimal +from decimal import InvalidOperation from pathlib import Path +import dateparser +import magic +import zxingcpp from django.conf import settings from django.core.exceptions import ValidationError from django.db import transaction @@ -105,7 +102,6 @@ def run_zone_extraction( if not templates: return [] - # Resolve the document file: prefer archive (PDF/A), then source, then signal arg doc_path = _resolve_doc_path(document, original_file) if doc_path is None: logger.warning( @@ -118,30 +114,34 @@ def run_zone_extraction( for zone in template.zones.all() ] + template_zones = [(template, list(template.zones.all())) for template in templates] + pages = { + _resolve_page_idx(zone.page, document.page_count) + for _, zones in template_zones + for zone in zones + } + if not pages: + return [] results = [] - for template in templates: - zones = list(template.zones.all()) - if not zones: - continue - - logger.info( - "Zone OCR: processing template '%s' for document %d (%d zones)", - template.name, - document.pk, - len(zones), - ) - - try: - results.extend(_process_template(document, doc_path, template, zones)) - except Exception: - logger.exception( - "Zone OCR: error processing template '%s' for document %d", - template.name, - document.pk, - ) - results.extend( - _zone_result(template, zone, reason="template_failed") for zone in zones - ) + with tempfile.TemporaryDirectory(dir=settings.SCRATCH_DIR) as tmp_dir: + tmp_path = Path(tmp_dir) + page_images = _render_pages(doc_path, pages, tmp_path, document.page_count) + for template, zones in template_zones: + try: + results.extend( + _process_template(document, page_images, template, zones, tmp_path), + ) + except Exception: + # One failed template must not prevent the remaining templates from running. + logger.exception( + "Zone OCR: error processing template '%s' for document %d", + template.name, + document.pk, + ) + results.extend( + _zone_result(template, zone, reason="template_failed") + for zone in zones + ) return results @@ -149,7 +149,7 @@ def _resolve_doc_path( document: Document, original_file: Path | None, ) -> Path | None: - """Find an accessible file for the document.""" + """Prefer the archive, then the source, then the ingestion signal's file.""" candidates = [] if document.has_archive_version: candidates.append(document.archive_path) @@ -163,78 +163,51 @@ def _resolve_doc_path( return None -def _resolve_page_idx(page_value, page_count) -> int: - """Resolve a 1-indexed page (1 = first, -1 = last) to a 0-indexed image - index. A blank page_value defaults to the first page.""" - if page_value is None: - return 0 +def _resolve_page_idx(page_value: int | None, page_count: int | None) -> int: + """Resolve a 1-based page or -1 (last page), defaulting to the first page.""" if page_value == -1: - return (page_count - 1) if page_count else 0 - if page_value >= 1: - return page_value - 1 - return 0 + return max((page_count or 1) - 1, 0) + return max((page_value or 1) - 1, 0) def _process_template( document: Document, - doc_path: Path, + page_images: dict[int, Path], template: OcrTemplate, zones: list[OcrTemplateZone], + tmp_path: Path, ) -> list[ZoneOcrResult]: """Extract and combine zone values before applying a template's updates.""" - pages_needed: set[int] = { - _resolve_page_idx(zone.page, document.page_count) for zone in zones - } - - with tempfile.TemporaryDirectory(dir=settings.SCRATCH_DIR) as tmp_dir: - tmp_path = Path(tmp_dir) - - page_images = _render_pages( - doc_path, - pages_needed, - tmp_path, - document.page_count, - ) - - grouped: dict[str, list[tuple[OcrTemplateZone, ZoneOcrResult]]] = {} - results = [] - for zone in zones: - page_idx = _resolve_page_idx(zone.page, document.page_count) - extracted = None - reason = "missing_page" - if page_idx in page_images: - extraction = _extract_zone( - page_images[page_idx], - zone, - zone.zone_source_width or template.source_width, - zone.zone_source_height or template.source_height, - tmp_path, - ) - extracted = extraction.value - reason = None if extracted else "no_value" - - if ( - extracted is not None - and zone.validation_regex - and not safe_regex_fullmatch(zone.validation_regex, extracted) - ): - logger.info( - "Zone OCR: '%s' value %r rejected by regex '%s'", - zone.name, - extracted[:100], - zone.validation_regex, - ) - extracted = None - reason = "regex_mismatch" - - result = _zone_result( - template, + grouped: dict[str, list[tuple[OcrTemplateZone, ZoneOcrResult]]] = {} + results = [] + for zone in zones: + page_idx = _resolve_page_idx(zone.page, document.page_count) + extracted = None + reason = "missing_page" + if page_idx in page_images: + extraction = _extract_zone( + page_images[page_idx], zone, - value=extracted if reason is None else None, - reason=reason, + zone.zone_source_width or template.source_width, + zone.zone_source_height or template.source_height, + tmp_path, ) - results.append(result) - grouped.setdefault(_field_key(zone), []).append((zone, result)) + extracted = extraction.value + reason = None if extracted else "no_value" + if ( + extracted is not None + and zone.validation_regex + and not safe_regex_fullmatch(zone.validation_regex, extracted) + ): + reason = "regex_mismatch" + result = _zone_result( + template, + zone, + value=extracted if reason is None else None, + reason=reason, + ) + results.append(result) + grouped.setdefault(_field_key(zone), []).append((zone, result)) updates = [] combine_formats = template.combine_formats or {} @@ -261,11 +234,9 @@ def _process_template( def _field_key(zone: OcrTemplateZone) -> str: - """Identify a zone's target field. Custom fields key by id, built-in targets - by their name. Matches the key used in OcrTemplate.combine_formats and on the - frontend field select.""" + """Use the same target keys as the editor's combination formats.""" target = zone.target - if target == "custom_field" and zone.custom_field_id: + if target == OcrTemplateZone.TargetType.CUSTOM_FIELD and zone.custom_field_id: return str(zone.custom_field_id) return target @@ -306,14 +277,12 @@ def _render_pages( mime = _detect_mime(doc_path) if mime and mime.startswith("image/"): - # Single-image document — use it directly as page 0. - result[0] = doc_path + if 0 in pages: + result[0] = doc_path return result - # Callers pass already-resolved 0-indexed page numbers (see _resolve_page_idx). - for actual_page in pages: - if actual_page < 0: - logger.warning("Zone OCR: invalid page index %d", actual_page) + for actual_page in sorted(pages): + if actual_page < 0 or (page_count and actual_page >= page_count): continue output_prefix = tmp_dir / f"page_{actual_page}" @@ -322,6 +291,7 @@ def _render_pages( [ "pdftoppm", "-png", + "-singlefile", "-r", "300", "-f", @@ -346,13 +316,12 @@ def _render_pages( ) continue except FileNotFoundError: - logger.error("Zone OCR: pdftoppm not found — is poppler-utils installed?") - return result # No point trying other pages + logger.error("Zone OCR: pdftoppm not found") + return result - # pdftoppm names output as prefix-NNNN.png - rendered = sorted(tmp_dir.glob(f"page_{actual_page}-*.png")) - if rendered: - result[actual_page] = rendered[0] + rendered = output_prefix.with_suffix(".png") + if rendered.is_file(): + result[actual_page] = rendered return result @@ -381,15 +350,15 @@ def _crop_zone( # Clamp to the image so an oversized zone can't crop out of bounds. crop_left = max(0, min(crop_left, img_width)) crop_top = max(0, min(crop_top, img_height)) - crop_right = max(crop_left + 1, min(crop_right, img_width)) - crop_bottom = max(crop_top + 1, min(crop_bottom, img_height)) + crop_right = min(crop_right, img_width) + crop_bottom = min(crop_bottom, img_height) if crop_right - crop_left < 2 or crop_bottom - crop_top < 2: logger.warning("Zone OCR: crop too small for zone '%s'", zone.name) return None return img.crop((crop_left, crop_top, crop_right, crop_bottom)) - except Exception: + except (OSError, ValueError, Image.DecompressionBombError): logger.exception("Zone OCR: crop failed for zone '%s'", zone.name) return None @@ -397,23 +366,9 @@ def _crop_zone( def _read_barcode(cropped: Image.Image, zone_name: str) -> str | None: """Read QR/barcode from a cropped image using zxingcpp.""" try: - import zxingcpp - results = zxingcpp.read_barcodes(cropped) - if results: - text = results[0].text - logger.debug( - "Zone OCR: barcode found in zone '%s': %s", - zone_name, - text[:100], - ) - return text - logger.debug("Zone OCR: no barcode found in zone '%s'", zone_name) - return None - except ImportError: - logger.error("Zone OCR: zxingcpp not available — install zxing-cpp") - return None - except Exception: + return results[0].text if results else None + except (RuntimeError, ValueError, TypeError): logger.exception("Zone OCR: barcode read failed for zone '%s'", zone_name) return None @@ -451,7 +406,7 @@ def _ocr_text(cropped: Image.Image, zone: OcrTemplateZone, tmp_dir: Path) -> str ) return None except FileNotFoundError: - logger.error("Zone OCR: Tesseract not found — is tesseract-ocr installed?") + logger.error("Zone OCR: Tesseract not found") return None @@ -508,21 +463,25 @@ def extract_zone_preview( return {"raw_text": result.raw_text, "value": result.value} -def _parse_date(text: str, fmt: str) -> str: - """Parse a date from OCR text. With a Python strptime `fmt`, try that first; - otherwise (or on failure) fall back to dateparser auto-detection. Returns an - ISO date string, or the original text if nothing parses.""" - text = text.strip() - if not text: - return text +def _parse_iso_date(text: str) -> date | None: + match = re.search(r"\d{4}-\d{2}-\d{2}", text) + return date.fromisoformat(match[0]) if match else None + + +def _parse_date(text: str, fmt: str = "") -> date | None: + """Try the zone's date format, ISO dates, then automatic date detection.""" if fmt: try: - return datetime.strptime(text, fmt).date().isoformat() + return datetime.strptime(text, fmt).date() except ValueError: pass try: - import dateparser - + iso_date = _parse_iso_date(text) + except ValueError: + iso_date = None + if iso_date is not None: + return iso_date + try: parsed = dateparser.parse( text, settings={ @@ -531,10 +490,10 @@ def _parse_date(text: str, fmt: str) -> str: }, ) if parsed: - return parsed.date().isoformat() - except Exception: - logger.debug("Zone OCR: dateparser failed for %r", text[:50]) - return text + return parsed.date() + except (ValueError, OverflowError): + logger.debug("Zone OCR: date parsing failed") + return None def _apply_transform(text: str, transform: str, date_format: str = "") -> str: @@ -546,7 +505,8 @@ def _apply_transform(text: str, transform: str, date_format: str = "") -> str: if transform in ("strip", "none"): return text elif transform == "date": - return _parse_date(text, date_format) + parsed = _parse_date(text, date_format) + return parsed.isoformat() if parsed else text elif transform == "uppercase": return text.upper() elif transform == "lowercase": @@ -556,41 +516,9 @@ def _apply_transform(text: str, transform: str, date_format: str = "") -> str: return result if result else text elif transform == "strip_punctuation": return text.strip(string.punctuation + " \t\r\n") - elif transform == "qr_code": - # Barcode/QR content as read by _read_barcode. - return text return text -def _parse_created_datetime(value: str): - """Parse an extracted value into a tz-aware datetime for document.created. - - Prefers an ISO date (the zone should use a date transform); falls back to - dateparser. Returns None if no date can be parsed. - """ - from django.utils import timezone as djtz - - m = re.search(r"(\d{4})-(\d{2})-(\d{2})", value) - if m: - try: - dt = datetime(int(m[1]), int(m[2]), int(m[3])) - return djtz.make_aware(dt) if djtz.is_naive(dt) else dt - except ValueError: - pass - try: - import dateparser - - parsed = dateparser.parse( - value, - settings={"RETURN_AS_TIMEZONE_AWARE": False}, - ) - if parsed: - return djtz.make_aware(parsed) if djtz.is_naive(parsed) else parsed - except Exception: - logger.debug("Zone OCR: dateparser failed for created value %r", value[:50]) - return None - - def _prepare_field_update( zone: OcrTemplateZone, value: str, @@ -613,8 +541,7 @@ def _prepare_field_update( typed_value = int(digits) if digits else None model = Document elif zone.target == OcrTemplateZone.TargetType.CREATED: - parsed = _parse_created_datetime(value) - field_name, typed_value = "created", parsed.date() if parsed else None + field_name, typed_value = "created", _parse_date(value) model = Document else: return None @@ -655,6 +582,20 @@ def _apply_field_updates(document: Document, updates: list[FieldUpdate]) -> None raise +def _normalize_number(value: str) -> str: + cleaned = re.sub(r"[^\d.,\-]", "", value) + # When both separators occur, the rightmost one is the decimal separator. + if "," in cleaned and "." in cleaned: + if cleaned.rindex(",") > cleaned.rindex("."): + cleaned = cleaned.replace(".", "").replace(",", ".") + else: + cleaned = cleaned.replace(",", "") + else: + cleaned = cleaned.replace(",", ".") + Decimal(cleaned) + return cleaned + + def _convert_value(value: str, data_type: str) -> object | None: """Convert an extracted OCR string to the appropriate type for the custom field.""" if not value: @@ -672,56 +613,23 @@ def _convert_value(value: str, data_type: str) -> object | None: elif data_type == CustomField.FieldDataType.INT: digits = re.sub(r"[^\d\-]", "", value) - # Handle edge case: only dashes or empty - digits = digits.lstrip("-") or "" + digits = digits.lstrip("-") if not digits: return None - # Restore leading minus if original had one if value.strip().startswith("-"): digits = "-" + digits return int(digits) elif data_type == CustomField.FieldDataType.FLOAT: - # Handle European format: 1.234,56 → 1234.56 - cleaned = re.sub(r"[^\d.,\-]", "", value) - if not cleaned or cleaned in (".", ",", "-"): - return None - # If both . and , present, the last one is the decimal separator - if "," in cleaned and "." in cleaned: - if cleaned.rindex(",") > cleaned.rindex("."): - # European: 1.234,56 - cleaned = cleaned.replace(".", "").replace(",", ".") - else: - # US: 1,234.56 - cleaned = cleaned.replace(",", "") - elif "," in cleaned: - # Only comma — treat as decimal separator - cleaned = cleaned.replace(",", ".") - return float(cleaned) + number = float(_normalize_number(value)) + return number if math.isfinite(number) else None elif data_type == CustomField.FieldDataType.DATE: - match = re.search(r"(\d{4})-(\d{2})-(\d{2})", value) - if match: - y, m, d = match.groups() - # Validate the date - date(int(y), int(m), int(d)) - return f"{y}-{m}-{d}" - return None + parsed = _parse_iso_date(value) + return parsed.isoformat() if parsed else None elif data_type == CustomField.FieldDataType.MONETARY: - cleaned = re.sub(r"[^\d.,\-]", "", value) - if not cleaned or cleaned in (".", ",", "-"): - return None - if "," in cleaned and "." in cleaned: - if cleaned.rindex(",") > cleaned.rindex("."): - cleaned = cleaned.replace(".", "").replace(",", ".") - else: - cleaned = cleaned.replace(",", "") - elif "," in cleaned: - cleaned = cleaned.replace(",", ".") - # Validate it parses as a number - float(cleaned) - return cleaned + return _normalize_number(value) elif data_type == CustomField.FieldDataType.BOOL: lower = value.lower().strip() @@ -732,27 +640,22 @@ def _convert_value(value: str, data_type: str) -> object | None: return None else: - # Unsupported types (DOCUMENTLINK, SELECT) — can't OCR into these logger.debug( "Zone OCR: unsupported custom field type %s for OCR extraction", data_type, ) return None - except (ValueError, TypeError) as e: - logger.warning("Zone OCR: could not convert %r to %s: %s", value, data_type, e) + except (ValueError, InvalidOperation): + logger.info("Zone OCR: could not convert value to %s", data_type) return None def _detect_mime(path: Path) -> str | None: """Detect MIME type of a file.""" try: - import magic - return magic.from_file(str(path), mime=True) - except ImportError: - pass - except Exception: + except (OSError, magic.MagicException): logger.debug("Zone OCR: magic failed for %s, falling back to extension", path) suffix = path.suffix.lower()