from sqlalchemy import create_engine, text from sqlalchemy.exc import SQLAlchemyError import os, requests, json, pytz, ast, zipfile, hmac, hashlib, base64, traceback, time, calendar, math from dotenv import load_dotenv import pandas as pd import numpy as np import numpy_financial as npf from hashids import Hashids from datetime import datetime from Crypto.Cipher import AES from Crypto.Util.Padding import unpad from core_utilities.cloudwatch_service import CloudWatchService from core_utilities.cf_encrypt_decrypt import wall_decrypt import mysql.connector load_dotenv() local_tz = pytz.timezone("Asia/Kolkata") # Create SQLAlchemy engine using pymysql MYSQL_USER = os.getenv("MYSQL_USER") MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD") MYSQL_HOST = os.getenv("MYSQL_HOST") MYSQL_DB = os.getenv("MYSQL_DB") MYSQL_HOST_READ = os.getenv("MYSQL_HOST_READ") MYSQL_PORT = os.getenv("DB_PORT") db_url = f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}" # print(f"Engine: {db_url}") # added to debug db connection engine = create_engine(db_url, pool_pre_ping=True) # db_url_read = f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST_READ}/{MYSQL_DB}" # engine_read = create_engine(db_url_read, pool_pre_ping=True) db_url_read = f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST_READ}:{MYSQL_PORT}/{MYSQL_DB}" # print(f"Engine read: {db_url_read}") # added to debug db connection engine_read = create_engine(db_url_read, pool_pre_ping=True) TC_CODES = [ "RNR", "SW", "NC", "CD", "PTP", "WANT SETT", "CB", "SETT", "RTP-DSP", "RTP-FI", "RTP-AP", "BPTP", "EXPIRED", "WRN_NUM", "RTP-NITP", "LM", "CUSTODY", "RTP-SI", ] FOS_CODES = [ "ANF", "DL", "SHIFTED", "PTP", "CB", "SETT", "Want SETT", "RTP-DSP", "RTP-FI", "RTP-AP", "BPTP", "EXPIRED", "RTP-NITP", "LM", "CUSTODY", "RTP-SI", "SKIP", "NSP", "ENA", ] # Utility function to execute SQL queries def execute_query(sql_query, params=None): """ Executes a SQL query using SQLAlchemy. Parameters: sql_query (str): The SQL query to execute. params (dict, optional): Parameters to safely bind to the SQL query. Returns: pd.DataFrame | str: - For SELECT or WITH queries: returns result as a pandas DataFrame. - For other queries (INSERT, UPDATE, DELETE): returns success message. Raises: RuntimeError: If a database or unexpected error occurs. Notes: - Uses `engine` for database connection. - Automatically commits non-SELECT queries within a transaction block. """ try: sql_text = text(sql_query) if sql_query.strip().lower().startswith(("select", "with")): with engine_read.connect() as conn: # print(f"Engine read from execute query: {db_url_read}") # added to debug db connection return pd.read_sql(sql_text, conn, params=params) else: # print(f"Engine from execute query: {db_url}") # added to debug db connection with engine.begin() as conn: if isinstance(params, list) and params: # <-- New line conn.execute(sql_text, params) # executemany mode else: conn.execute(sql_text, params or {}) return "Query executed successfully" except SQLAlchemyError as e: raise RuntimeError(f"Database error: {e}") from e except Exception as e: raise RuntimeError(f"Unexpected error: {e}") from e # Upload document function def upload_docs(file_type, app_id, file_string, file_name, user_id): """ Uploads a document to the checkout service API. Parameters: file_type (str): File format (e.g., 'pdf', 'jpg'). app_id (int): Application ID (will be encoded using Hashids). file_string (str): Base64-encoded file content. file_name (str): Name of the file (e.g., 'aadhaar_front.pdf'). user_id (str): ID of the user uploading the document. Returns: dict: API response as a dictionary. On success, returns parsed JSON. On failure, returns a dict with 'status' and error details. Notes: - Uses environment variables: CHECKOUT_URL, CHECKOUT_CLIENT_ID, CHECKOUT_CLIENT_SECRET, CHECKOUT_HASH_KEY. - `app_id` is encoded via Hashids for obfuscation. """ hashids = Hashids(salt=os.environ.get("CHECKOUT_HASH_KEY"), min_length=20) api_url = f'{os.environ.get("CHECKOUT_URL")}/api/upload-document' headers = { "Content-Type": "application/json", "X-Client-Id": os.environ.get("CHECKOUT_CLIENT_ID"), "X-Client-Secret": os.environ.get("CHECKOUT_CLIENT_SECRET") } data = { "application_id": hashids.encode(app_id), "docName": file_name, "docFormat": file_type, "docPassword": "", "docContent": file_string, "user_id": user_id } try: response = requests.post(url=api_url, data=json.dumps(data), headers=headers) if response.status_code == 200: return response.json() else: return { "status": "ERROR", "message": f"HTTP {response.status_code}: {response.reason}", "response_text": response.text } except requests.RequestException as e: return { "status": "ERROR", "message": f"Request failed: {e}" } except json.JSONDecodeError as e: return { "status": "ERROR", "message": f"Invalid JSON response: {e}", "response_text": response.text } def build_payment_transaction_sql(data): """Builds the INSERT statement for a single payment_transactions row. Extracted so both insert_payment_transaction() (single, own transaction) and insert_payment_transactions_atomic() (many rows, one transaction) share the exact same column mapping and stay in sync. """ now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') formatted_payment_date = data.get("payment_date") or None formatted_payment_recon_date = data.get("payment_reconciliation_date") or None insert_pt_query = f""" insert into payment_transactions ( emi_payment_id, amount, mode, trasanction_id, trasanction_acknowledgement, payment_date, status, created_at, updated_at, paid_by, collected_by, wallet_name, reason, umrn, application_id, payment_for, payment_reconciliation_date, cf_transaction_id, bank_name, bank_id, deleted_at, is_deleted_at, deleted_by, refunded_at, collected_by_type, adjust_with_disbursement, adjusted_to, adjusted_by, differ_id, refund_maker_id, refund_maker_at, settlement_date ) values ( 0, {int(data.get("amount") or 0)}, '{data.get("mode", "")}', {f"'{data['trasanction_id']}'" if data.get("trasanction_id") else 'NULL'}, {f"'{data['trasanction_acknowledgement']}'" if data.get("trasanction_acknowledgement") else 'NULL'}, {f"'{formatted_payment_date}'" if formatted_payment_date else 'NULL'}, 'success', '{now}', '{now}', {f"'{data['paid_by']}'" if data.get("paid_by") else 'NULL'}, {f"'{data['collected_by']}'" if data.get("collected_by") else 'NULL'}, {f"'{data['wallet_name']}'" if data.get("wallet_name") else 'NULL'}, {f"'{data['reason']}'" if data.get("reason") else 'NULL'}, {f"'{data['umrn']}'" if data.get("umrn") else 'NULL'}, {data.get("application_id")}, '{data.get("payment_for","")}', {f"'{formatted_payment_recon_date}'" if formatted_payment_recon_date else 'NULL'}, {f"'{data['cf_transaction_id']}'" if data.get("cf_transaction_id") else 'NULL'}, {f"'{data['bank_name']}'" if data.get("bank_name") else 'NULL'}, {int(data.get("bank_id") or 0)}, NULL, NULL, NULL, NULL, {data.get("collected_by_type") if data.get("collected_by_type") is not None else 'NULL'}, {data.get("adjust_with_disbursement") if data.get("adjust_with_disbursement") is not None else 'NULL'}, {data.get("adjusted_to") if data.get("adjusted_to") is not None else 'NULL'}, {data.get("adjusted_by") if data.get("adjusted_by") is not None else 'NULL'}, {data.get("differ_id") if data.get("differ_id") is not None else 'NULL'}, {data.get("refund_maker_id") if data.get("refund_maker_id") is not None else 'NULL'}, {f"'{data['refund_maker_at']}'" if data.get("refund_maker_at") else 'NULL'}, {f"'{data['settlement_date']}'" if data.get("settlement_date") else 'NULL'} ); """ return insert_pt_query def insert_payment_transaction(data): execute_query(build_payment_transaction_sql(data)) def insert_payment_transactions_atomic(rows): """Inserts multiple payment_transactions rows inside a SINGLE transaction. Either every row is committed or none are (all-or-nothing). Use this for the foreclosure allocation so the monolith + all breakup splits can never end up partially written if the process dies mid-way. Parameters: rows (list[dict]): each dict is a payment_transactions row (same shape as insert_payment_transaction's `data`). """ if not rows: return "No rows to insert" with engine.begin() as conn: for row in rows: conn.execute(text(build_payment_transaction_sql(row))) return "Query executed successfully" # Calculate Bounce Penal Function def calculate_bounce_penal_function(df): """ Calculates bounce charges and delay penalties on EMIs based on payment history. Parameters: df (pd.DataFrame): DataFrame containing loan-related columns such as: - 'emi_date', 'due_date', 'emi_amount', 'pre_emi_amount', 'charges_waived', 'total_paid' - 'payment_list': list of {date: amount} strings (to be parsed) Returns: pd.DataFrame: Updated DataFrame with calculated columns: - 'bounce_charges', 'delay_days', 'delay_penalty', 'total_payable', 'amount_due' """ # --- Embedded lender-specific rules --- lender_bounce_penal_rules = { 'bounce': {3: 500, 4: 0, 9: 350, 10: 300, 20: 300}, 'penal': {3: (2*12/365)/100, 4: (2*12/365)/100, 9: (2*12/365)/100, 10: 0.0006, 20: 0.0006}, 'default': {'bounce': 300, 'penal': 0.0015} # 0.15% daily } df['emi_date'] = pd.to_datetime(df['emi_date'], format='%d-%m-%Y').dt.date df['due_date'] = pd.to_datetime(df['due_date'], format='%d-%m-%Y').dt.date df['payment_list'] = df['payment_list'].apply(ast.literal_eval) df.loc[:, ['bounce_charges', 'delay_days', 'delay_penalty']] = 0 df.sort_values(by='due_date', inplace=True) df['payment_date'] = pd.NaT # Set bounce charges current_date = pd.Timestamp.today().date() for ind in df.index: if df.at[ind, 'emi_amount'] <= 0 or df.at[ind, 'due_date'] >= current_date: continue # if lender_id present in lender_bounce_penal_rules dict then apply bounce w.r.t dict else apply default bounce lender_id = df.at[ind, 'lender_id'] bounce_amount = lender_bounce_penal_rules['bounce'].get(lender_id, lender_bounce_penal_rules['default']['bounce']) paid_upto_now = sum( float(list(payment.values())[0]) for payment in df.at[ind, 'payment_list'] if datetime.strptime(list(payment.keys())[0], '%d-%m-%Y').date() <= df.at[ind, 'due_date'] ) df.at[ind, 'bounce_charges'] = 0 if paid_upto_now >= df.at[ind, 'emi_amount'] * 0.9 else bounce_amount # Set penal charges for ind in df.index: if df.at[ind, 'emi_amount'] <= 0 or df.at[ind, 'due_date'] >= current_date: continue # if lender_id present in lender_bounce_penal_rules dict then apply penal w.r.t dict else apply default penal lender_id = df.at[ind, 'lender_id'] daily_rate = lender_bounce_penal_rules['penal'].get(lender_id, lender_bounce_penal_rules['default']['penal']) pending_amount = df.at[ind, 'emi_amount'] df.at[ind, 'delay_days'] = 0 df.at[ind, 'delay_penalty'] = 0 df.at[ind, 'payment_date'] = pd.NaT for payment_dict in df.at[ind, 'payment_list']: payment_date = datetime.strptime(list(payment_dict.keys())[0], '%d-%m-%Y').date() amount_paid = float(list(payment_dict.values())[0]) if pd.isna(df.at[ind, 'payment_date']): delay_days = max(0, (payment_date - df.at[ind, 'due_date']).days) else: delay_days = max(0, (payment_date - df.at[ind, 'payment_date']).days - 1) df.at[ind, 'delay_days'] += round(delay_days) if pending_amount > 0 else 0 df.at[ind, 'delay_penalty'] += round(pending_amount * daily_rate * delay_days) if pending_amount > 0 else 0 df.at[ind, 'payment_date'] = payment_date if payment_date > df.at[ind, 'due_date'] else pd.NaT pending_amount -= amount_paid df['total_payable'] = round(df['pre_emi_amount'] + df['emi_amount'] + df['bounce_charges'] + df['delay_penalty'] - df['charges_waived'], 0) df['amount_due'] = round(np.maximum(0, df['total_payable'] - df['total_paid']), 0) return df # Compress file to ZIP function def zip_file(input_file_path, output_zip_path): """ Compresses a single file into a ZIP archive. Parameters: input_file_path (str): Path to the file to compress. output_zip_path (str): Path to save the resulting .zip file. Notes: The file is stored in the ZIP using its base name only. Example: zip_file("logs/output.txt", "archive/output.zip") """ with zipfile.ZipFile(output_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: zipf.write(input_file_path, arcname=os.path.basename(input_file_path)) # --- Decrypt DataFrame Columns --- def decrypt_dataframe_columns(df: pd.DataFrame, columns: list) -> pd.DataFrame: """ Decrypts specified columns in a pandas DataFrame using the `wall_decrypt` function. Parameters: df (pd.DataFrame): The input DataFrame containing encrypted data. columns (list): A list of column names (strings) in `df` that should be decrypted. Returns: pd.DataFrame: A copy of the original DataFrame with the specified columns decrypted (if they exist in the DataFrame). Non-null values in those columns are passed to `wall_decrypt`, nulls are left unchanged. Example: decrypted_df = decrypt_dataframe_columns(df, ['email', 'username']) """ decrypted_df = df.copy() for col in columns: if col in decrypted_df.columns: decrypted_df[col] = decrypted_df[col].map(lambda x: wall_decrypt(x,key="CRDRIAF") if pd.notnull(x) else x) return decrypted_df # --- CloudWatch Logging --- def get_cloudwatch_logger(stream_name: str = None): """ Initializes and returns a CloudWatchService logger instance. Reads configuration from environment variables: - CLOUDWATCH_GROUP: The name of the CloudWatch log group. - CLOUDWATCH_STREAM: The name of the CloudWatch log stream. - CLOUDWATCH_ACCESS_KEY: AWS access key for CloudWatch. - CLOUDWATCH_SECRET_KEY: AWS secret key for CloudWatch. Returns: CloudWatchService: An instance of the CloudWatchService for logging. """ log_group = os.environ.get('CLOUDWATCH_GROUP') log_stream = 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 ) def log_exception(logger, message="Exception occurred"): """ Logs an exception with a full stack trace to CloudWatch. Parameters: logger (CloudWatchService): The logger instance. message (str): A custom message to prepend to the log. """ error_stack_trace = traceback.format_exc() log_message = f"EXCEPTION: {message}\n{error_stack_trace}" if logger: logger.send_log_event(log_message) def send_communication_logs(mobile_number, message, request_url, response,provider,_id, id_type, created_by): now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') try: query = """ INSERT INTO sms_logs ( mobile_number, message, request_url, response, provider, created_at, _id, id_type, created_by ) VALUES ( :mobile_number, :message, :request_url, :response, :provider, :created_at, :_id, :id_type, :created_by ) """ params = { "mobile_number": mobile_number, "message": message, "request_url": request_url, "response": response, "provider": provider, "created_at": now, "_id": _id, "id_type": id_type, "created_by": created_by } execute_query(query, params) return True except Exception as e: print("DB Insert Error:", e) return False # Homville Token Manager function (Authentication function) def homeville_token_manager(client_id, client_secret, auth_url, token_dir="homeville_dir", token_filename="homeville_token.json"): """ Manages Homeville API authentication token by storing it locally, checking expiration, and refreshing it as needed. Returns a function to make authenticated API requests. Parameters: ---------- client_id : str The Homeville API client ID. client_secret : str The Homeville API client secret. auth_url : str The authentication endpoint URL to fetch the token. token_dir : str, optional Directory to store the token file (default "homeville_dir"). token_filename : str, optional Name of the token file (default "homeville_token.json"). Returns: ------- request : function A function to make authenticated requests to the Homeville API. Internal Functions: ------------------- - `load_token_from_file`: Loads token and expiry from file if it exists. - `save_token_to_file`: Saves token and expiry to file. - `authenticate`: Fetches a new token and expiry from the API. - `get_token`: Returns a valid token, refreshing it if expired. - `get_auth_header`: Returns the authentication header with the token. - `request`: Makes an authenticated request and retries if the token is expired. """ token_file = os.path.join(token_dir, token_filename) os.makedirs(token_dir, exist_ok=True) # Create folder to store JSON token file, ignore if exists token = None expiry = 0 def load_token_from_file(): """Load token + expiry from file if it exists.""" nonlocal token, expiry if not os.path.exists(token_file): return try: with open(token_file, "r") as f: data = json.load(f) token = data.get("token") expiry = data.get("expiry", 0) except Exception as e: print(f"Error loading token from file: {e}") token = None expiry = 0 def save_token_to_file(): """Save current token + expiry to file.""" nonlocal token, expiry # nonlocal uses variables assigned in parent function data = {"token": token, "expiry": expiry} with open(token_file, "w") as f: json.dump(data, f) def authenticate(): """Call auth API and store new token.""" nonlocal token, expiry payload = {"client_id": client_id, "client_secret": client_secret} headers = {"Accept": "application/json", "Content-Type": "application/json"} try: print("Fetching new token from Homeville auth API...") res = requests.post(auth_url, json=payload, headers=headers) res.raise_for_status() response = res.json() token = response["token"] expires_in = response["expires_in"] expiry = time.time() + expires_in - 60 save_token_to_file() except requests.exceptions.RequestException as e: print(f"Error occurred while requesting the token: {e}") raise # Re-raise the exception if needed def get_token(): """Return valid token; refresh if missing or expired.""" nonlocal token, expiry if not token or time.time() >= expiry: authenticate() return token def get_auth_header(): """Return header with valid token.""" return {"Authorization": f"Bearer {get_token()}"} def request(method, url, **kwargs): """Make an authenticated request. If 401, refresh token and retry once.""" headers = kwargs.pop("headers", {}) headers.update(get_auth_header()) kwargs["headers"] = headers response = requests.request(method, url, **kwargs) # If token expired on server-side → refresh and retry ONCE if response.status_code == 401: print("Token expired → refreshing...") authenticate() headers["Authorization"] = f"Bearer {token}" response = requests.request(method, url, headers=headers, **kwargs) return response load_token_from_file() # Try to load token from file return request def get_homeville_data(api_url, client_id=None, client_secret=None, homeville_auth_url=None): """ This functioon call homeville_token_manager() for authentication and returns response for passed url """ client_id = client_id or os.getenv("HOMEVILLE_CLIENT_ID") client_secret = client_secret or os.getenv("HOMEVILLE_SECRET_KEY") hm_auth_url = homeville_auth_url or os.getenv("HOMEVILLE_AUTH_URL") if not client_id or not client_secret or not hm_auth_url: raise ValueError("Missing HOMEVILLE_CLIENT_ID or HOMEVILLE_SECRET_KEY or HOMEVILLE_AUTH_URL in env.") api_request = homeville_token_manager(client_id=client_id, client_secret=client_secret, auth_url=hm_auth_url) try: response = api_request("GET", api_url) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"Error during request: {e}") raise def send_communication_logs(mobile_number, message, request_url, response,provider,_id, id_type, created_by): now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') try: query = """ INSERT INTO sms_logs ( mobile_number, message, request_url, response, provider, created_at, _id, id_type, created_by ) VALUES ( :mobile_number, :message, :request_url, :response, :provider, :created_at, :_id, :id_type, :created_by ) """ params = { "mobile_number": mobile_number, "message": message, "request_url": request_url, "response": response, "provider": provider, "created_at": now, "_id": _id, "id_type": id_type, "created_by": created_by } execute_query(query, params) return True except Exception as e: print("DB Insert Error:", e) return False def send_communication_logs(mobile_number, message, request_url, response,provider,_id, id_type, created_by): now = datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S') try: query = """ INSERT INTO sms_logs ( mobile_number, message, request_url, response, provider, created_at, _id, id_type, created_by ) VALUES ( :mobile_number, :message, :request_url, :response, :provider, :created_at, :_id, :id_type, :created_by ) """ params = { "mobile_number": mobile_number, "message": message, "request_url": request_url, "response": response, "provider": provider, "created_at": now, "_id": _id, "id_type": id_type, "created_by": created_by } execute_query(query, params) return True except Exception as e: print("DB Insert Error:", e) return False # Homville Token Manager function (Authentication function) def homeville_token_manager(client_id, client_secret, auth_url, token_dir="homeville_dir", token_filename="homeville_token.json"): """ Manages Homeville API authentication token by storing it locally, checking expiration, and refreshing it as needed. Returns a function to make authenticated API requests. Parameters: ---------- client_id : str The Homeville API client ID. client_secret : str The Homeville API client secret. auth_url : str The authentication endpoint URL to fetch the token. token_dir : str, optional Directory to store the token file (default "homeville_dir"). token_filename : str, optional Name of the token file (default "homeville_token.json"). Returns: ------- request : function A function to make authenticated requests to the Homeville API. Internal Functions: ------------------- - `load_token_from_file`: Loads token and expiry from file if it exists. - `save_token_to_file`: Saves token and expiry to file. - `authenticate`: Fetches a new token and expiry from the API. - `get_token`: Returns a valid token, refreshing it if expired. - `get_auth_header`: Returns the authentication header with the token. - `request`: Makes an authenticated request and retries if the token is expired. """ token_file = os.path.join(token_dir, token_filename) os.makedirs(token_dir, exist_ok=True) # Create folder to store JSON token file, ignore if exists token = None expiry = 0 def load_token_from_file(): """Load token + expiry from file if it exists.""" nonlocal token, expiry if not os.path.exists(token_file): return try: with open(token_file, "r") as f: data = json.load(f) token = data.get("token") expiry = data.get("expiry", 0) except Exception as e: print(f"Error loading token from file: {e}") token = None expiry = 0 def save_token_to_file(): """Save current token + expiry to file.""" nonlocal token, expiry # nonlocal uses variables assigned in parent function data = {"token": token, "expiry": expiry} with open(token_file, "w") as f: json.dump(data, f) def authenticate(): """Call auth API and store new token.""" nonlocal token, expiry payload = {"client_id": client_id, "client_secret": client_secret} headers = {"Accept": "application/json", "Content-Type": "application/json"} try: print("Fetching new token from Homeville auth API...") res = requests.post(auth_url, json=payload, headers=headers) res.raise_for_status() response = res.json() token = response["token"] expires_in = response["expires_in"] expiry = time.time() + expires_in - 60 save_token_to_file() except requests.exceptions.RequestException as e: print(f"Error occurred while requesting the token: {e}") raise # Re-raise the exception if needed def get_token(): """Return valid token; refresh if missing or expired.""" nonlocal token, expiry if not token or time.time() >= expiry: authenticate() return token def get_auth_header(): """Return header with valid token.""" return {"Authorization": f"Bearer {get_token()}"} def request(method, url, **kwargs): """Make an authenticated request. If 401, refresh token and retry once.""" headers = kwargs.pop("headers", {}) headers.update(get_auth_header()) kwargs["headers"] = headers response = requests.request(method, url, **kwargs) # If token expired on server-side → refresh and retry ONCE if response.status_code == 401: print("Token expired → refreshing...") authenticate() headers["Authorization"] = f"Bearer {token}" response = requests.request(method, url, headers=headers, **kwargs) return response load_token_from_file() # Try to load token from file return request def get_homeville_data(api_url, client_id=None, client_secret=None, homeville_auth_url=None): """ This functioon call homeville_token_manager() for authentication and returns response for passed url """ client_id = client_id or os.getenv("HOMEVILLE_CLIENT_ID") client_secret = client_secret or os.getenv("HOMEVILLE_SECRET_KEY") hm_auth_url = homeville_auth_url or os.getenv("HOMEVILLE_AUTH_URL") if not client_id or not client_secret or not hm_auth_url: raise ValueError("Missing HOMEVILLE_CLIENT_ID or HOMEVILLE_SECRET_KEY or HOMEVILLE_AUTH_URL in env.") api_request = homeville_token_manager(client_id=client_id, client_secret=client_secret, auth_url=hm_auth_url) try: response = api_request("GET", api_url) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"Error during request: {e}") raise def get_hm_rps(app_id): """ This functioon call homevilles' RPS API and returns dataframe as response for given application_id """ # Fetch homeville case_id for given application_id hm_loans_fetch_query = """ select homeville_loans.case_id from homeville_loans -- join consumer_application on consumer_application.id = homeville_loans.application_id -- and consumer_application.deleted_at is null and consumer_application.is_deleted is null and consumer_application.merchant_id not in (73,9421) -- and consumer_application.state in ('disbursed','nach registered') and consumer_application.lender_id = 10 where homeville_loans.application_id = :application_id """ params = {'application_id':app_id} hm_loans_df = execute_query(hm_loans_fetch_query, params=params) if hm_loans_df.empty: return {"status": "Failed", "message": f"No case_id present in homville_loans table for application_id: {app_id}"} # Request RPS from homeville case_id = hm_loans_df['case_id'].iloc[0] rps_api_url = f"https://api-v2.bharathousingnetwork.com/api/v1/cases/{case_id}/repayment-schedule" rps_api_response = get_homeville_data(rps_api_url) if rps_api_response.get('status') != 'success': return rps_api_response rps_json = rps_api_response['data']['schedule']['user'] # print(f"Homeville RPS fetched: {rps_json}") hm_rps_df = pd.DataFrame(rps_json) # hm_rps_df = pd.read_json("D:\\Python scripts\\cf-python-scripts\\testing_dir\\rps_normal.json") hm_rps_df['application_id'] = app_id hm_rps_df['repay_date'] = pd.to_datetime(hm_rps_df['repay_date']).dt.date hm_rps_df['repaid_at'] = pd.to_datetime(hm_rps_df['repaid_at']).dt.date cols = ['repay_amount', 'repay_interest', 'repay_principle', 'repay_os', 'interest_collected', 'principle_collected'] hm_rps_df[cols] = hm_rps_df[cols].apply(pd.to_numeric, errors='coerce').fillna(0) hm_rps_df['amount_paid'] = hm_rps_df['interest_collected'] + hm_rps_df['principle_collected'] if hm_rps_df['repayment_type'].iloc[0] == 'PEMI': hm_rps_df['installment_number'] = hm_rps_df.reset_index().index else: hm_rps_df['installment_number'] = hm_rps_df.reset_index().index + 1 hm_rps_df.loc[hm_rps_df['repayment_type'] == 'part_payment', 'installment_number'] = 0 # Set installment_number to 0 where repayment_type is 'part_payment' hm_rps_df.loc[hm_rps_df['installment_number'] > 0, 'installment_number'] = range(1, len(hm_rps_df[hm_rps_df['installment_number'] > 0]) + 1) # fix the numbering again hm_rps_df = hm_rps_df[['application_id','installment_number','repay_date','repay_amount','repay_interest','repay_principle','amount_paid','repay_os','pending_emi_amount','repaid_at','repayment_type','interest_rate']] return hm_rps_df # Apply waiver allocation function - prepartpayment.py def apply_waiver_allocation(df: pd.DataFrame) -> pd.DataFrame: """ Applies charges_waived to delay_penalty, bounce_charges, emi_amount, and pre_emi_amount in that exact order, overwriting the original columns. """ used_on_delay = np.minimum(df['charges_waived'], df['delay_penalty']) df['delay_penalty'] = df['delay_penalty'] - used_on_delay remaining_waiver = df['charges_waived'] - used_on_delay used_on_bounce = np.minimum(remaining_waiver, df['bounce_charges']) df['bounce_charges'] = df['bounce_charges'] - used_on_bounce remaining_waiver = remaining_waiver - used_on_bounce used_on_emi = np.minimum(remaining_waiver, df['emi_amount']) df['emi_amount'] = df['emi_amount'] - used_on_emi remaining_waiver = remaining_waiver - used_on_emi used_on_pre_emi = np.minimum(remaining_waiver, df['pre_emi_amount']) df['pre_emi_amount'] = df['pre_emi_amount'] - used_on_pre_emi return df # Apply waiver allocation function - foreclosure_breakup.py, update_loans_data.py, prepartpayment.py def reconciliation(df, loan_principal, irr_rate): beginning_balance = loan_principal results = [] previous_emi_date = pd.NaT previous_emipayment_id = None annual_interest_rate = irr_rate * 12 # tenure = len(df[df['emi_amount'] > 0]) for row in df.itertuples(index=False): pre_emi = getattr(row, 'pre_emi_amount', 0) moratorium_amount = getattr(row, 'moratorium_amount', 0) emi_amount = getattr(row, 'emi_amount', 0) # emi_amount = round(npf.pmt(annual_interest_rate/12,tenure,-loan_principal,0),2) if getattr(row, 'emi_amount', 0) != 0 else 0 bounce = getattr(row, 'bounce_charges', 0) penal = getattr(row, 'delay_penalty', 0) total_paid = getattr(row, 'total_paid', 0) total_paid_for_bpi = getattr(row,'total_paid_for_bpi', 0) emi_date = getattr(row, 'emi_date', pd.NaT) emi_date = pd.to_datetime(emi_date, format='%d-%m-%Y') emi_payment_id = getattr(row, 'emi_payment_id', 0) opening_pos = beginning_balance # assign opening balance remaining_amount = total_paid # begin allocation from total_paid(for EMI/Pre-part payment) of that EMI # Initialize variables to avoid leakage from previous loop iterations pre_emi_paid = 0 morat_paid = 0 interest_paid = 0 principal_paid = 0 bounce_paid = 0 penal_paid = 0 if pd.notna(previous_emi_date) and emi_payment_id == 0: # Calculation for pre part payment days_since_old_pos = (emi_date - previous_emi_date).days # calculate days_diff betn prev emi date and current emi date(prepartpayment date) interest_component = round(opening_pos * days_since_old_pos * (annual_interest_rate / 360), 2) # calculate interest for above days principal_component = round(max(emi_amount - interest_component,0), 2) interest_paid = min(remaining_amount, interest_component) remaining_amount -= interest_paid principal_paid = min(remaining_amount, principal_component) remaining_amount -= principal_paid closing_pos = opening_pos - principal_component else: # Normal EMI calculation if previous_emipayment_id == 0: # if previous row was of prepartpayment days_after_new_pos = (emi_date - previous_emi_date).days # calculate days_diff betn prev emi date(prepartpayment date) and current emi date interest_component = round(opening_pos * days_after_new_pos * (annual_interest_rate / 360), 2) # calculate interest fro above days else: interest_component = round(emi_amount if (beginning_balance * irr_rate) > emi_amount else (beginning_balance * irr_rate), 2) # normal interest calculation principal_component = round(max(emi_amount - interest_component, 0), 2) pre_emi_paid = min(total_paid_for_bpi, pre_emi) # allocate total_paid_for_bpi to pre_emi morat_paid = min(remaining_amount, moratorium_amount) remaining_amount -= morat_paid interest_paid = min(remaining_amount, interest_component) remaining_amount -= interest_paid principal_paid = min(remaining_amount, principal_component) remaining_amount -= principal_paid bounce_paid = min(remaining_amount, bounce) remaining_amount -= bounce_paid penal_paid = min(remaining_amount, penal) remaining_amount -= penal_paid closing_pos = opening_pos - principal_component # assign closing balance beginning_balance = closing_pos # update opening balance for next row # Update previous_emi_date & previous emi_payment_id to current values for the next iteration previous_emi_date = emi_date previous_emipayment_id = emi_payment_id results.append({ # Record the results for the row 'emi_payment_id': row.emi_payment_id, 'application_id': row.application_id, 'emi_date': row.emi_date if emi_payment_id !=0 else pd.NaT, # dont show date if row is of Pre part payment 'due_date': row.due_date if emi_payment_id !=0 else pd.NaT, # dont show date if row is of Pre part payment 'pre_emi_amount': pre_emi, 'interest_due': round(interest_component, 2) if emi_payment_id != 0 else 0, # dont show due if row is of Pre part payment 'principal_due': round(principal_component, 2) if emi_payment_id != 0 else 0, # dont show due if row is of Pre part payment 'emi_amount': emi_amount if emi_payment_id != 0 else 0, # dont show amount if row is of Pre part payment 'bounce_charges': bounce, 'delay_penalty': penal, 'charges_waived': row.charges_waived, 'total_amount_due': (pre_emi + emi_amount + bounce + penal + moratorium_amount), 'total_amount_paid': round(total_paid + total_paid_for_bpi, 2), 'pre_emi_paid': round(pre_emi_paid, 2), 'interest_paid': round(interest_paid, 2), 'principal_paid': round(principal_paid, 2), 'bounce_paid': round(bounce_paid, 2), 'penal_paid': round(penal_paid, 2), 'charges_to_be_waived': (bounce - round(bounce_paid, 2)) + (penal - round(penal_paid, 2)), 'payment_details': row.payment_details, 'unpaid_emi_date': row.emi_date if interest_paid + principal_paid < emi_amount else pd.NaT, 'opening_pos': round(opening_pos, 2), 'closing_pos': round(closing_pos, 2), 'latest_payment_date': row.latest_payment_date }) return pd.DataFrame(results) # --- Financial Math Utilities --- def days360(start_date, end_date): """Calculates the days between two dates based on the 360-day year (NASD method).""" start_day = start_date.day start_month = start_date.month start_year = start_date.year end_day = end_date.day end_month = end_date.month end_year = end_date.year if start_day == 31: start_day = 30 if end_day == 31 and start_day == 30: end_day = 30 return 360 * (end_year - start_year) + 30 * (end_month - start_month) + (end_day - start_day) def apply_rounding(value: float, method: str, digits: int) -> float: """Rounds a value up, down, or naturally according to given decimal digits.""" if value is None: return None multiplier = 10 ** digits if method == "ROUND_UP": res = math.ceil(value * multiplier) / multiplier elif method == "ROUND_DOWN": res = math.floor(value * multiplier) / multiplier else: res = math.floor(value * multiplier + 0.5) / multiplier return int(res) if digits <= 0 else res def calculate_emi(loan_amount: float, annual_rate: float, tenure_months: int, rounding_config: dict) -> float: """Calculates standard EMI using PMT formula.""" if annual_rate == 0: return apply_rounding(loan_amount / tenure_months, rounding_config["emi_rounding"], rounding_config["emi_rounding_digits"]) monthly_rate = annual_rate / 12 emi = loan_amount * monthly_rate * ((1 + monthly_rate) ** tenure_months) / (((1 + monthly_rate) ** tenure_months) - 1) return apply_rounding(emi, rounding_config["emi_rounding"], rounding_config["emi_rounding_digits"]) def calculate_bpi_raw(principal: float, rate: float, disb_dt, bpi_due_dt, bpi_day_convention: str, tenure: int = 0) -> float: """Calculates raw Broken Period Interest based on LMS day conventions.""" bpi_days = (bpi_due_dt - disb_dt).days if bpi_day_convention in ["30_360", "DAYS360"]: calc_days = days360(disb_dt, bpi_due_dt) return principal * rate * (calc_days / 360.0) elif bpi_day_convention == "ACTUAL_ACTUAL": denominator = 366.0 if calendar.isleap(disb_dt.year) else 365.0 return principal * rate * (bpi_days / denominator) elif bpi_day_convention == "ACTUAL_360": return principal * rate * (bpi_days / 360.0) elif bpi_day_convention == "IPMT": return float(npf.ipmt(rate=rate/12, per=1, nper=tenure, pv=-principal)) else: # ACTUAL_365 return principal * rate * (bpi_days / 365.0) def calculate_interest_raw(principal: float, rate: float, start_date, end_date, calculation_basis: str, period: int = 1, tenure: int = 0, original_loan_amount: float = 0.0) -> float: """Calculates raw interest for standard EMI periods based on LMS day conventions.""" days = (end_date - start_date).days if calculation_basis in ["30_360", "DAYS360"]: calc_days = days360(start_date, end_date) return principal * rate * (calc_days / 360.0) elif calculation_basis == "ACTUAL_ACTUAL": denominator = 366.0 if calendar.isleap(start_date.year) else 365.0 return principal * rate * (days / denominator) elif calculation_basis == "ACTUAL_360": return principal * rate * (days / 360.0) elif calculation_basis == "IPMT": # IPMT uses the original loan amount and period number instead of declining balance return float(npf.ipmt(rate=rate/12, per=period, nper=tenure, pv=-original_loan_amount)) else: # ACTUAL_365 return principal * rate * (days / 365.0)