import re import os import time import json import copy import warnings import pytesseract from src.common.logger import get_cloudwatch_logger from src.utils.image import ( img_pdf_check, ensure_grayscale, resize_image, rotate_image, tilt_image ) import gc from src.utils.ocr_engines import find_best_rotation_pan, get_easyocr_reader from src.utils.pan_parser import get_structured_pan_data from src.utils.pan_rule_parser import extract_pan_details_rule_based warnings.filterwarnings('ignore') logger = get_cloudwatch_logger(stream_name="pan_ocr_stream") # Using shared reader from src.utils.ocr_engines def refine_pan_number(pan): """ Correct common OCR misreads based on PAN format: 5 Letters, 4 Digits, 1 Letter. """ if not pan or len(pan) != 10: return pan pan = pan.upper() corrected = "" # Mapping for OCR misreads to_letter = {'0': 'O', '1': 'I', '2': 'Z', '5': 'S', '8': 'B'} to_digit = {'O': '0', 'I': '1', 'Z': '2', 'S': '5', 'B': '8', '=': '5', 'G': '6'} for i in range(10): char = pan[i] if i < 5 or i == 9: # Should be Letter corrected += to_letter.get(char, char) else: # Should be Digit (Indices 5, 6, 7, 8) corrected += to_digit.get(char, char) if corrected != pan: logger.send_log_event(f"PAN Refined: {pan} -> {corrected}") return corrected async def process_pan(img_path, app_id, user_id, doc_type, ai_provider="gemini"): """ Robust flow for PAN card processing. """ start_time = time.time() logger.send_log_event(f"PAN OCR received for application id: {app_id}.") img = img_pdf_check(img_path) if isinstance(img, dict) and img.get("status") == "Failed": return {"application_id": app_id, "status": "failed", "message": img.get("message")} if img is None: return {"application_id": app_id, "status": "failed", "message": "Image loading returned None"} try: # --- 1. Preprocessing --- gray_img = ensure_grayscale(img) h, w = gray_img.shape[:2] if w < 1000 or h < 600: gray_img = resize_image(gray_img, max_dim=1200) # 1.3 Rotation (Efficient Short-circuiting) best_angle, rotated_img = find_best_rotation_pan(gray_img) logger.send_log_event(f"Application_id: {app_id}. Best rotation angle detected: {best_angle}") # Apply deskew processed_img = tilt_image(rotated_img) # --- 2. Dual Engine OCR --- # Tesseract tess_text = pytesseract.image_to_string(processed_img, config="--psm 6") tess_lines = [line.strip() for line in tess_text.split("\n") if len(line.strip()) > 2] # EasyOCR (Always run to provide full context for the AI/Robustness) reader = get_easyocr_reader() easy_results = reader.readtext(processed_img, detail=0) easy_lines = [r.strip() for r in easy_results if len(r.strip()) > 2] combined_text = f"--- TESSERACT ---\n{tess_text}\n\n--- EASYOCR ---\n" + "\n".join(easy_lines) # --- 3. Initial Extraction (Rule-Based for PAN/DOB) --- pan_strict_pattern = r"\b[A-Z]{5}[0-9]{4}[A-Z]\b" pan_lenient_pattern = r"\b[A-Z0-9]{5}[0-9A-Z]{4}[A-Z0-9]\b" def find_pan(text_lines): for line in text_lines: match = re.search(pan_strict_pattern, line.upper()) if match: return match.group().upper() match = re.search(pan_lenient_pattern, line.upper()) if match: refined = refine_pan_number(match.group()) if re.match(pan_strict_pattern, refined): return refined return None # Try rule-based first for DOB/PAN (very accurate) data = extract_pan_details_rule_based(tess_lines + easy_lines) data["pan_number"] = find_pan(tess_lines + easy_lines) # --- 4. AI Refinement (For Name and Father Name) --- logger.send_log_event(f"Application_id: {app_id}. Sending dual-engine text to AI for Name extraction.") # Redact PAN before sending redacted_text = combined_text if data["pan_number"]: redacted_text = redacted_text.replace(data["pan_number"], "[REDACTED]") redacted_text = re.sub(pan_strict_pattern, "[REDACTED]", redacted_text, flags=re.IGNORECASE) ai_data = get_structured_pan_data(redacted_text, app_id, provider="gemini", logger=logger) # Prioritize AI for names as requested for robustness data["name"] = ai_data.get("name") data["father_name"] = ai_data.get("father_name") if not data["dob"]: data["dob"] = ai_data.get("dob") status = "success" if data["pan_number"] else "failed" # Explicit cleanup to free memory in low-RAM environments del processed_img, rotated_img, gray_img, img gc.collect() elapsed_time = round(time.time() - start_time, 2) response = { "application_id": app_id, "user_id": user_id, "vendor": "CF", "raw_lines": combined_text, "result": [ { "details": { "panNo": {"value": data["pan_number"]}, "name": {"value": data["name"]}, "father": {"value": data["father_name"]}, "date": {"value": data["dob"]} }, "type": "pan_card" } ], "status": status, "statusCode": 200, "error_msg": "" if status == "success" else "PAN number not detected" } # Logging masked response for security log_res = copy.deepcopy(response) if status == "success" and data["pan_number"]: p = data["pan_number"] # Masking middle characters Example: ABCDE1234F -> ABCXXXXX4F masked_pan = p[:3] + "XXXXX" + p[-2:] log_res["result"][0]["details"]["panNo"]["value"] = masked_pan logger.send_log_event(f"Application_id: {app_id}. PAN OCR Completed. Time: {elapsed_time}s. Result: {json.dumps(log_res)}") response["response_time"] = elapsed_time return response except Exception as e: elapsed_time = round(time.time() - start_time, 2) logger.send_log_event(f"Error in process_pan for {app_id}: {str(e)}. Time: {elapsed_time}s") return { "application_id": app_id, "status": "failed", "message": str(e), "response_time": elapsed_time }