from datetime import datetime from scripts.mysql_queries import dpd_report_cases_query, dpd_report_data_query from scripts import loan_metrics_calculation from core_utilities import mail_service from utils import execute_query, decrypt_dataframe_columns, zip_file, get_cloudwatch_logger, log_exception import asyncio, pytz, os import pandas as pd local_tz = pytz.timezone("Asia/Kolkata") DPD_DIR = "dpd_report_folder" os.makedirs(DPD_DIR, exist_ok=True) logger = get_cloudwatch_logger(stream_name="dpd_report_script") async def function1(data): mail_result = None elapsed_time = 0 response_time = datetime.now() try: request_time = response_time logger.send_log_event(f"[{request_time.strftime('%Y-%m-%d %H:%M:%S')}] INFO: In dpd_report function - Request received:\n{data}") email_list = data.get("email_list") if not email_list or not isinstance(email_list, list) or len(email_list) == 0: # if proper email_list not provided raise error return {"status": "FAILED", "message": "email_list not provided or empty. Provide recipients email in list []."} dpd_report_cases_query_v1 = dpd_report_cases_query mapping = { # Map input keys to DB columns "lender_id": "lenders.id", "merchant_id": "merchant.id", "anchor_merchant": "anchor.id", "merchant_category": "merchant_category.id", "advisor": "users.id"} # Handle IN clauses for all filters query_params = {} param_counter = 0 for key, column in mapping.items(): values = data.get(key) if values: if not isinstance(values, (list, tuple)): values = [values] # wrap single value in list placeholders = [] for val in values: param_name = f"param_{param_counter}" param_counter += 1 placeholders.append(f":{param_name}") query_params[param_name] = val dpd_report_cases_query_v1 += f" AND {column} IN ({', '.join(placeholders)})" # --- Date filtering --- start_date = data.get("disbursement_date_from") end_date = data.get("disbursement_date_to") if start_date: start_date_obj = datetime.strptime(start_date, "%d-%m-%Y").date() query_params["start_date"] = start_date_obj if end_date: end_date_obj = datetime.strptime(end_date, "%d-%m-%Y").date() query_params["end_date"] = end_date_obj if start_date and end_date: dpd_report_cases_query_v1 += ( " AND date(status_logs.created_at) BETWEEN :start_date AND :end_date" ) elif start_date: dpd_report_cases_query_v1 += ( " AND date(status_logs.created_at) >= :start_date" ) elif end_date: dpd_report_cases_query_v1 += ( " AND date(status_logs.created_at) <= :end_date" ) # get list of application_ids to include in DPD report app_ids = execute_query(dpd_report_cases_query_v1, query_params)['application_id'].to_list() # app_ids = [13306, 12474, 13677, 12431, 10648] # Update bounce penal, POS, TOS, overdue & other amounts prior to foreclosure calculation to get updated data await loan_metrics_calculation({"app_id_list": app_ids}) base_df = execute_query(dpd_report_data_query, params={'application_ids': tuple(app_ids)}) base_df = decrypt_dataframe_columns(base_df, ["Contact Number", "Email", "PAN"]) if data.get("unpaid_charges") == 0: base_df = base_df[(base_df['DPD days'] > 0) & (base_df['Overdue Amount'] > 0)] # Keep only rows where DPD and overdue_amount > 0 else: # Filter rows where DPD = 0 and Overdue Amount > 0 and last_emi_date < current_date base_df['last_emi_date'] = pd.to_datetime(base_df['last_emi_date'], format='%d-%m-%Y').dt.date base_df = base_df[(base_df['DPD days'] == 0) & (base_df['Overdue Amount'] > 0) & (base_df['last_emi_date'] < pd.Timestamp.today().date())] if base_df.empty: return{"status": "INFO", "message": f"No data to show"} base_df.drop(columns=['first_emi_date','last_emi_date'], axis=1, inplace=True) # Compress .xlsx file as zip and share via mail now_str = datetime.now(local_tz).strftime('%Y-%m-%d_%H-%M-%S') for f in os.listdir(DPD_DIR): file_path = os.path.join(DPD_DIR, f) if os.path.isfile(file_path): os.remove(file_path) dpd_report_file = os.path.join(DPD_DIR, f"dpd_report_{now_str}.xlsx") base_df.to_excel(dpd_report_file, index=False) zip_path = dpd_report_file.replace('.xlsx', f'_compressed.zip') zip_file(dpd_report_file, zip_path) msg = f"DPD Report {now_str}" mail_result = mail_service.send_mail_with_attachment(email_list, f"Hi Team\nPFA {msg}", msg, zip_path) try: # after sharing mail delete file from dpd_report_folder os.remove(dpd_report_file) for f in os.listdir(DPD_DIR): file_path = os.path.join(DPD_DIR, f) os.remove(file_path) logger.send_log_event("INFO: Removed file from dpd_report_folder.") except Exception as e: log_exception(logger, f"Error while cleaning up {DPD_DIR}: {e}") response_time = datetime.now() elapsed_time = (response_time - request_time).total_seconds() except Exception as e: log_exception(logger, f"Error while generating DPD report {DPD_DIR}: {e}") if mail_result: logger.send_log_event( f"[{response_time.strftime('%Y-%m-%d %H:%M:%S')}] INFO: DPD report shared via email successfully. " f"Response time: {elapsed_time:.2f} seconds.") return {"status": "Success", "message": f"DPD report shared with {email_list}", "response_time_sec": elapsed_time,} else: logger.send_log_event( f"[{response_time.strftime('%Y-%m-%d %H:%M:%S')}] FAILED: DPD report mail sharing failed. " f"Mail API response: {mail_result}. Response time: {elapsed_time:.2f} seconds.") return {"status": "Failed", "message": "DPD report mail sharing failed.", "response_time_sec": elapsed_time,} # if __name__ == '__main__': # data = { # "lender_name": [], # "merchant_name": [], # "anchor_merchant": [], # "category": [], # "advisor": [], # "disbursement_date_from": "01-03-2025", # "disbursement_date_to": "", # "email_list": ["rudraksha.vaity@creditfair.in"], # "unpaid_charges":1 # } # result = asyncio.run(function1(data)) # print(result)