import cv2, os, pytesseract, re, warnings, base64, gc import pandas as pd import numpy as np from pytesseract import Output from src.common.logger import get_cloudwatch_logger from src.utils.image import ( img_pdf_check, ensure_grayscale, resize_image, rotate_image, tilt_image, resize_v2 ) from src.utils.ocr_engines import ( find_best_rotation, extract_aadhaar_num, extract_aadhaar_easy_ocr ) from src.utils.address_parser import ( preprocess_ocr_address, get_structured_address ) warnings.filterwarnings('ignore') logger = get_cloudwatch_logger(stream_name="aadhaar_ocr_stream") def address_extract_v2(gray_img, app_id): """Secondary address extraction for scanned copies.""" logger.send_log_event(f"Application_id: {app_id}. In address_extract_v2.") resized_img = resize_v2(gray_img) best_angle, _ = find_best_rotation(resized_img) resized_img = rotate_image(resized_img, best_angle) resized_tilt_img = tilt_image(resized_img) ocr_data = pytesseract.image_to_data(resized_tilt_img, output_type=Output.DICT) ocr_df = pd.DataFrame(ocr_data) filtered_df = ocr_df[ (ocr_df['word_num'].astype(int) > 0) & (ocr_df['conf'].astype(int) > 5) ].reset_index(drop=True) filtered_df['is_pincode'] = filtered_df['text'].str.match(r'^\d{6}$', na=False) pincode_df = filtered_df[filtered_df['is_pincode']] if pincode_df.empty: return None pincode_row = pincode_df.iloc[0] pin_top = pincode_row['top'] address_area = filtered_df[ (filtered_df['top'] <= pin_top) & (pin_top - filtered_df['top'] < 150) ] full_address = ' '.join(address_area['text'].astype(str)) full_address = re.sub(r"(S/O|D/O|C/O|W/o|$/0)\s+[^,]+,", "", full_address) return full_address async def process_image(img_path, app_id, user_id, doc_type): """ Core flow for Aadhaar processing with masking and color/PDF support. """ logger.send_log_event(f"Aadhaar 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"} # Determine original extension ext = os.path.splitext(img_path)[1].lower() # Create color copy for output color_img = img.copy() color_img1 = img.copy() # Process color image matching the original grayscale flow color_img = resize_image(color_img, max_dim=1200) # Find rotation based on the resized grayscale version (original logic) gray_for_rot = ensure_grayscale(color_img) best_angle, _ = find_best_rotation(gray_for_rot) color_img = rotate_image(color_img, best_angle) color_img = tilt_image(color_img) # Replicate the user's expected grayscale flow resized_img = ensure_grayscale(color_img) gray_img1 = ensure_grayscale(color_img1) # Basic sharpening (original logic) kernel = np.array([[0, -1, 0], [-1, 5, -1], [0, -1, 0]]) sharpened = cv2.filter2D(resized_img, -1, kernel) sharpened = resize_image(sharpened, max_dim=1100) # Resize color image to match sharpened coordinates color_img = resize_image(color_img, max_dim=1100) # OCR Data data = pytesseract.image_to_data(sharpened, output_type=Output.DICT) df = pd.DataFrame(data) aadhaar_num, mask_bboxes = extract_aadhaar_num(df) if aadhaar_num is None: aadhaar_num, mask_bboxes = extract_aadhaar_easy_ocr(sharpened, app_id, logger=logger) # Masking logic masked_data_b64 = "" mask_img = color_img.copy() if aadhaar_num and mask_bboxes: try: for bbox in mask_bboxes: if isinstance(bbox, dict): # Tesseract style x, y, w, h = bbox['left'], bbox['top'], bbox['width'], bbox['height'] cv2.rectangle(mask_img, (x, y), (x + w, y + h), (128, 128, 128), -1) elif isinstance(bbox, list): # EasyOCR style pts = np.array(bbox, np.int32) x_min = np.min(pts[:, 0]) y_min = np.min(pts[:, 1]) x_max = np.max(pts[:, 0]) y_max = np.max(pts[:, 1]) w = x_max - x_min cv2.rectangle(mask_img, (x_min, y_min), (int(x_min + (w * 0.66)), y_max), (128, 128, 128), -1) # # Save locally for testing (as JPG) # local_mask_path = f"masked_{app_id}.jpg" # cv2.imwrite(local_mask_path, mask_img) # Prepare Base64 output if ext == ".pdf": # Convert masked color image to PDF properly import fitz img_bytes = cv2.imencode('.jpg', mask_img)[1].tobytes() pdf_doc = fitz.open() # Create a new page with the image dimensions page = pdf_doc.new_page(width=mask_img.shape[1], height=mask_img.shape[0]) page.insert_image(page.rect, stream=img_bytes) pdf_bytes = pdf_doc.tobytes() masked_data_b64 = base64.b64encode(pdf_bytes).decode('utf-8') pdf_doc.close() else: _, buffer = cv2.imencode('.jpg', mask_img) masked_data_b64 = base64.b64encode(buffer).decode('utf-8') except Exception as e: if logger: logger.send_log_event(f"Masking error: {str(e)}") address_key_word = df[df['text'].str.contains('Address', case=False, na=False)] # Extract address if 'Address' keyword is found or via secondary extraction address_string = "" parsed_address = {} if not address_key_word.empty: addr = address_key_word.iloc[0] address_candidates = df[ (df['left'] >= addr['left'] - 10) & (df['top'] >= addr['top'] - 10) & (df['word_num'] > 0) ] full_address = ' '.join(address_candidates['text'].astype(str)) full_address = re.sub(r"\b\d{4}\s?\d{4}\s?\d{4}\b", "", full_address) full_address = re.sub(r"(S/O|D/O|C/O|W/o|$/0)\s+[^,]+,", "", full_address) full_address = preprocess_ocr_address(full_address) parsed_address = get_structured_address(full_address, app_id, provider="gemini", logger=logger) address_string = " ".join(str(v) for v in parsed_address.values() if v is not None) else: full_address = address_extract_v2(gray_img1, app_id) if full_address and len(full_address.strip()) >= 4: full_address = preprocess_ocr_address(full_address) parsed_address = get_structured_address(full_address, app_id, provider="gemini", logger=logger) address_string = " ".join(str(v) for v in parsed_address.values() if v is not None) status = "success" if not aadhaar_num or not str(aadhaar_num).strip(): status = "failed" # Explicit cleanup for low-RAM Docker environments try: del sharpened, resized_img, color_img, color_img1, gray_img1, img except NameError: pass gc.collect() return { "application_id": app_id, "user_id": user_id, "vendor": "CF", "api_url": "", "Request": "", "requestId": "", "masked_b64": masked_data_b64, "result": [ { "details": { "aadhaar": { "value": aadhaar_num, "isMasked": "yes" if masked_data_b64 else "no" }, "aadhaar_number": { "value": aadhaar_num, "isMasked": "yes" if masked_data_b64 else "no" }, "address": { "value": address_string }, "pin": { "value": parsed_address.get("pincode", "") }, "imageUrl": { "value": "" }, "qr": { "value": "" }, "father": { "value": "" }, "husband": { "value": "" }, "addressSplit": { "careOf": "", "houseNumber": "", "city": parsed_address.get("city", ""), "line1": parsed_address.get("address_line_1", ""), "line2": "", "street": "", "locality": "", "landmark": "", "district": "", "state": parsed_address.get("state", ""), "pin": parsed_address.get("pincode", "") } }, "type": doc_type } ], "status": status, "statusCode": 200, "error_msg": "" if status == "success" else "Aadhaar number not detected" } # if __name__=='__main__': # import asyncio # file_path = "sample_docs//full_scan.jpg" # print(asyncio.run(process_image(file_path, "", "", "aadhaar_back")))