import os import json import pandas as pd from datetime import datetime import pytz import time from sqlalchemy import text from utils import engine, execute_query from core_utilities.cloudwatch_service import CloudWatchService from scripts.blacksoil_lms.co_lending_regular_allocation import co_lending_regular_allocation, log_exception def get_cloudwatch_logger(stream_name: str = None): log_group = os.environ.get('CLOUDWATCH_GROUP') log_stream = stream_name or os.environ.get('CLOUDWATCH_STREAM') aws_access_key = os.environ.get('CLOUDWATCH_ACCESS_KEY') aws_secret_key = os.environ.get('CLOUDWATCH_SECRET_KEY') return CloudWatchService( log_group_name=log_group, log_stream_name=log_stream, aws_access_key_id=aws_access_key, aws_secret_access_key=aws_secret_key ) logger = get_cloudwatch_logger(stream_name="co_lending_parked_cron") def process_parked_funds(): """ CRON script to consume parked advance collections. Finds all loans with AVAILABLE parked funds, checks if they have any current dues, and if so, drains the parked funds and feeds them into the regular allocation engine. The regular allocation engine will automatically pay dues and re-park any excess. """ try: # 1. Get all loans with available parked funds parked_sql = """ SELECT pt.id as pt_id, pt.loan_id, pt.available_amount, pl.id as original_payment_log_id, pl.payment_date, pl.transaction_number, pl.transaction_mode, pl.utr FROM co_lending_parked_transactions pt LEFT JOIN co_lending_payment_logs pl ON pt.co_lending_payment_logs_id = pl.id WHERE pt.status = 'AVAILABLE' AND pt.available_amount > 0 AND (pl.deleted_at is null) ORDER BY pt.loan_id, pt.created_at ASC """ parked_df = execute_query(parked_sql) if parked_df is None or parked_df.empty: logger.send_log_event("INFO: No parked funds available to process.") return ist_now = datetime.now(pytz.timezone('Asia/Kolkata')).strftime('%Y-%m-%d %H:%M:%S') today_str = datetime.now().strftime("%d-%m-%Y") today_obj = datetime.now().date() def format_loan_id(lid): s = str(lid).strip() if s.endswith('.0') and s[:-2].isdigit(): return s[:-2] return s loan_ids = list(set([format_loan_id(lid) for lid in parked_df['loan_id'].unique()])) loan_dues = {lid: 0.0 for lid in loan_ids} successful_loan_ids = set() if loan_ids: # Batch queries in chunks of 1000 to prevent query string too long chunk_size = 1000 for i in range(0, len(loan_ids), chunk_size): chunk = loan_ids[i:i + chunk_size] format_strings = ','.join([':l_' + str(j) for j in range(len(chunk))]) params_check = {f'l_{j}': lid for j, lid in enumerate(chunk)} # Query expected due expected_sql = f""" SELECT loan_id, SUM(principal_due + interest_due + bpi_due) as total_expected FROM co_lending_emi_payments WHERE loan_id IN ({format_strings}) AND due_date <= CURRENT_DATE() AND rps_type = 'orig' GROUP BY loan_id """ expected_df = execute_query(expected_sql, params=params_check) # Query paid amounts paid_sql = f""" SELECT ep.loan_id, SUM(pt.amount) as total_paid FROM co_lending_payment_transactions pt JOIN co_lending_emi_payments ep ON pt.co_lending_emi_payments_id = ep.id and pt.deleted_at is null and ep.deleted_at is null WHERE ep.loan_id IN ({format_strings}) AND ep.due_date <= CURRENT_DATE() AND ep.rps_type = 'orig' GROUP BY ep.loan_id """ paid_df = execute_query(paid_sql, params=params_check) # Calculate total due = expected - paid if expected_df is not None and not expected_df.empty: for _, expected_row in expected_df.iterrows(): lid = format_loan_id(expected_row['loan_id']) expected = float(expected_row['total_expected']) if pd.notnull(expected_row['total_expected']) else 0.0 paid = 0.0 if paid_df is not None and not paid_df.empty: paid_match = paid_df[paid_df['loan_id'].astype(str) == lid] if not paid_match.empty: paid = float(paid_match.iloc[0]['total_paid']) if pd.notnull(paid_match.iloc[0]['total_paid']) else 0.0 loan_dues[lid] = expected - paid for _, row in parked_df.iterrows(): loan_id = format_loan_id(row['loan_id']) pt_id = int(row['pt_id']) total_available = float(row['available_amount']) payment_date_val = row['payment_date'] if pd.isnull(payment_date_val): p_date_str = today_str else: if hasattr(payment_date_val, 'strftime'): p_date_str = payment_date_val.strftime("%d-%m-%Y") else: p_date_str = str(payment_date_val) txn_num = row['transaction_number'] if pd.notnull(row['transaction_number']) else None txn_mode = row['transaction_mode'] if pd.notnull(row['transaction_mode']) else None utr = row['utr'] if pd.notnull(row['utr']) else None original_payment_log_id = row['original_payment_log_id'] if pd.notnull(row['original_payment_log_id']) else None total_due = loan_dues.get(loan_id, 0.0) if total_due > 0: start_time = time.time() amount_to_drain = min(total_due, total_available) logger.send_log_event(f"INFO: Processing parked funds (pt_id: {pt_id}) for loan_id: {loan_id}, available: {total_available}, due: {total_due}, draining: {amount_to_drain}") try: # 3. Drain the specific parked transaction drain_sql = """ UPDATE co_lending_parked_transactions SET available_amount = available_amount - :amount, status = CASE WHEN available_amount <= 0.01 THEN 'EXHAUSTED' ELSE 'AVAILABLE' END, updated_at = :ist_now WHERE id = :pt_id AND status = 'AVAILABLE' """ with engine.begin() as conn: conn.execute(text(drain_sql), {"pt_id": pt_id, "amount": amount_to_drain, "ist_now": ist_now}) # 4. Feed into regular allocation engine with metadata payload = { "loan_id": loan_id, "dry_run": False, "allocation_type": "ADVANCE_COLLECTION", "payment_list": { "amount": amount_to_drain, "payment_date": p_date_str, "transaction_number": txn_num, "transaction_mode": txn_mode, "utr": utr, "original_payment_log_id": original_payment_log_id } } result = co_lending_regular_allocation(payload) if "error" in result: logger.send_log_event(f"ERROR: Failed to allocate parked funds for loan {loan_id} (pt_id {pt_id}): {result['error']}") else: loan_dues[loan_id] -= amount_to_drain successful_loan_ids.add(loan_id) execution_time_ms = int((time.time() - start_time) * 1000) print(f"[{ist_now}] Processed loan_id: {loan_id} | Response Time: {execution_time_ms}ms") except Exception as e: logger.send_log_event(f"ERROR: Error processing loan {loan_id} in parked cron: {str(e)}") log_exception(logger, f"Error processing loan {loan_id} in parked cron: {str(e)}") if successful_loan_ids: from scripts.blacksoil_lms.co_lending_bounce_penal import process_bounce_penal_updates logger.send_log_event(f"INFO: Triggering bounce/penal & metrics updates for {len(successful_loan_ids)} loans.") process_bounce_penal_updates(loan_ids=list(successful_loan_ids)) except Exception as e: logger.send_log_event(f"FATAL: Error in parked cron: {str(e)}") log_exception(logger, f"Fatal error in parked cron: {str(e)}") finally: try: engine.dispose() from utils import engine_read engine_read.dispose() except: pass if __name__ == "__main__": from dotenv import load_dotenv load_dotenv() process_parked_funds()