import pandas as pd import json, pytz from datetime import datetime from utils import execute_query, insert_payment_transaction, get_cloudwatch_logger from scripts.mysql_queries import existing_ep_query, existing_rps_query, old_rps_query, overdue_calc_query logger = get_cloudwatch_logger(stream_name="pre_part_payment_upload") # function for checks and validations before updating RPS in database def validate_func(new_rps_file): # Raise error if file type is not excel / csv if not new_rps_file.endswith(('.xlsx', '.csv')): return {"status": "FAILED", "message": "Please upload file in CSV or Excel file format."} new_rps_df = pd.read_excel(new_rps_file) if new_rps_file.endswith('.xlsx') else pd.read_excel(new_rps_file) # Raise error if any column is missing in uploaded file column_checklist = ['application_id','installment_number','emi_date','due_date','interest_comp', 'principal_comp','emi_amount','principal_outstanding'] for col in column_checklist: if col not in new_rps_df.columns: return {"status": "FAILED", "message": f"{col} column does not exist in uploaded file"} # Raise error if any null or empty values in the DataFrame if new_rps_df.isnull().values.any() or (new_rps_df == '').values.any(): return {"status": "FAILED", "message": "Uploaded file should not contain any empty values"} # Raise error if file contains any -ve value numeric_df = new_rps_df.select_dtypes(include='number') if (numeric_df < 0).any().any(): return {"status": "FAILED", "message": "Uploaded file has negative values. Negative values are not allowed"} # Raise error if principal, emi_amount column contains values <= 0 if (new_rps_df['principal_comp'] <= 0).any() or (new_rps_df['emi_amount'] <= 0).any(): return {"status": "FAILED", "message": "Principal or EMI cannot be <= 0"} # Raise error if last pos value is 0 if new_rps_df['principal_outstanding'].iloc[-1] != 0: return {"status": "FAILED", "message": "Last Principal outstanding should be 0"} # Raise error if previous POS values are <= 0 if any(new_rps_df['principal_outstanding'].iloc[:-1] <= 0): return {"status": "FAILED", "message": "All values before the last principal outstanding should be greater than 0"} # Raise error if more than one application_id exist in application_id column if new_rps_df['application_id'].nunique() != 1: return {"status": "FAILED", "message": "The application_id column should contain only one unique value"} # Raise error if 'emi_date' and 'due_date' are not in 'dd-mm-yyyy' format date_format = "%d-%m-%Y" try: # Convert columns to datetime to check if they are in the correct format emi_dates = pd.to_datetime(new_rps_df['emi_date'], format=date_format, errors='raise') due_dates = pd.to_datetime(new_rps_df['due_date'], format=date_format, errors='raise') except ValueError: return {"status": "FAILED", "message": "Both emi_date and due_date columns should be in 'dd-mm-yyyy' format"} # Raise error if day difference between emi_date and due_date is more than 2 day_diff = (emi_dates - due_dates).abs().dt.days if (day_diff > 2).any(): return {"status": "FAILED", "message": "The difference between emi_date and due_date should not be more than 2 days."} application_id = new_rps_df['application_id'].iloc[0] # Raise error if provided application_id is Multi-tranche case - commented to allow pre_part_payment on multi-tranche cases # multi_phase_qry = f"""select distinct application_id from disbursement_log # where (deleted_at is null) and (application_id = {application_id})""" # multi_phase_df = execute_query(multi_phase_qry) # if "Database connection or query failed" in multi_phase_df: # return multi_phase_df # elif not multi_phase_df.empty: # return f"Application id: {application_id} is Multi-tranche case. Pre partpayment not allowed for Multi-tranche cases." # # Raise error if provided application_id has any overdue amount # overdue_query = f"""select application_id from emi_payments where (application_id = {application_id}) # and (date(emi_date) < date(current_date())) group by application_id having sum(amount_due) > 0""" # overdue_df = execute_query(overdue_query) # if "Database connection or query failed" in overdue_df: # return overdue_df # elif not overdue_df.empty: # return {"status": "FAILED", "message": f"Application id: {application_id} has overdue amount. Kindly clear all past due EMIs/Pre_EMI and other charges."} # Raise error if provided application_id lender is not BlackSoil lender_query = f"select lender_id from consumer_application where id = {application_id}" lender_df = execute_query(lender_query) if "Database connection or query failed" in lender_df: return lender_df elif lender_df.empty: return {"status": "FAILED", "message": f"Application id: {application_id} not found."} elif int(lender_df['lender_id'].iloc[0]) != 20: return {"status": "FAILED", "message": f"Application id: {application_id} is not a BlackSoil case. Pre partpayment upload allowed only for BlackSoil cases."} # # Raise error if any emi_date with pre_emi is present in the file # pre_emi_query = f""" # select * # from( # select ep.id,ep.application_id,ep.pre_emi_amount,ep.emi_date,ep.due_date,ep.emi_amount,ep.total_payable,ep.amount_due,ep.payment_status, # coalesce(ne.rn, 0) AS installment_number,ep.updated_at,ep.created_at # from emi_payments ep # left join( # select id,row_number() over (partition by application_id order by emi_date) as rn from emi_payments where emi_amount > 0 # ) ne on ep.id = ne.id # where ep.application_id = {application_id} # )yy where pre_emi_amount > 0 # """ # pre_emi_df = execute_query(pre_emi_query) # if "Database connection or query failed" in pre_emi_df: # return pre_emi_df # elif not pre_emi_df.empty: # pre_emi_df['emi_date'] = pd.to_datetime(pre_emi_df['emi_date']).dt.date # new_rps_df['emi_date'] = pd.to_datetime(new_rps_df['emi_date']).dt.date # common_dates = set(pre_emi_df['emi_date']) & set(new_rps_df['emi_date']) # if common_dates: # dates_str = ', '.join(sorted(date.strftime('%Y-%m-%d') for date in common_dates)) # return ( # f"{dates_str} has Pre-EMI, remove the same from uploaded file." # ) # # Raise error if pre_emi installment number present in file # pre_emi_query = f""" # select * # from( # select ep.id,ep.application_id,ep.pre_emi_amount,ep.emi_date,ep.due_date,ep.emi_amount,ep.total_payable,ep.amount_due,ep.payment_status, # coalesce(ne.rn, 0) AS installment_number,ep.updated_at,ep.created_at # from emi_payments ep # left join( # select id,row_number() over (partition by application_id order by emi_date) as rn from emi_payments where emi_amount > 0 # ) ne on ep.id = ne.id # where ep.application_id = {application_id} # )yy where pre_emi_amount > 0 # """ # pre_emi_df = execute_query(pre_emi_query) # if "Database connection or query failed" in pre_emi_df: # return pre_emi_df # elif not pre_emi_df.empty: # exist_installments = pre_emi_df['installment_number'].to_list() # new_installments = new_rps_df['installment_number'].to_list() # common_installments = list(set(exist_installments) & set(new_installments)) # if common_installments: # return {"status": "FAILED", "message": f"Following installment number have pre_emi {common_installments}. Remove them from uploaded file."} # # Raise error if any emi_date with payment_status = ‘success’ or ‘partial’ is included in the uploaded file. # success_partial_query = f"""select distinct emi_date from emi_payments where (application_id = {application_id}) # and (payment_status != 'pending')""" # success_partial_df = execute_query(success_partial_query) # if "Database connection or query failed" in success_partial_df: # return success_partial_df # elif not success_partial_df.empty: # success_partial_df['emi_date'] = pd.to_datetime(success_partial_df['emi_date']).dt.date # new_rps_df['emi_date'] = pd.to_datetime(new_rps_df['emi_date']).dt.date # common_dates = set(success_partial_df['emi_date']) & set(new_rps_df['emi_date']) # if common_dates: # dates_str = ', '.join(sorted(date.strftime('%Y-%m-%d') for date in common_dates)) # return ( # f"{dates_str} EMI Installation(s) marked as success/partial. These should not be present in the uploaded file." # ) # # Raise error if any installment_number with payment_status = ‘success’ or ‘partial’ is included in the uploaded file. # success_partial_installment_query = f""" # select ep.id,ep.application_id,ep.emi_date,ep.due_date,ep.emi_amount,ep.total_payable,ep.amount_due,ep.payment_status,coalesce(ne.rn, 0) AS installment_number, # ep.updated_at,ep.created_at # from emi_payments ep # left join ( # select # id,row_number() over (partition by application_id order by emi_date) as rn from emi_payments where emi_amount > 0 # ) ne on ep.id = ne.id # where ep.application_id = {application_id} and payment_status != 'pending' # """ # success_partial_installment_df = execute_query(success_partial_installment_query) # if "Database connection or query failed" in success_partial_installment_df: # return success_partial_installment_df # elif not success_partial_installment_df.empty: # exist_installments = success_partial_installment_df['installment_number'].to_list() # new_installments = new_rps_df['installment_number'].to_list() # common_installments = list(set(exist_installments) & set(new_installments)) # if common_installments: # return {"status": "FAILED", "message": f"Following installment number are paid {common_installments}. Remove them from uploaded file."} return "success" # Actual function to update emi_payments, rps_logs payment_transactions & prepartpayment_rps_logs table def update_tables(new_rps_file,data): local_tz = pytz.timezone("Asia/Kolkata") now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') # Prepare uploaded rps by adding total_payable and foreclosure_amount column uploaded_rps_df = pd.read_excel(new_rps_file) if new_rps_file.endswith('.xlsx') else pd.read_excel(new_rps_file) application_id = uploaded_rps_df['application_id'].iloc[0] uploaded_rps_df['total_payable'] = uploaded_rps_df['amount_due'] = uploaded_rps_df['emi_amount'] uploaded_rps_df['foreclosure_amount'] = uploaded_rps_df['principal_outstanding'] + uploaded_rps_df['interest_comp'].shift(-1).fillna(0) uploaded_rps_df = uploaded_rps_df[['installment_number','emi_date','due_date','interest_comp','principal_comp', 'emi_amount','total_payable','amount_due','principal_outstanding','foreclosure_amount']] uploaded_rps_df['emi_date'] = pd.to_datetime(uploaded_rps_df['emi_date'], format='%d-%m-%Y') uploaded_rps_df['emi_date'] = uploaded_rps_df['emi_date'].dt.strftime('%Y-%m-%d') uploaded_rps_df['due_date'] = pd.to_datetime(uploaded_rps_df['due_date'], format='%d-%m-%Y') uploaded_rps_df['due_date'] = uploaded_rps_df['due_date'].dt.strftime('%Y-%m-%d') def update_emi_payments(uploaded_rps_df): existing_ep_df = execute_query(existing_ep_query,params={"application_id":application_id}) existing_ep_df['emi_date'] = pd.to_datetime(existing_ep_df['emi_date']).dt.date existing_ep_df['due_date'] = pd.to_datetime(existing_ep_df['due_date']).dt.date # ------------------ Update EP(emi_payments) block ---------------------------- ep_merged_df = existing_ep_df.merge(uploaded_rps_df,on='installment_number')[['id','application_id','emi_date_y', 'due_date_y','interest_comp','principal_comp','emi_amount_y','total_payable_y','amount_due_y','principal_outstanding', 'foreclosure_amount']].rename(columns={'emi_date_y':'emi_date','due_date_y':'due_date','emi_amount_y':'emi_amount', 'total_payable_y':'total_payable','amount_due_y':'amount_due'}) for _, row in ep_merged_df.iterrows(): update_ep_query = f""" update emi_payments set emi_date = "{row['emi_date']}", due_date = "{row['due_date']}", emi_amount = round({row['emi_amount']}), total_payable = round({row['total_payable']}), amount_due = round({row['amount_due']}), is_pre_part_payment = 1, updated_at = "{now}" where application_id = {application_id} and id = {row['id']} """ execute_query(update_ep_query) logger.send_log_event(f"INFO: Log 1: emi_payments table updated for application_id: {application_id}") # Create sets of installment_number to_insert, to_soft_delete existing_emis = set( existing_ep_df.loc[ (existing_ep_df['payment_status'] == 'pending'), 'installment_number'] ) uploaded_emis = set(uploaded_rps_df['installment_number']) # ---------------- Insert EP block --------------------------- # to insert get all installments which are in uploaded_emis and not in existing_emis # and these installments should be more than last installment in existing_df max_installment_in_existing = max(existing_emis) to_insert = { emis for emis in uploaded_emis - existing_emis if emis > max_installment_in_existing } logger.send_log_event(f"INFO: Log 2: installment_no to_insert in emi_payments : {to_insert}") for emis in to_insert: row = uploaded_rps_df[uploaded_rps_df['installment_number'] == emis].iloc[0] insert_ep_query = f""" insert into emi_payments ( emi_date, pre_emi_amount, emi_amount, bounce_charges, delay_penalty, charges_waived, payment_status, created_at, updated_at, application_id, due_date, total_payable, amount_due, is_pre_part_payment ) values ( "{row['emi_date']}", 0, round({row['emi_amount']}), 0, 0, 0, "pending", "{now}", "{now}", {application_id}, "{row['due_date']}", round({row['total_payable']}), round({row['total_payable']}), 1 ); """ execute_query(insert_ep_query) # -------------- Delete EP block ----------------------- # to delete get all installments which are in existing_emis and not in uploaded_emis # and these installments should be more than last installment in uploaded_df max_installment_in_uploaded = max(uploaded_emis) to_soft_delete = { emis for emis in existing_emis - uploaded_emis if emis > max_installment_in_uploaded } logger.send_log_event(f"INFO: Log 3: installment_no to_delete from emi_payments : {to_soft_delete}") for emis in to_soft_delete: row = existing_ep_df[existing_ep_df['installment_number'] == emis].iloc[0] delete_ep_query = f""" delete from emi_payments where application_id = {application_id} and id = "{row['id']}"; """ execute_query(delete_ep_query) def update_rps_logs(uploaded_rps_df): existing_rps_df = execute_query(existing_rps_query,params={"application_id":application_id}) existing_rps_df['emi_date'] = pd.to_datetime(existing_rps_df['emi_date']).dt.date existing_rps_df['due_date'] = pd.to_datetime(existing_rps_df['due_date']).dt.date # ------------------ Update RPS(rps_logs) block ---------------------------- rps_merged_df = existing_rps_df.merge(uploaded_rps_df,on='installment_number')[['id','installment_number','emi_date_y','due_date_y', 'interest_comp_y','principal_comp_y','emi_amount_y','total_payable_y','principal_outstanding_y', 'foreclosure_amount_y']].rename(columns={'emi_date_y':'emi_date','due_date_y':'due_date', 'interest_comp_y':'interest_comp','principal_comp_y':'principal_comp','emi_amount_y':'emi_amount','total_payable_y':'total_payable', 'principal_outstanding_y':'principal_outstanding','foreclosure_amount_y':'foreclosure_amount'}) for _, row in rps_merged_df.iterrows(): update_rps_query = f""" update rps_logs set installment_number = {row['installment_number']}, emi_date = "{row['emi_date']}", due_date = "{row['due_date']}", interest_comp = {row['interest_comp']}, principal_comp = {row['principal_comp']}, emi_amount = round({row['emi_amount']}), total_payable = round({row['total_payable']}), principal_outstanding = {row['principal_outstanding']}, foreclosure_amount = {row['foreclosure_amount']}, updated_at = "{now}" where application_id = {application_id} and id = {row['id']} """ execute_query(update_rps_query) logger.send_log_event(f"INFO: Log 4: rps_logs table updated for application_id {application_id}") # Create sets of installment_number to_insert, to_soft_delete existing_emis = set( existing_rps_df.loc[ (existing_rps_df['payment_status'] == 'pending'), 'installment_number'] ) uploaded_emis = set(uploaded_rps_df['installment_number']) # ---------------- Insert RPS block --------------------------- max_installment_in_existing = max(existing_emis) to_insert = { emis for emis in uploaded_emis - existing_emis if emis > max_installment_in_existing } logger.send_log_event(f"INFO: Log 5: installment_no to_insert in rps_logs : {to_insert} \n") for emis in to_insert: row = uploaded_rps_df[uploaded_rps_df['installment_number'] == emis].iloc[0] insert_rps_query = f""" insert into rps_logs ( application_id, installment_number, emi_date, due_date, bpi_interest, morat_interest, interest_comp, principal_comp, emi_amount, total_payable, principal_outstanding, foreclosure_amount, created_at, updated_at ) values ( {application_id}, {row['installment_number']}, "{row['emi_date']}", "{row['due_date']}", 0, 0, round({row['interest_comp']},2), round({row['principal_comp']},2), round({row['emi_amount']}), round({row['total_payable']}), round({row['principal_outstanding']},2), round({row['foreclosure_amount']},2), "{now}", "{now}" ); """ execute_query(insert_rps_query) # -------------- Delete EP block ----------------------- # to delete get all installments which are in existing_emis and not in uploaded_emis # and these installments should be more than last installment in uploaded_df max_installment_in_uploaded = max(uploaded_emis) to_soft_delete = { emis for emis in existing_emis - uploaded_emis if emis > max_installment_in_uploaded } logger.send_log_event(f"INFO: Log 6: installment_no to_soft_delete from rps_logs : {to_soft_delete}") for emis in to_soft_delete: row = existing_rps_df[existing_rps_df['installment_number'] == emis].iloc[0] delete_rps_query = f""" update rps_logs set deleted_at = "{now}" where application_id = {application_id} and id = "{row['id']}"; """ execute_query(delete_rps_query) # delete additional records from rps_logs which is not present in emi_payments delete_extra_query = f""" select r1.id,r1.application_id,r1.installment_number,r1.emi_date from (select * from rps_logs where application_id = {application_id} and deleted_at is null) as r1 where r1.emi_date not in ( select emi_date from emi_payments where application_id = {application_id} ) order by r1.emi_date; """ delete_extra_df = execute_query(delete_extra_query) for _, row in delete_extra_df.iterrows(): update_rps_query = f""" update rps_logs set deleted_at = "{now}" where application_id = {application_id} and id = {row['id']}; """ execute_query(update_rps_query) # ----------- Keep logs of changes by making an entry in prepartpayment_rps_logs table ----------------------------- old_rps_df = execute_query(old_rps_query,params={"application_id":application_id}) # get rps before the changes update_emi_payments(uploaded_rps_df) # call update rps function only if application_id present in rps_logs exist_in_rps_logs_query = f""" select distinct application_id from rps_logs where (deleted_at is null) and (application_id = {application_id}) """ exist_in_rps_logs_df = execute_query(exist_in_rps_logs_query) if not exist_in_rps_logs_df.empty: update_rps_logs(uploaded_rps_df) new_rps_df = execute_query(old_rps_query,params={"application_id":application_id}) # get rps after the changes for df in [old_rps_df, uploaded_rps_df, new_rps_df]: # change date dtype to string df[['emi_date', 'due_date']] = df[['emi_date', 'due_date']].astype(str) # get first_emi_date, last_emi_date, first_emi_amount, remaining_emi(net_tenure) and return these in disctionary, # this data will be updated in other table like emi_calculation, consumer_application_element data_dict = { "first_emi_date" : new_rps_df.loc[new_rps_df['emi_amount'] > 0, 'emi_date'].iloc[0], "last_emi_date" : new_rps_df.loc[df.index[-1],'emi_date'], "first_emi_amount" : int(new_rps_df.loc[new_rps_df['emi_amount'] > 0, 'emi_amount'].iloc[0]), "remaining_emi" : int((new_rps_df['emi_amount'] > 0).sum()) } # convert dataframes to json old_rps_df_json = json.dumps(old_rps_df.to_dict(orient='records'),default=str) uploaded_rps_df_json = json.dumps(uploaded_rps_df.to_dict(orient='records'),default=str) new_rps_df_json = json.dumps(new_rps_df.to_dict(orient='records'),default=str) # Insert logs into prepartpayment_rps_logs table now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') insert_logs_query = f""" insert into prepartpayment_rps_logs ( application_id, old_rps, changed_rps, final_rps, user_id, created_at, updated_at ) values ( {application_id}, '{old_rps_df_json}', '{uploaded_rps_df_json}','{new_rps_df_json}', {data.get("user_id")},'{now}','{now}' ); """ execute_query(insert_logs_query) logger.send_log_event(f"INFO: Log 7: record created in prepartpayment_rps_logs.") # Insert pre_part_payment record into payment_transactions table data["payment_for"] = "pre_part_payment" data["application_id"] = application_id insert_payment_transaction(data) logger.send_log_event(f"INFO: Log 8: record of pre_part_payment created in payment_transactions table.") # Return emi_paid count, total_amount_due, amount_over_due in response emi_paid_count_query = f"""select ifnull(count(id),0) emi_paid_count from emi_payments where (deleted_at is null) and (payment_status = 'success') and (emi_amount > 0) and application_id = {application_id}""" emi_paid_df = execute_query(emi_paid_count_query) if "Database connection or query failed" in emi_paid_df: return emi_paid_df elif not emi_paid_df.empty: data_dict['emi_paid'] = int(emi_paid_df['emi_paid_count'].iloc[0]) logger.send_log_event(f"INFO: Log 9: EMI paid count fetched.") # get total_amount_due, amount_over_due params = {'app_ids': [int(application_id)]} overdue_df = execute_query(overdue_calc_query, params=params) if not overdue_df.empty: data_dict['principal_due'] = int(overdue_df['principal_due'].iloc[0]) data_dict['total_principal_due'] = int(overdue_df['total_principal_due'].iloc[0]) data_dict['amount_over_due'] = int(overdue_df['amount_over_due'].iloc[0]) data_dict['total_amount_due'] = int(overdue_df['total_amount_due'].iloc[0]) logger.send_log_event(f"INFO: Log 10: principal_due, total_principal_due, amount_over_due, total_amount_due value fetched.") data_dict['message'] = f"RPS updated for application_id: {application_id}" return data_dict async def function1(data): new_rps_file = data.get("uploaded_file_path") user_id = data.get("user_id") # user_id of user who uploaded rps logger.send_log_event(f"INFO: In pre_part_payment function - Request received: \n{data}") validate_check = validate_func(new_rps_file) if validate_check == "success": return update_tables(new_rps_file,data) else: return validate_check