"""Cloud OCR for image-only (scanned) transit declarations. The production host (o2switch, shared cPanel) gives no root access, so a system binary like Tesseract cannot be installed there. Scanned PDFs are therefore rasterised locally with PyMuPDF — a pure-Python wheel, no system dependency — and the page images are sent to a hosted OCR API over HTTPS. Two providers are supported, tried in the order listed by OCR_PROVIDER: OCR.space — free tier, simple API key, 1 MB/file on the free plan. Google Cloud Vision — better accuracy on poor scans, needs an API key with the Vision API enabled. Both are optional. With no key configured this module reports "unavailable" and the caller falls back to the "À confirmer" status rather than failing the upload — an unreadable scan must never lose the operator's document. All credentials come from the environment (see settings.OCR_*); nothing is hard-coded here. """ import base64 import logging import os from io import BytesIO from django.conf import settings logger = logging.getLogger(__name__) # Rasterisation settings. 300 DPI is the usual floor for reliable OCR on a # form; the base 72 DPI of a PDF point grid means a zoom of ~4.17. _OCR_DPI = 300 _PDF_BASE_DPI = 72 # Free OCR tiers cap the upload size (OCR.space: 1 MB). Pages are re-encoded at # a lower quality/scale until they fit rather than being rejected outright. _MAX_IMAGE_BYTES = 1024 * 1024 _MAX_PAGES = 3 # a transit declaration is 1-2 pages; guards against huge scans # Seconds to wait on the OCR service. The free OCR.space endpoint is a shared # resource and can be slow, but shared hosting kills a request that runs too # long — and the browser then shows a bare "Failed to fetch". Keeping this well # under the usual 60 s proxy limit means a slow OCR degrades to "À confirmer" # instead of killing the whole upload. Override with OCR_HTTP_TIMEOUT. _HTTP_TIMEOUT = int(os.environ.get('OCR_HTTP_TIMEOUT', '25')) class OcrUnavailable(Exception): """No OCR provider is configured, or every configured one failed. Raised instead of returning empty text so the caller can tell "the scan contains no officer name" apart from "we could not read the scan at all". """ def _page_images(path, dpi=_OCR_DPI): """Yields each page of the PDF as PNG bytes, downscaling any page that exceeds the free-tier size cap.""" import fitz # PyMuPDF — pure wheel, installs fine on shared hosting with fitz.open(path) as pdf: for index, page in enumerate(pdf): if index >= _MAX_PAGES: break current_dpi = dpi while True: zoom = current_dpi / _PDF_BASE_DPI pixmap = page.get_pixmap(matrix=fitz.Matrix(zoom, zoom), alpha=False) data = pixmap.tobytes('png') if len(data) <= _MAX_IMAGE_BYTES or current_dpi <= 120: yield data break # Too big for the free tier — step the resolution down and retry. current_dpi = int(current_dpi * 0.75) def _compress_png(data, max_bytes=_MAX_IMAGE_BYTES): """Last-resort shrink for a page still over the cap at the minimum DPI: re-encode as grayscale JPEG, which OCR engines handle just as well.""" if len(data) <= max_bytes: return data try: from PIL import Image image = Image.open(BytesIO(data)).convert('L') for quality in (75, 60, 45, 30): buffer = BytesIO() image.save(buffer, format='JPEG', quality=quality, optimize=True) if buffer.tell() <= max_bytes: return buffer.getvalue() return buffer.getvalue() except Exception: # noqa: BLE001 - compression is best-effort return data def _ocr_space(image_bytes): """OCR.space free API. Returns the page text, or '' when it reads nothing.""" import requests api_key = settings.OCR_SPACE_API_KEY if not api_key: raise OcrUnavailable('OCR_SPACE_API_KEY is not set') payload = { 'apikey': api_key, 'language': settings.OCR_SPACE_LANGUAGE, 'isOverlayRequired': 'false', # Engine 2 is markedly better on structured forms than the default. 'OCREngine': '2', 'scale': 'true', 'isTable': 'true', } response = requests.post( settings.OCR_SPACE_ENDPOINT, data=payload, files={'file': ('page.png', image_bytes)}, timeout=_HTTP_TIMEOUT, ) response.raise_for_status() result = response.json() if result.get('IsErroredOnProcessing'): message = result.get('ErrorMessage') or result.get('ErrorDetails') or 'unknown error' if isinstance(message, list): message = '; '.join(message) raise OcrUnavailable(f'OCR.space error: {message}') return '\n'.join( parsed.get('ParsedText') or '' for parsed in result.get('ParsedResults') or [] ).strip() def _google_vision(image_bytes): """Google Cloud Vision DOCUMENT_TEXT_DETECTION via the REST API + API key. The REST endpoint with an API key is used rather than the google-cloud client library so no service-account file has to live on the shared host. """ import requests api_key = settings.GOOGLE_VISION_API_KEY if not api_key: raise OcrUnavailable('GOOGLE_VISION_API_KEY is not set') body = { 'requests': [{ 'image': {'content': base64.b64encode(image_bytes).decode('ascii')}, # DOCUMENT_TEXT_DETECTION is tuned for dense forms, unlike the # sparser TEXT_DETECTION. 'features': [{'type': 'DOCUMENT_TEXT_DETECTION'}], 'imageContext': {'languageHints': settings.GOOGLE_VISION_LANGUAGE_HINTS}, }] } response = requests.post( f'{settings.GOOGLE_VISION_ENDPOINT}?key={api_key}', json=body, timeout=_HTTP_TIMEOUT, ) response.raise_for_status() payload = response.json() responses = payload.get('responses') or [{}] error = responses[0].get('error') if error: raise OcrUnavailable(f"Google Vision error: {error.get('message', 'unknown')}") annotation = responses[0].get('fullTextAnnotation') or {} return (annotation.get('text') or '').strip() _PROVIDERS = { 'ocrspace': _ocr_space, 'google': _google_vision, } def configured_providers(): """The provider callables to try, in the order given by OCR_PROVIDER, skipping any whose API key is missing.""" names = [n.strip().lower() for n in (settings.OCR_PROVIDER or '').split(',') if n.strip()] available = [] for name in names: provider = _PROVIDERS.get(name) if provider is None: logger.warning('Unknown OCR provider %r in OCR_PROVIDER, ignoring.', name) continue if name == 'ocrspace' and not settings.OCR_SPACE_API_KEY: continue if name == 'google' and not settings.GOOGLE_VISION_API_KEY: continue available.append((name, provider)) return available def is_available(): """True when at least one OCR provider has a usable API key.""" return bool(configured_providers()) def ocr_pdf(path): """Runs cloud OCR over a scanned PDF and returns the recognised text. Raises OcrUnavailable when no provider is configured or all of them fail (network error, exhausted quota, invalid key), so the caller can mark the document "À confirmer" instead of silently recording "no officer". """ providers = configured_providers() if not providers: raise OcrUnavailable( 'No cloud OCR provider configured — set OCR_SPACE_API_KEY or ' 'GOOGLE_VISION_API_KEY to classify scanned declarations.' ) try: pages = [_compress_png(image) for image in _page_images(path)] except Exception as exc: # noqa: BLE001 - missing/corrupt file, not an OCR fault raise OcrUnavailable(f'Could not rasterise {path}: {exc}') from exc if not pages: raise OcrUnavailable('The PDF has no rasterisable page.') failures = [] for name, provider in providers: try: text = '\n'.join(provider(image) for image in pages).strip() except OcrUnavailable as exc: # Quota exhausted or bad key — try the next provider. failures.append(f'{name}: {exc}') logger.warning('OCR provider %s unusable: %s', name, exc) continue except Exception as exc: # noqa: BLE001 - network/HTTP/JSON failures failures.append(f'{name}: {exc}') logger.warning('OCR provider %s failed: %s', name, exc) continue if text: logger.info('OCR succeeded via %s (%d chars).', name, len(text)) return text failures.append(f'{name}: returned no text') raise OcrUnavailable('; '.join(failures) or 'all OCR providers failed')