import pandas as pd from datetime import datetime from utils import execute_query, get_cloudwatch_logger, local_tz, log_exception, engine from scripts.mysql_queries import ( blacksoil_insert_payment_query, blacksoil_check_attributes_query, blacksoil_insert_attributes_query, blacksoil_update_attributes_query ) logger = get_cloudwatch_logger(stream_name="blacksoil_payments_upload") def validate_func(file_path): if not file_path.endswith(('.xlsx', '.csv')): return {"status": "FAILED", "message": "Please upload file in CSV or Excel file format."} try: df = pd.read_excel(file_path) if file_path.endswith('.xlsx') else pd.read_csv(file_path) except Exception as e: log_exception(logger, f"Error reading file: {str(e)}") return {"status": "FAILED", "message": f"Error reading file: {str(e)}"} required_columns = [ 'Cust ID', 'Loan ID', 'Amount received', 'Principal', 'Interest', 'Foreclosed Principal', 'Foreclosed Interest', 'Overdue Principal', 'Overdue Interest', 'Penal Charges', 'Prepayment Principal', 'Prepayment Interest', 'Foreclosed Charges', 'Payment Date', 'DPD' ] for col in required_columns: if col not in df.columns: logger.send_log_event(f"Incorrect file format uploaded. Missing column: {col}") return {"status": "FAILED", "message": "Incorrect file format uploaded"} if df.isnull().values.any(): logger.send_log_event("Incorrect file format uploaded .No blank cells are allowed to be uploaded.") return {"status": "FAILED", "message": "Incorrect file format uploaded .No blank cells are allowed to be uploaded."} numeric_cols = df.select_dtypes(include=['number']).columns if (df[numeric_cols] < 0).values.any(): logger.send_log_event("Incorrect file format uploaded.Negative value are not allowed") return {"status": "FAILED", "message": "Incorrect file format uploaded.Negative value are not allowed"} try: pd.to_datetime(df['Payment Date'], format='%d-%m-%Y') except ValueError: logger.send_log_event("Incorrect file format uploaded. Date should be in dd-mm-yyyy format.") return {"status": "FAILED", "message": "Incorrect file format uploaded. Date should be in dd-mm-yyyy format."} return "success" def update_tables(file_path, data): try: # Get current time in local timezone now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') df = pd.read_excel(file_path) if file_path.endswith('.xlsx') else pd.read_csv(file_path) if df.empty: logger.send_log_event("Uploaded file is empty.") return {"status": "FAILED", "message": "Uploaded file is empty."} # Convert Payment Date to string format, keep it for insertion, parse as datetime for grouping df['parsed_payment_date'] = pd.to_datetime(df['Payment Date'], dayfirst=True) df['Formatted Date'] = df['parsed_payment_date'].dt.strftime('%Y-%m-%d %H:%M:%S') # DPD extraction logic: group by Loan ID, get max date, extract DPD idx = df.groupby('Loan ID')['parsed_payment_date'].idxmax() latest_dpd_df = df.loc[idx, ['Loan ID', 'DPD']] latest_dpd_dict = dict(zip(latest_dpd_df['Loan ID'], latest_dpd_df['DPD'])) payment_mapping = { 'Principal': 'principal', 'Interest': 'interest', 'Foreclosed Principal': 'foreclosure_principal', 'Foreclosed Interest': 'foreclosure_interest', 'Foreclosed Charges': 'foreclosure_charges', 'Overdue Principal': 'overdue_principal', 'Overdue Interest': 'overdue_interest', 'Penal Charges': 'penal_charges', 'Prepayment Principal': 'prepayment_principal', 'Prepayment Interest': 'prepayment_interest' } params_list = [] for row in df.to_dict(orient='records'): for excel_col, payment_type in payment_mapping.items(): amount = float(row.get(excel_col, 0) or 0) if amount > 0: params_list.append({ "emi_payment_id": 0, "amount": amount, "cust_id": int(row['Cust ID']), "loan_id": int(row['Loan ID']), "payment_for": payment_type, "payment_date": row['Formatted Date'], "status": "Success", "created_at": now, "updated_at": now }) # Check existing loan_ids in attributes table unique_loan_ids = list(latest_dpd_dict.keys()) existing_loans_df = execute_query(blacksoil_check_attributes_query, params={"loan_ids": tuple(unique_loan_ids)}) if isinstance(existing_loans_df, str): logger.send_log_event(f"ERROR checking attributes: {existing_loans_df}") raise Exception(existing_loans_df) existing_loan_ids = set() if not existing_loans_df.empty: existing_loan_ids = set(existing_loans_df['loan_id'].tolist()) insert_params = [] update_params = [] for loan_id, dpd in latest_dpd_dict.items(): if loan_id in existing_loan_ids: update_params.append({ "loan_id": int(loan_id), "dpd": int(dpd), "updated_at": now }) else: insert_params.append({ "loan_id": int(loan_id), "dpd": int(dpd), "created_at": now, "updated_at": now }) if not params_list and not insert_params and not update_params: return {"status": "SUCCESS", "message": "No relevant entries found in the file."} # Execute batched queries batch_size = 1000 if params_list: payment_df = pd.DataFrame(params_list) if not payment_df.empty: with engine.begin() as conn: payment_df.to_sql( name='blacksoil_payment_transactions', con=conn, if_exists='append', index=False, method='multi', chunksize=1000 ) logger.send_log_event(f"INFO: Successfully inserted {len(params_list)} payment records into blacksoil_payment_transactions.") if insert_params: insert_attr_df = pd.DataFrame(insert_params) if not insert_attr_df.empty: with engine.begin() as conn: insert_attr_df.to_sql( name='blacksoil_loan_attributes', con=conn, if_exists='append', index=False, method='multi', chunksize=1000 ) logger.send_log_event(f"INFO: Inserted attributes for {len(insert_params)} new loans.") if update_params: for i in range(0, len(update_params), batch_size): execute_query(blacksoil_update_attributes_query, params=update_params[i:i+batch_size]) logger.send_log_event(f"INFO: Updated attributes for {len(update_params)} existing loans.") min_payment_date = df['parsed_payment_date'].min().strftime('%Y-%m-%d') if not df.empty else None return { "status": "SUCCESS", "message": f"Successfully uploaded {len(params_list)} payment transactions.", "loan_ids": unique_loan_ids, "min_payment_date": min_payment_date } except Exception as e: log_exception(logger, f"Exception in update_tables: {str(e)}") return {"status": "FAILED", "message": f"Error uploading payments: {str(e)}"} async def function1(data): file_path = data.get("uploaded_file_path") is_verify_flag = data.get("is_verify") logger.send_log_event(f"INFO: BlackSoil Payments upload request received: {data}") if is_verify_flag is None: return {"status": "FAILED", "message": "Verify flag not provided."} validate_check = validate_func(file_path) if validate_check != "success": return validate_check if int(is_verify_flag) == 1: return update_tables(file_path, data) else: return {"status": "SUCCESS", "message": "Validation Successful."}