""" OCR (Optical Character Recognition) Service Extract text from images using Tesseract and EasyOCR """ import logging from typing import List, Dict, Any, Optional, Tuple from io import BytesIO import numpy as np from PIL import Image import cv2 logger = logging.getLogger(__name__) class OCRService: """Text extraction from images""" def __init__(self): self.tesseract_available = False self.easyocr_available = False self.reader = None self._init_tesseract() self._init_easyocr() def _init_tesseract(self): """Initialize Tesseract OCR""" try: import pytesseract # Test if tesseract is available pytesseract.get_tesseract_version() self.pytesseract = pytesseract self.tesseract_available = True logger.info("Tesseract OCR initialized") except Exception as e: logger.warning(f"Tesseract not available: {e}") def _init_easyocr(self): """Initialize EasyOCR""" try: import easyocr self.reader = easyocr.Reader(['en', 'hi'], gpu=False) self.easyocr_available = True logger.info("EasyOCR initialized") except Exception as e: logger.warning(f"EasyOCR not available: {e}") def preprocess_image(self, image: np.ndarray) -> np.ndarray: """Preprocess image for better OCR results""" # Convert to grayscale if len(image.shape) == 3: gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) else: gray = image # Apply thresholding _, thresh = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) # Denoise denoised = cv2.fastNlMeansDenoising(thresh) return denoised async def extract_text( self, image_bytes: bytes, languages: List[str] = ["en"], preprocess: bool = True ) -> Dict[str, Any]: """ Extract text from an image Args: image_bytes: Image data as bytes languages: List of language codes preprocess: Whether to preprocess the image Returns: Extracted text and metadata """ # Load image image = Image.open(BytesIO(image_bytes)) image_np = np.array(image) # Preprocess if requested if preprocess: processed = self.preprocess_image(image_np) else: processed = image_np results = { "text": "", "confidence": 0.0, "blocks": [], "method": "none" } # Try EasyOCR first (better for multiple languages) if self.easyocr_available: try: results = await self._extract_with_easyocr(image_np, languages) except Exception as e: logger.error(f"EasyOCR failed: {e}") # Fallback to Tesseract if results["method"] == "none" and self.tesseract_available: try: results = await self._extract_with_tesseract(processed, languages) except Exception as e: logger.error(f"Tesseract failed: {e}") results["image_size"] = {"width": image.width, "height": image.height} results["languages_requested"] = languages return results async def _extract_with_easyocr( self, image: np.ndarray, languages: List[str] ) -> Dict[str, Any]: """Extract text using EasyOCR""" results = self.reader.readtext(image) blocks = [] all_text = [] total_confidence = 0 for bbox, text, confidence in results: blocks.append({ "text": text, "confidence": float(confidence), "bbox": { "top_left": [int(bbox[0][0]), int(bbox[0][1])], "bottom_right": [int(bbox[2][0]), int(bbox[2][1])] } }) all_text.append(text) total_confidence += confidence avg_confidence = total_confidence / len(results) if results else 0 return { "text": " ".join(all_text), "confidence": round(avg_confidence, 3), "blocks": blocks, "method": "easyocr", "block_count": len(blocks) } async def _extract_with_tesseract( self, image: np.ndarray, languages: List[str] ) -> Dict[str, Any]: """Extract text using Tesseract""" lang_str = "+".join(languages) # Get detailed data data = self.pytesseract.image_to_data( image, lang=lang_str, output_type=self.pytesseract.Output.DICT ) blocks = [] all_text = [] confidences = [] for i in range(len(data['text'])): text = data['text'][i].strip() conf = int(data['conf'][i]) if text and conf > 0: blocks.append({ "text": text, "confidence": conf / 100, "bbox": { "x": data['left'][i], "y": data['top'][i], "width": data['width'][i], "height": data['height'][i] }, "level": data['level'][i] }) all_text.append(text) confidences.append(conf) avg_confidence = sum(confidences) / len(confidences) if confidences else 0 return { "text": " ".join(all_text), "confidence": round(avg_confidence / 100, 3), "blocks": blocks, "method": "tesseract", "block_count": len(blocks) } async def extract_structured_data( self, image_bytes: bytes, document_type: str = "general" ) -> Dict[str, Any]: """ Extract structured data from documents Args: image_bytes: Image data document_type: Type of document (invoice, receipt, id_card, etc.) Returns: Structured extracted data """ # First extract all text ocr_result = await self.extract_text(image_bytes) # Parse based on document type if document_type == "invoice": return self._parse_invoice(ocr_result) elif document_type == "receipt": return self._parse_receipt(ocr_result) elif document_type == "id_card": return self._parse_id_card(ocr_result) else: return { "document_type": document_type, "raw_text": ocr_result["text"], "blocks": ocr_result["blocks"] } def _parse_invoice(self, ocr_result: Dict) -> Dict: """Parse invoice data""" import re text = ocr_result["text"].upper() # Extract common invoice fields using regex invoice_number = re.search(r'INVOICE\s*#?\s*:?\s*(\w+)', text) date = re.search(r'DATE\s*:?\s*(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})', text) total = re.search(r'TOTAL\s*:?\s*₹?\$?\s*([\d,]+\.?\d*)', text) gst = re.search(r'GST\s*:?\s*(\d{2}[A-Z]{5}\d{4}[A-Z]\d[A-Z\d]{2})', text) return { "document_type": "invoice", "extracted_fields": { "invoice_number": invoice_number.group(1) if invoice_number else None, "date": date.group(1) if date else None, "total_amount": total.group(1) if total else None, "gst_number": gst.group(1) if gst else None }, "raw_text": ocr_result["text"], "confidence": ocr_result["confidence"] } def _parse_receipt(self, ocr_result: Dict) -> Dict: """Parse receipt data""" import re text = ocr_result["text"] # Extract items and prices items = [] lines = text.split('\n') for line in lines: # Look for price patterns match = re.search(r'(.+?)\s+₹?\$?\s*([\d,]+\.?\d*)\s*$', line) if match: items.append({ "description": match.group(1).strip(), "amount": match.group(2) }) # Find total total = re.search(r'TOTAL\s*:?\s*₹?\$?\s*([\d,]+\.?\d*)', text.upper()) return { "document_type": "receipt", "items": items, "total": total.group(1) if total else None, "raw_text": text, "confidence": ocr_result["confidence"] } def _parse_id_card(self, ocr_result: Dict) -> Dict: """Parse ID card data""" import re text = ocr_result["text"] # Extract common ID fields name = re.search(r'NAME\s*:?\s*([A-Za-z\s]+)', text.upper()) dob = re.search(r'DOB\s*:?\s*(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})', text.upper()) id_number = re.search(r'\b(\d{4}\s?\d{4}\s?\d{4})\b', text) # Aadhaar pattern return { "document_type": "id_card", "extracted_fields": { "name": name.group(1).strip() if name else None, "date_of_birth": dob.group(1) if dob else None, "id_number": id_number.group(1) if id_number else None }, "raw_text": text, "confidence": ocr_result["confidence"] } # Singleton instance ocr_service = OCRService()