Chinenye Nmoh
390 lines
14 KiB
Python
390 lines
14 KiB
Python
from datetime import datetime, timezone
|
|
from sqlalchemy.orm import relationship
|
|
from dateutil.relativedelta import relativedelta
|
|
from datetime import timedelta
|
|
import logging
|
|
from sqlalchemy import and_, or_, not_
|
|
from sqlalchemy.sql import func
|
|
from app.utils.logger import logger
|
|
from app.extensions import db
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from datetime import datetime, timezone
|
|
|
|
class Loan(db.Model):
|
|
__tablename__ = "loans"
|
|
|
|
id = db.Column(
|
|
db.Integer,
|
|
primary_key=True,
|
|
autoincrement=True,
|
|
)
|
|
customer_id = db.Column(db.String(50), nullable=False)
|
|
transaction_id = db.Column(db.String(50), nullable=True)
|
|
original_transaction = db.Column(db.String(50), nullable=True)
|
|
account_id = db.Column(db.String(50), nullable=False)
|
|
offer_id = db.Column(db.String(20), nullable=False)
|
|
product_id = db.Column(db.String(20), nullable=True)
|
|
collection_type = db.Column(db.String(20), nullable=True)
|
|
current_loan_amount = db.Column(db.Float, nullable=True)
|
|
initial_loan_amount = db.Column(db.Float, nullable=False)
|
|
default_penalty_fee = db.Column(db.Float, default=0)
|
|
continuous_fee = db.Column(db.Float, default=0)
|
|
upfront_fee = db.Column(db.Float, nullable=True, default=0.0)
|
|
repayment_amount = db.Column(db.Float, nullable=True, default=0.0)
|
|
balance = db.Column(db.Float, nullable=True, default=0.0)
|
|
installment_amount = db.Column(db.Float, nullable=True, default=0.0)
|
|
status = db.Column(db.String(20), default='pending')
|
|
tenor = db.Column(db.Integer, nullable=True)
|
|
due_date = db.Column(db.DateTime, nullable=True)
|
|
created_at = db.Column(db.DateTime(timezone=True), server_default=func.now())
|
|
updated_at = db.Column(db.DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
|
|
eligible_amount = db.Column(db.Float, nullable=True, default=0.0)
|
|
disburse_date = db.Column(db.DateTime, nullable=True)
|
|
disburse_verify = db.Column(db.DateTime, nullable=True)
|
|
disburse_result = db.Column(db.String(10), nullable=True)
|
|
disburse_description = db.Column(db.String(100), nullable=True)
|
|
verify_result = db.Column(db.String(10), nullable=True)
|
|
verify_description = db.Column(db.String(100), nullable=True)
|
|
reference = db.Column(db.String(50), nullable=True)
|
|
|
|
customer = relationship(
|
|
"Customer",
|
|
primaryjoin="Customer.id == Loan.customer_id",
|
|
foreign_keys=[customer_id],
|
|
back_populates="loans",
|
|
)
|
|
|
|
loan_charges = relationship(
|
|
"LoanCharge",
|
|
primaryjoin="Loan.id == LoanCharge.loan_id",
|
|
foreign_keys="LoanCharge.loan_id",
|
|
back_populates="loan",
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<Loan {self.id}>"
|
|
|
|
def to_dict(self):
|
|
"""
|
|
Convert the Loan object to a dictionary format for JSON serialization.
|
|
"""
|
|
return {
|
|
'debtId': self.id,
|
|
"customerId": self.customer_id,
|
|
'initialLoanAmount': self.initial_loan_amount,
|
|
'currentLoanAmount': self.current_loan_amount,
|
|
'defaultPenaltyFee': self.default_penalty_fee,
|
|
'continuousFee': self.continuous_fee,
|
|
'collectionType': self.collection_type,
|
|
'repaymentAmount':self.repayment_amount,
|
|
'status': self.status,
|
|
'productId': self.product_id,
|
|
'disburseResult': self.disburse_result,
|
|
'disburseDescription': self.disburse_description,
|
|
'verifyResult': self.verify_result,
|
|
'verifyDescription': self.verify_description,
|
|
'transactionId': self.transaction_id,
|
|
'accountId':self.account_id,
|
|
'dueDate': self.due_date.isoformat() if self.due_date else None,
|
|
'loanDate': self.created_at.isoformat() if self.created_at else None,
|
|
'disburseDate': self.disburse_date.isoformat() if self.disburse_date else None,
|
|
'disburseVerify': self.disburse_verify.isoformat() if self.disburse_verify else None,
|
|
'reference': self.reference,
|
|
'balance': self.balance,
|
|
'tenor': self.tenor,
|
|
}
|
|
|
|
@classmethod
|
|
def get_loan_by_transaction_id(cls, transaction_id):
|
|
return cls.query.filter_by(transaction_id=transaction_id).first()
|
|
@classmethod
|
|
def get_loan_by_loan_id(cls, loan_id):
|
|
return cls.query.filter_by(id=loan_id).first()
|
|
|
|
@classmethod
|
|
def set_disbursement_date(cls, loan_id, customer_id):
|
|
"""
|
|
Update the disburse date of the loan with the given loan_id.
|
|
"""
|
|
# Retrieve loan
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
# Check if customer_id matches
|
|
if loan.customer_id != customer_id:
|
|
raise ValueError(f"Customer ID {customer_id} does not match the loan's customer ID.")
|
|
|
|
current_time = datetime.now()
|
|
logger.info(f"What is now ======= ==== ==> : {current_time}")
|
|
# Update loan disburse_date
|
|
loan.disburse_date = current_time
|
|
|
|
# Commit changes to database
|
|
try:
|
|
logger.info(f"Updating disburse date for loan ID {loan_id} to {current_time}")
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Failed to update disburse date: {e}")
|
|
raise
|
|
@classmethod
|
|
def set_disburse_verify_date(cls, loan_id, customer_id):
|
|
"""
|
|
Update the disburse verify date of the loan with the given loan_id.
|
|
"""
|
|
# Retrieve loan
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
# Check if customer_id matches
|
|
if loan.customer_id != customer_id:
|
|
raise ValueError(f"Customer ID {customer_id} does not match the loan's customer ID.")
|
|
|
|
current_time = datetime.now()
|
|
logger.info(f"What is now ======= ==== ==> : {current_time}")
|
|
# Update loan verify_date
|
|
loan.disburse_verify = current_time
|
|
|
|
# Commit changes to database
|
|
try:
|
|
logger.info(f"Updating disburse verify date for loan ID {loan_id} to {current_time}")
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Failed to update disburse verify date: {e}")
|
|
raise
|
|
|
|
@classmethod
|
|
def set_disbursement_message(cls, loan_id, description):
|
|
"""
|
|
Update the disburse result and description of the loan with the given loan_id.
|
|
"""
|
|
# Retrieve loan
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
# Update disburse description only
|
|
loan.disburse_description = description
|
|
|
|
# Commit changes to database
|
|
try:
|
|
logger.info(f"Updating disburse result for loan ID {loan_id} with description {description}")
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Failed to update disbursement result: {e}")
|
|
raise
|
|
|
|
|
|
@classmethod
|
|
def set_disbursement_result(cls, loan_id, result, description):
|
|
"""
|
|
Update the disburse result and description of the loan with the given loan_id.
|
|
"""
|
|
# Retrieve loan
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
# Update disburse result and description
|
|
loan.disburse_result = result
|
|
loan.disburse_description = description
|
|
|
|
# Commit changes to database
|
|
try:
|
|
logger.info(f"Updating disburse result for loan ID {loan_id} to {result} with description {description}")
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Failed to update disbursement result: {e}")
|
|
raise
|
|
|
|
|
|
@classmethod
|
|
def set_disburse_verify_result(cls, loan_id, result, description):
|
|
"""
|
|
Update the verify result and description of the loan with the given loan_id.
|
|
"""
|
|
# Retrieve loan
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
# Update disburse result and description
|
|
loan.verify_result = result
|
|
loan.verify_description = description
|
|
|
|
# Commit changes to database
|
|
try:
|
|
logger.info(f"Updating verify result for loan ID {loan_id} to {result} with description {description}")
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Failed to update verify result: {e}")
|
|
raise
|
|
@classmethod
|
|
def get_latest_loan_without_disburse_date(cls):
|
|
"""
|
|
Get the latest loan without a disbursement date.
|
|
"""
|
|
return cls.query.filter(
|
|
cls.disburse_date.is_(None)
|
|
).order_by(cls.created_at.desc()).first()
|
|
|
|
@classmethod
|
|
def get_latest_loan_with_disburse_date(cls):
|
|
"""
|
|
Get the latest loan with a disbursement date and no verification date.
|
|
"""
|
|
return cls.query.filter(
|
|
cls.disburse_date.isnot(None),
|
|
cls.disburse_verify.is_(None)
|
|
).order_by(cls.created_at.desc()).first()
|
|
|
|
@classmethod
|
|
def get_customer_loans(cls, customer_id):
|
|
"""
|
|
Get customer's active loans and sum by customer_id.
|
|
"""
|
|
customer_loans = cls.query.filter_by( customer_id = customer_id).all()
|
|
if not customer_loans:
|
|
raise ValueError(f"Customer with Id {customer_id} does not have any loan.")
|
|
|
|
total_amount = (
|
|
cls.query.with_entities(func.coalesce(func.sum(cls.balance), 0.0))
|
|
.filter_by(customer_id=customer_id)
|
|
.scalar()
|
|
)
|
|
|
|
logger.info(f"Found {len(customer_loans)} loans for customer ID: {customer_id} with total amount: {total_amount}")
|
|
|
|
return customer_loans, total_amount
|
|
@classmethod
|
|
def get_customer_active_loans(cls, customer_id):
|
|
"""
|
|
Get customer's active loans and sum by customer_id.
|
|
"""
|
|
customer_loans = cls.query.filter(
|
|
cls.customer_id == customer_id,
|
|
cls.status != 'repaid'
|
|
).all()
|
|
|
|
if not customer_loans:
|
|
raise ValueError(f"Customer with Id {customer_id} does not have any active loan.")
|
|
|
|
total_amount = (
|
|
cls.query
|
|
.with_entities(func.coalesce(func.sum(cls.balance), 0.0))
|
|
.filter(
|
|
cls.customer_id == customer_id,
|
|
cls.status != 'repaid'
|
|
)
|
|
.scalar()
|
|
)
|
|
|
|
logger.info(f"Found {len(customer_loans)} active loans for customer ID: {customer_id} with total amount: {total_amount}")
|
|
|
|
return customer_loans, total_amount
|
|
|
|
|
|
@classmethod
|
|
def update_status(cls, loan_id, status):
|
|
"""
|
|
Update the status of the loan record with the given loan_id.
|
|
"""
|
|
try:
|
|
# Retrieve loan record
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
if loan.status == status:
|
|
return loan.to_dict() # Still return the current state if no change
|
|
|
|
# Update status and timestamp
|
|
loan.status = status
|
|
loan.updated_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
|
|
logger.info("Loan status updated and committed.")
|
|
return loan.to_dict()
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Error updating loan status: {e}")
|
|
raise Exception(f"Error updating loan status: {str(e)}")
|
|
|
|
|
|
@classmethod
|
|
def update_loan_balance(cls, loan_id, amount_collected):
|
|
"""
|
|
Update the balance of a loan after successful repayment.
|
|
"""
|
|
try:
|
|
# Fetch the loan record
|
|
loan = cls.query.get(loan_id)
|
|
|
|
if not loan:
|
|
raise ValueError(f"Loan with ID {loan_id} does not exist.")
|
|
|
|
# Convert to Decimal and round to 2 decimal places
|
|
amount_collected = Decimal(str(amount_collected)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
balance = Decimal(str(loan.balance or 0)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
|
|
# Ensure valid repayment amount
|
|
# if amount_collected <= Decimal("0.00"):
|
|
# raise ValueError("Repayment amount must be greater than zero.")
|
|
if balance <= Decimal("0.00"):
|
|
raise ValueError("There is no balance for this loan.")
|
|
if amount_collected > balance:
|
|
# allow tiny rounding diff
|
|
if abs(amount_collected - balance) <= Decimal("0.01"):
|
|
amount_collected = balance
|
|
else:
|
|
raise ValueError("Repayment amount exceeds current loan balance.")
|
|
|
|
|
|
# Deduct the amount from the current balance
|
|
new_balance = balance - amount_collected
|
|
loan.balance = float(new_balance)
|
|
loan.updated_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
|
|
logger.info(f"Loan balance updated for loan ID {loan_id}. New balance: {loan.balance}")
|
|
return loan.to_dict()
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.error(f"Error updating loan balance: {e}")
|
|
raise Exception(f"Error updating loan balance: {str(e)}")
|
|
|
|
@classmethod
|
|
def get_overdue_loans(cls):
|
|
"""
|
|
Get all overdue loans.
|
|
"""
|
|
try:
|
|
overdue_loans = cls.query.filter(
|
|
cls.due_date < datetime.now(timezone.utc),
|
|
cls.status != 'repaid'
|
|
).all()
|
|
|
|
if not overdue_loans:
|
|
logger.info("No overdue loans found.")
|
|
return []
|
|
|
|
logger.info(f"Found {len(overdue_loans)} overdue loans.")
|
|
return overdue_loans
|
|
except Exception as e:
|
|
logger.error(f"Error fetching overdue loans: {e}")
|
|
return []
|