import sys, os, time import pandas as pd from datetime import datetime, timedelta import numpy_financial as npf # Ensure utils and scripts can be imported sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from utils import execute_query, reconciliation, apply_waiver_allocation, get_cloudwatch_logger, log_exception, zip_file from scripts.mysql_queries import net_loan_irr_query, collection_report_emi_payments_query, collection_report_additional_info_query from core_utilities import mail_service def generate_collection_report_job(payload: dict): logger = get_cloudwatch_logger(stream_name="daily_accounts_collection_report") start_time = time.time() app_id_list = payload.get("app_id_list", []) from_date = payload.get("from_date") to_date = payload.get("to_date") # Date defaults: exactly yesterday if not passed yesterday_str = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') if not from_date or not to_date: from_date = yesterday_str to_date = yesterday_str msg = f"Starting Collection Report. Payload: {payload} | Resolved Dates: {from_date} to {to_date}" print(msg) logger.send_log_event(f"INFO: {msg}") # Build raw transaction query dynamically based on app_id_list app_filter = "" if app_id_list: app_list_str = ','.join(map(str, app_id_list)) app_filter = f"and application_id in ({app_list_str})" raw_tx_query = f""" select application_id, payment_date, amount, payment_for, trasanction_id, mode from payment_transactions where (deleted_at is null and is_deleted_at is null and refunded_at is null) and payment_date >= '{from_date}' and payment_date <= '{to_date}' and payment_for not in ('written_off_principal', 'written_off_interest', 'written_off_bounce', 'written_off_penal', 'Write-off') and exists (select 1 from emi_payments where emi_payments.application_id = payment_transactions.application_id) {app_filter} """ tx_df = execute_query(raw_tx_query) if isinstance(tx_df, str) or tx_df.empty: msg = "No transactions found in this date range." print(msg) logger.send_log_event(f"INFO: {msg}") return # Ensure payment_date is datetime tx_df['payment_date'] = pd.to_datetime(tx_df['payment_date']).dt.date app_ids = tx_df['application_id'].unique().tolist() msg = f"Found {len(app_ids)} unique applications with transactions." print(msg) logger.send_log_event(f"INFO: {msg}") chunk_size = 1000 all_results = [] def agg_rec(df): agg = {} for row in df.itertuples(index=False): pid = row.emi_payment_id due = row.due_date key = (pid, due) if key not in agg: agg[key] = {'pre_emi_paid': 0, 'interest_paid': 0, 'principal_paid': 0, 'bounce_paid': 0, 'penal_paid': 0, 'due_date': due} agg[key]['pre_emi_paid'] += row.pre_emi_paid agg[key]['interest_paid'] += row.interest_paid agg[key]['principal_paid'] += row.principal_paid agg[key]['bounce_paid'] += row.bounce_paid agg[key]['penal_paid'] += row.penal_paid return agg columns = [ 'application_id', 'merchant_name', 'lender_name', 'dpd', 'payment_date', 'total_amount', 'transaction_id', 'mode', 'pre_emi', 'regular_interest', 'regular_principal', 'overdue_interest', 'overdue_principal', 'bounce_collected', 'penal_collected', 'prepartpayment_interest', 'prepartpayment_principal', 'foreclosure', 'settled_principal', 'settled_interest', 'settled_bounce', 'settled_penal', 'processing_fee', 'legal_charges', 'insurance_fee', 'advance_amount', 'cancellation_charges', 'over_payment' ] for i in range(0, len(app_ids), chunk_size): chunk = app_ids[i:i + chunk_size] msg = f"Processing chunk {i//chunk_size + 1} ({len(chunk)} apps)..." print(msg) logger.send_log_event(f"INFO: {msg}") chunk_tx = tx_df[tx_df['application_id'].isin(chunk)] irr_df = execute_query(net_loan_irr_query, params={'application_ids': tuple(chunk)}) if isinstance(irr_df, str): err_msg = f"Error fetching IRR: {irr_df}" print(err_msg) logger.send_log_event(f"ERROR: {err_msg}") continue irr_dict = irr_df.set_index('application_id').to_dict('index') dates_in_chunk = chunk_tx['payment_date'].unique() for target_date in dates_in_chunk: # We don't log this to CloudWatch to avoid spam # print(f" Processing date: {target_date}") target_date_str = target_date.strftime('%Y-%m-%d') target_date_end = f"{target_date_str}" prev_date_str = (target_date - timedelta(days=1)).strftime('%Y-%m-%d') prev_date_end = f"{prev_date_str}" date_tx = chunk_tx[chunk_tx['payment_date'] == target_date] apps_on_date = date_tx['application_id'].unique() open_ep = execute_query(collection_report_emi_payments_query, params={'application_ids': tuple(apps_on_date), 'up_to_date': prev_date_end}) if not isinstance(open_ep, str) and not open_ep.empty: open_ep = apply_waiver_allocation(open_ep) close_ep = execute_query(collection_report_emi_payments_query, params={'application_ids': tuple(apps_on_date), 'up_to_date': target_date_end}) if not isinstance(close_ep, str) and not close_ep.empty: close_ep = apply_waiver_allocation(close_ep) # --- OPTIMIZATION: O(1) Lookups using groupby --- tx_grouped = date_tx.groupby('application_id') open_ep_grouped = open_ep.groupby('application_id') if not isinstance(open_ep, str) and not open_ep.empty else None close_ep_grouped = close_ep.groupby('application_id') if not isinstance(close_ep, str) and not close_ep.empty else None # --- OPTIMIZATION: IRR Cache --- irr_cache = {} for app_id in apps_on_date: row_dict = {col: 0.0 if col not in ['application_id', 'payment_date', 'transaction_id', 'mode', 'merchant_name', 'lender_name', 'dpd'] else '' for col in columns} row_dict['application_id'] = app_id row_dict['payment_date'] = target_date_str app_tx = tx_grouped.get_group(app_id) # O(1) lookup row_dict['total_amount'] = round(app_tx['amount'].sum(), 2) t_ids = set() modes = set() for tx_row in app_tx.itertuples(): p_for = getattr(tx_row, 'payment_for', '') amt = float(getattr(tx_row, 'amount', 0)) tid = str(getattr(tx_row, 'trasanction_id', '')) mod = str(getattr(tx_row, 'mode', '')) if tid and tid != 'None': t_ids.add(tid) if mod and mod != 'None': modes.add(mod) if p_for == 'Foreclosure': row_dict['foreclosure'] += amt elif p_for == 'settled_principal': row_dict['settled_principal'] += amt elif p_for == 'settled_interest': row_dict['settled_interest'] += amt elif p_for == 'settled_bounce': row_dict['settled_bounce'] += amt elif p_for == 'settled_penal': row_dict['settled_penal'] += amt elif p_for == 'processing_fee': row_dict['processing_fee'] += amt elif p_for == 'legal_charges': row_dict['legal_charges'] += amt elif p_for == 'insurance_fee': row_dict['insurance_fee'] += amt elif p_for == 'advance_amount': row_dict['advance_amount'] += amt elif p_for == 'cancellation_charges': row_dict['cancellation_charges'] += amt elif p_for == 'Over Payment': row_dict['over_payment'] += amt row_dict['transaction_id'] = ",".join(filter(None, t_ids)) row_dict['mode'] = ",".join(filter(None, modes)) delta_tx = app_tx[app_tx['payment_for'].isin(['EMI', 'Pre-EMI', 'pre_part_payment'])] if not delta_tx.empty: irr_info = irr_dict.get(app_id) if irr_info: net_loan = float(irr_info['net_loan_amount']) # Fetch from cache if exists if app_id in irr_cache: irr_rate = irr_cache[app_id] else: if irr_info['loan_type'] == 'subvention': irr_rate = 0.00 else: if irr_info['pre_part_flag'] == 1: tenure = irr_info['tenure'] emi_value = irr_info['emi_value'] irr_rate = npf.rate(nper=tenure, pmt=-emi_value, pv=net_loan, fv=0, when='end') else: emis_list = [int(emi) for emi in irr_info['emis_list'].split(',')] cash_flows = [-int(net_loan)] + emis_list irr_rate = npf.irr(cash_flows) irr_cache[app_id] = irr_rate # Save to cache # O(1) Lookups for df_open and df_close df_open = open_ep_grouped.get_group(app_id) if (open_ep_grouped and app_id in open_ep_grouped.groups) else pd.DataFrame(columns=open_ep.columns if not isinstance(open_ep, str) else []) df_close = close_ep_grouped.get_group(app_id) if (close_ep_grouped and app_id in close_ep_grouped.groups) else pd.DataFrame(columns=close_ep.columns if not isinstance(close_ep, str) else []) if not df_close.empty: rec_close_df = reconciliation(df_close, net_loan, irr_rate) rec_close_dict = agg_rec(rec_close_df) if not df_open.empty: rec_open_df = reconciliation(df_open, net_loan, irr_rate) rec_open_dict = agg_rec(rec_open_df) else: rec_open_dict = {} # --- OPTIMIZATION: Eliminate pd.merge --- all_keys = set(rec_close_dict.keys()).union(set(rec_open_dict.keys())) for key in all_keys: pid = key[0] close_row = rec_close_dict.get(key, {}) open_row = rec_open_dict.get(key, {}) p_emi_paid_delta = close_row.get('pre_emi_paid', 0) - open_row.get('pre_emi_paid', 0) int_paid_delta = close_row.get('interest_paid', 0) - open_row.get('interest_paid', 0) pri_paid_delta = close_row.get('principal_paid', 0) - open_row.get('principal_paid', 0) bounce_paid_delta = close_row.get('bounce_paid', 0) - open_row.get('bounce_paid', 0) penal_paid_delta = close_row.get('penal_paid', 0) - open_row.get('penal_paid', 0) if p_emi_paid_delta > 0: row_dict['pre_emi'] += round(p_emi_paid_delta, 2) if bounce_paid_delta > 0: row_dict['bounce_collected'] += round(bounce_paid_delta, 2) if penal_paid_delta > 0: row_dict['penal_collected'] += round(penal_paid_delta, 2) if pid == 0: row_dict['prepartpayment_interest'] += round(int_paid_delta, 2) row_dict['prepartpayment_principal'] += round(pri_paid_delta, 2) else: due_date = close_row.get('due_date') or open_row.get('due_date') due_date_dt = pd.to_datetime(due_date, format='%d-%m-%Y').date() if pd.notna(due_date) else None if due_date_dt and due_date_dt < target_date: row_dict['overdue_interest'] += round(int_paid_delta, 2) row_dict['overdue_principal'] += round(pri_paid_delta, 2) else: row_dict['regular_interest'] += round(int_paid_delta, 2) row_dict['regular_principal'] += round(pri_paid_delta, 2) all_results.append(row_dict) if all_results: final_df = pd.DataFrame(all_results) # Merge Additional Info (merchant_name, lender_name, dpd) msg = "Fetching additional info (merchant, lender, dpd)..." print(msg) logger.send_log_event(f"INFO: {msg}") # Drop empty columns to prevent _x and _y suffixes on merge final_df.drop(columns=['merchant_name', 'lender_name', 'dpd'], inplace=True, errors='ignore') add_info_df = execute_query(collection_report_additional_info_query, params={'application_ids': tuple(app_ids)}) if not isinstance(add_info_df, str) and not add_info_df.empty: final_df = pd.merge(final_df, add_info_df, on='application_id', how='left') # Fill missing values and ensure columns are ordered ordered_cols = [c for c in columns if c in final_df.columns] + [c for c in final_df.columns if c not in columns] final_df = final_df[ordered_cols] final_df.rename(columns={"dpd": "yesterday_dpd"}, inplace=True) final_df.fillna('', inplace=True) # Save and email now_str = datetime.now().strftime('%Y-%m-%d_%H-%M-%S') csv_filename = f"collection_report_{from_date}_to_{to_date}_{now_str}.csv" zip_filename = csv_filename.replace('.csv', '.zip') final_df.to_csv(csv_filename, index=False) zip_file(csv_filename, zip_filename) msg_body = f"Collection Report for {from_date} to {to_date}." email_list = ["accounts@creditfair.in","ravi.mehta@creditfair.in","aditya.vartak@creditfair.in", "santosh@creditfair.in"] mail_result = mail_service.send_mail_with_attachment(email_list, f"Collection Report {from_date} to {to_date}", msg_body, zip_filename) if mail_result: msg = f"Report emailed successfully to {email_list}." print(msg) logger.send_log_event(f"INFO: {msg}") else: msg = f"Failed to send email. Mail result: {mail_result}" print(msg) logger.send_log_event(f"ERROR: {msg}") # Cleanup try: os.remove(csv_filename) os.remove(zip_filename) except Exception as e: log_exception(logger, f"Error cleaning up temporary files: {e}") else: msg = "No results generated." print(msg) logger.send_log_event(f"INFO: {msg}") end_time = time.time() elapsed_time = end_time - start_time msg = f"Collection Report Job Completed. Total Response Time: {elapsed_time:.2f} seconds." print(msg) logger.send_log_event(f"INFO: {msg}") # if __name__ == "__main__": # # For local testing purposes # test_payload = { # "from_date": "2026-08-06", # "to_date": "2026-08-31" # } # generate_collection_report_job(test_payload)