import io import pandas as pd from typing import Dict, List, Tuple, Any from datetime import datetime, date from utils import get_cloudwatch_logger, execute_query, TC_CODES, FOS_CODES import numpy as np import re import pytz from scripts.mysql_queries import ( email_list_of_collection_and_user_types, insert_collection_action_query, insert_file_upload_record_query, get_user_name_by_id_query ) STREAM_NAME = 'collection_ptp_data' logger = get_cloudwatch_logger(stream_name=STREAM_NAME) class PTPExcelProcessor: EXPECTED_COLUMNS = [ "application_id", "Email_id", "action_date", "dispo_code", "remarks", "next_action_date", "actual_amount_to_be_collected", "amount_to_be_received", "ptp_date", "interaction_type" ] def __init__(self, user_id: str, file_name: str): self.user_id = user_id self.df = None self.email_to_id_map = {} self.file_name = file_name def log_error(self, message: str): full_msg = f"[User: {self.user_id}] Error: {message}" print(full_msg) try: logger.send_log_event(full_msg) except Exception as log_err: print(f"Failed to send log event: {log_err}") def fetch_valid_emails(self): """Fetches valid emails and creates a mapping of {email: id} from DataFrame.""" try: df_results = execute_query(email_list_of_collection_and_user_types) if df_results is not None and not df_results.empty: self.email_to_id_map = dict(zip(df_results['email'], df_results['id'])) else: self.email_to_id_map = {} except Exception as e: self.log_error(f"Failed to fetch valid emails: {str(e)}") self.email_to_id_map = {} def load_file(self, file_content: bytes) -> Tuple[bool, str]: try: if not file_content: return False, "File content is empty." self.df = pd.read_excel(io.BytesIO(file_content)) return True, "File loaded successfully." except Exception as e: return False, f"Failed to parse Excel file: {str(e)}" def validate_columns(self) -> Tuple[bool, str]: if self.df is None: return False, "DataFrame not loaded." self.df.columns = [str(col).strip() for col in self.df.columns] uploaded_columns = set(self.df.columns) required_columns = set(self.EXPECTED_COLUMNS) missing = required_columns - uploaded_columns if missing: return False, f"Missing required columns: {', '.join(missing)}" return True, "Column structure valid." def get_uploader_name(self) -> str: """Fetches the name of the logged-in user uploading the file (DataFrame version).""" try: query = get_user_name_by_id_query.replace("%s", str(self.user_id)) df_result = execute_query(query) if df_result is not None and not df_result.empty: return df_result.iloc[0]["name"] return "Unknown" except Exception as e: self.log_error(f"Could not fetch uploader name: {e}") return "System" def validate_data_integrity(self) -> Tuple[bool, Any]: """ Vectorized validation with Format Checking for Optional Fields. """ errors = [] # 1. Standardize Data Types (Vectorized) # -------------------------------------- # CLEAN: Application ID (Mandatory) self.df["application_id_clean"] = pd.to_numeric( self.df["application_id"], errors="coerce" ) # CLEAN: Actual Amount (Mandatory) self.df["actual_amt_clean"] = pd.to_numeric( self.df["actual_amount_to_be_collected"], errors="coerce" ) # CLEAN: Received Amount (Optional) self.df["recv_amt_clean"] = pd.to_numeric( self.df["amount_to_be_received"], errors="coerce" ) # CLEAN: Dates date_cols = ["action_date", "next_action_date", "ptp_date"] for col in date_cols: # Note: format='mixed' helps pandas guess format, but 'coerce' silences errors self.df[f"{col}_clean"] = pd.to_datetime( self.df[col], errors="coerce", dayfirst=False ) # 2. Define Validation Masks (True = Error) # ----------------------------------------- # Helper: Check if original had value but clean is Null (Format Error) def is_format_error(orig_col, clean_col): # True if: Original is NOT (Null or Empty String) AND Clean IS Null return ( (self.df[orig_col].notna()) & (self.df[orig_col].astype(str).str.strip() != "") & (clean_col.isna()) ) # --- A. Format Validity Checks (Crucial for Optional Fields) --- mask_fmt_next_date = is_format_error( "next_action_date", self.df["next_action_date_clean"] ) mask_fmt_ptp_date = is_format_error("ptp_date", self.df["ptp_date_clean"]) mask_fmt_recv_amt = is_format_error( "amount_to_be_received", self.df["recv_amt_clean"] ) # Mandatory fields also benefit from specific format error messages mask_fmt_action_date = is_format_error( "action_date", self.df["action_date_clean"] ) mask_fmt_app_id = is_format_error( "application_id", self.df["application_id_clean"] ) # --- B. Mandatory Field Checks (Missing Values) --- # Note: We check the CLEAN column for missingness. # If it was junk, it's now NaN, so it triggers "Missing" unless we catch it with "Format Error" first. mask_app_id_missing = self.df["application_id_clean"].isna() & ( ~mask_fmt_app_id ) mask_action_date_missing = self.df["action_date_clean"].isna() & ( ~mask_fmt_action_date ) mask_amt_missing = self.df["actual_amt_clean"].isna() # Mandatory numeric mask_email_missing = self.df["Email_id"].isna() | ( self.df["Email_id"].astype(str).str.strip() == "" ) mask_dispo_missing = self.df["dispo_code"].isna() | ( self.df["dispo_code"].astype(str).str.strip() == "" ) # --- C. Email Authorization --- self.df["user_id_mapped"] = self.df["Email_id"].map(self.email_to_id_map) mask_invalid_email = (~mask_email_missing) & (self.df["user_id_mapped"].isna()) # --- D. Code Validity --- valid_codes = set(TC_CODES + FOS_CODES) mask_invalid_dispo = (~mask_dispo_missing) & ( ~self.df["dispo_code"].isin(valid_codes) ) mask_invalid_interaction = ~self.df["interaction_type"].isin(["TC", "FOS"]) mask_tc_mismatch = (self.df["interaction_type"] == "TC") & ( ~self.df["dispo_code"].isin(TC_CODES) ) mask_fos_mismatch = (self.df["interaction_type"] == "FOS") & ( ~self.df["dispo_code"].isin(FOS_CODES) ) # --- E. Date Logic Checks --- # Logic: Both dates exist, but order is wrong mask_next_date_logic = ( (self.df["next_action_date_clean"].notna()) & (self.df["action_date_clean"].notna()) & (self.df["next_action_date_clean"] < self.df["action_date_clean"]) ) mask_ptp_date_logic = ( (self.df["ptp_date_clean"].notna()) & (self.df["action_date_clean"].notna()) & (self.df["ptp_date_clean"] < self.df["action_date_clean"]) ) mask_ptp_mandatory = (self.df["dispo_code"] == "PTP") & ( self.df["ptp_date_clean"].isna() ) # --- F. Amount Logic Checks --- mask_amt_logic = ( (self.df["recv_amt_clean"].notna()) & (self.df["actual_amt_clean"].notna()) & (self.df["recv_amt_clean"] > self.df["actual_amt_clean"]) ) # 3. Aggregate Errors # ------------------- def add_errors(mask, message): if mask.any(): bad_rows = mask[mask].index + 2 for r in bad_rows: errors.append(f"Row {r}: {message}") # Add FORMAT errors first (Specific) add_errors(mask_fmt_app_id, "'application_id' must be a valid integer.") add_errors(mask_fmt_action_date, "'action_date' format must be YYYY-MM-DD.") add_errors(mask_fmt_next_date, "'next_action_date' format must be YYYY-MM-DD.") add_errors(mask_fmt_ptp_date, "'ptp_date' format must be YYYY-MM-DD.") add_errors(mask_fmt_recv_amt, "'amount_to_be_received' must be a valid number.") # Add MISSING errors (Generic) add_errors(mask_app_id_missing, "Missing 'application_id'.") add_errors(mask_email_missing, "Missing 'Email_id'.") add_errors(mask_action_date_missing, "Missing 'action_date'.") add_errors(mask_dispo_missing, "Missing 'dispo_code'.") add_errors( mask_amt_missing, "Invalid or Missing 'actual_amount_to_be_collected'." ) # Add LOGIC errors add_errors(mask_invalid_email, "Email unauthorized (not found in DB).") add_errors(mask_invalid_dispo, "Invalid 'dispo_code'.") add_errors( mask_invalid_interaction, "'interaction_type' must be 'TC' or 'FOS'." ) add_errors(mask_tc_mismatch, "Disposition code not valid for 'TC' interaction.") add_errors( mask_fos_mismatch, "Disposition code not valid for 'FOS' interaction." ) add_errors( mask_next_date_logic, "'next_action_date' cannot be before 'action_date'." ) add_errors(mask_ptp_date_logic, "'ptp_date' cannot be before 'action_date'.") add_errors(mask_ptp_mandatory, "'ptp_date' is mandatory for PTP disposition.") add_errors( mask_amt_logic, "'amount_to_be_received' cannot be greater than Actual Amount.", ) if errors: return False, errors return True, "Data integrity checks passed." def save_to_database(self) -> int: records_count = len(self.df) print(f"--- Inserting {records_count} records into DB ---") uploader_name = self.get_uploader_name() current_date = date.today() local_tz = pytz.timezone("Asia/Kolkata") current_timestamp = datetime.now(local_tz).strftime("%Y-%m-%d %H:%M:%S") # 1. Prepare Data using the Clean Columns we created in validation db_df = pd.DataFrame() db_df["application_id"] = self.df["application_id_clean"].astype(int) db_df["user_id"] = self.df["user_id_mapped"].astype(int) # Convert Timestamps to string 'YYYY-MM-DD' for SQL db_df["action_date"] = self.df["action_date_clean"].dt.strftime("%Y-%m-%d") db_df["dispo_code"] = self.df["dispo_code"].astype(str) db_df["remarks"] = self.df["remarks"].fillna("").astype(str).replace("", None) # Handle Nullable Dates db_df["next_action_date"] = ( self.df["next_action_date_clean"] .dt.strftime("%Y-%m-%d") .replace({np.nan: None}) ) db_df["actual_amount_to_be_collected"] = self.df["actual_amt_clean"] db_df["amount_to_be_received"] = self.df["recv_amt_clean"].replace( {np.nan: None} ) # Handle None for SQL db_df["ptp_date"] = ( self.df["ptp_date_clean"].dt.strftime("%Y-%m-%d").replace({np.nan: None}) ) # Vectorized Interaction Logic db_df["on_field_visit"] = np.where(self.df["interaction_type"] == "FOS", 1, 0) db_df["created_at"] = current_timestamp db_df["updated_at"] = current_timestamp # 2. Bulk Export to Dictionary (The Fast Way) bulk_data = db_df.to_dict(orient="records") # 3. Insert if bulk_data: execute_query(insert_collection_action_query, bulk_data) # 4. Log Upload file_upload_params = [ { "file_name": self.file_name, "uploaded_by": uploader_name, "upload_date": current_date, "total_rows": records_count, "created_at": current_timestamp, } ] execute_query(insert_file_upload_record_query, file_upload_params) return records_count def run(self, file_content: bytes) -> Dict[str, Any]: print(f"--- Processing Started for User: {self.user_id} ---") # 1. Fetch valid emails first self.fetch_valid_emails() if not self.email_to_id_map: return {"status": "error", "message": "System configuration error: No valid user emails found in database."} success, msg = self.load_file(file_content) if not success: self.log_error(msg) return {"status": "error", "message": msg} success, msg = self.validate_columns() if not success: self.log_error(msg) return {"status": "error", "message": msg} success, msg = self.validate_data_integrity() if not success: self.log_error(msg) return {"status": "error", "message": msg} try: count = self.save_to_database() return { "status": "success", "message": "File processed and data inserted successfully.", "processed_records": count } except Exception as e: err_msg = f"Database Insertion Failed: {str(e)}" self.log_error(err_msg) return {"status": "error", "message": err_msg} # --- Module Level Entry Point --- def process_ptp_excel(file_content: bytes, user_id: str, file_name: str,) -> Dict[str, Any]: """ Wrapper function to be called by the API endpoint. """ processor = PTPExcelProcessor(user_id, file_name) return processor.run(file_content)