""" Comprehensive Reports and Statements System for Branch Microfinance Complete report generator with logging, statements, and document generation. """ import io import csv import json import xlsxwriter from datetime import datetime, timedelta from decimal import Decimal import matplotlib matplotlib.use('Agg') # Set backend before importing pyplot import matplotlib.pyplot as plt from reportlab.lib import colors from reportlab.lib.pagesizes import letter, A4 from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, PageBreak, Image from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import inch, mm from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_RIGHT from reportlab.graphics.shapes import Drawing from reportlab.graphics.charts.linecharts import HorizontalLineChart from reportlab.graphics.charts.piecharts import Pie from django.db.models import Sum, Count, Avg, Q, F from django.utils import timezone from django.core.cache import cache from django.db import connection from loans.models import Loan, LoanApplication, Repayment from users.models import CustomUser from utils.models import AuditLog import logging logger = logging.getLogger(__name__) # Enhanced logging for all report generation activities def log_report_generation(user, report_type, parameters=None, success=True, error_message=None): """Log all report generation activities with comprehensive details""" try: description = f"Generated {report_type} report" if parameters: description += f" with parameters: {json.dumps(parameters, default=str)}" if not success and error_message: description += f" - ERROR: {error_message}" AuditLog.objects.create( user=user, action='generate_report', model_name='Report', object_id=report_type, description=description, ip_address=getattr(user, 'current_ip', None) ) logger.info(f"Report generation logged: {description}") except Exception as e: logger.error(f"Failed to log report generation: {e}") # Optimized matplotlib settings plt.style.use('default') plt.rcParams.update({ 'figure.figsize': (10, 6), 'figure.dpi': 100, 'savefig.dpi': 150, 'font.size': 10, 'axes.titlesize': 12, 'axes.labelsize': 10, 'xtick.labelsize': 9, 'ytick.labelsize': 9, 'legend.fontsize': 9 }) class OptimizedReportGenerator: """ High-performance report generator focused on essential metrics and fast execution. Designed for minimal system impact with maximum business value. """ # Cache timeouts (in seconds) CACHE_TIMEOUT_SHORT = 300 # 5 minutes for frequently changing data CACHE_TIMEOUT_MEDIUM = 1800 # 30 minutes for moderate data CACHE_TIMEOUT_LONG = 3600 # 1 hour for stable data @staticmethod def _get_cache_key(report_type, params=None): """Generate optimized cache key""" if params: param_str = json.dumps(params, sort_keys=True, default=str) return f"report_{report_type}_{hash(param_str)}" return f"report_{report_type}" @staticmethod def _get_cached_data(cache_key): """Get cached data with fallback""" try: return cache.get(cache_key) except Exception as e: logger.warning(f"Cache retrieval failed: {e}") return None @staticmethod def _cache_data(cache_key, data, timeout=None): """Cache data with error handling""" if timeout is None: timeout = OptimizedReportGenerator.CACHE_TIMEOUT_MEDIUM try: cache.set(cache_key, data, timeout) except Exception as e: logger.warning(f"Cache storage failed: {e}") @staticmethod def _execute_raw_query(query, params=None): """Execute optimized raw SQL query""" try: with connection.cursor() as cursor: cursor.execute(query, params or []) columns = [col[0] for col in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()] except Exception as e: logger.error(f"Raw query failed: {e}") return [] @staticmethod def generate_executive_summary(date_range_days=30, branch_id=None): """ Generate executive summary with key performance indicators. Optimized for speed and essential business metrics only. """ cache_key = OptimizedReportGenerator._get_cache_key('executive_summary', {'days': date_range_days, 'branch_id': branch_id}) cached_data = OptimizedReportGenerator._get_cached_data(cache_key) if cached_data: return cached_data try: end_date = timezone.now().date() start_date = end_date - timedelta(days=date_range_days) # Single optimized query for core metrics query = """ SELECT COUNT(*) as total_loans, COUNT(CASE WHEN status = 'active' THEN 1 END) as active_loans, COUNT(CASE WHEN status = 'completed' THEN 1 END) as completed_loans, COUNT(CASE WHEN status = 'defaulted' THEN 1 END) as defaulted_loans, COALESCE(SUM(principal_amount), 0) as total_disbursed, COALESCE(SUM(amount_paid), 0) as total_collected, COALESCE(AVG(principal_amount), 0) as avg_loan_size, COUNT(CASE WHEN due_date < %s AND status = 'active' THEN 1 END) as overdue_loans FROM loans_loan WHERE created_at >= %s AND created_at <= %s """ params = [timezone.now().date(), start_date, end_date] # Add branch filtering if specified if branch_id: query += " AND borrower_id IN (SELECT id FROM users_customuser WHERE branch_id = %s)" params.append(branch_id) results = OptimizedReportGenerator._execute_raw_query(query, params) if not results: return {'error': 'No data available'} data = results[0] # Calculate derived metrics portfolio_at_risk = (data['overdue_loans'] / max(data['active_loans'], 1)) * 100 collection_rate = (data['total_collected'] / max(data['total_disbursed'], 1)) * 100 default_rate = (data['defaulted_loans'] / max(data['total_loans'], 1)) * 100 summary = { 'period': f"{start_date} to {end_date}", 'total_loans': data['total_loans'], 'active_loans': data['active_loans'], 'completed_loans': data['completed_loans'], 'defaulted_loans': data['defaulted_loans'], 'overdue_loans': data['overdue_loans'], 'total_disbursed': float(data['total_disbursed']), 'total_collected': float(data['total_collected']), 'outstanding_amount': float(data['total_disbursed']) - float(data['total_collected']), 'avg_loan_size': float(data['avg_loan_size']), 'portfolio_at_risk': round(portfolio_at_risk, 2), 'collection_rate': round(collection_rate, 2), 'default_rate': round(default_rate, 2), 'generated_at': timezone.now().isoformat() } OptimizedReportGenerator._cache_data(cache_key, summary, OptimizedReportGenerator.CACHE_TIMEOUT_SHORT) return summary except Exception as e: logger.error(f"Executive summary generation failed: {e}") return {'error': str(e)} @staticmethod def generate_loan_performance_trends(months=6): """ Generate loan performance trends for the specified number of months. Focused on essential trend data for decision making. """ cache_key = OptimizedReportGenerator._get_cache_key('performance_trends', {'months': months}) cached_data = OptimizedReportGenerator._get_cached_data(cache_key) if cached_data: return cached_data try: # Optimized query for monthly trends query = """ SELECT DATE_FORMAT(created_at, '%%Y-%%m') as month, COUNT(*) as loan_count, COALESCE(SUM(principal_amount), 0) as disbursed_amount, COUNT(CASE WHEN status = 'completed' THEN 1 END) as completed_count, COUNT(CASE WHEN status = 'defaulted' THEN 1 END) as defaulted_count FROM loans_loan WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL %s MONTH) GROUP BY DATE_FORMAT(created_at, '%%Y-%%m') ORDER BY month DESC LIMIT %s """ results = OptimizedReportGenerator._execute_raw_query(query, [months, months]) trends = { 'months': [], 'loan_counts': [], 'disbursed_amounts': [], 'completion_rates': [], 'default_rates': [] } for row in reversed(results): # Reverse to get chronological order trends['months'].append(row['month']) trends['loan_counts'].append(row['loan_count']) trends['disbursed_amounts'].append(float(row['disbursed_amount'])) # Calculate rates completion_rate = (row['completed_count'] / max(row['loan_count'], 1)) * 100 default_rate = (row['defaulted_count'] / max(row['loan_count'], 1)) * 100 trends['completion_rates'].append(round(completion_rate, 2)) trends['default_rates'].append(round(default_rate, 2)) OptimizedReportGenerator._cache_data(cache_key, trends, OptimizedReportGenerator.CACHE_TIMEOUT_MEDIUM) return trends except Exception as e: logger.error(f"Performance trends generation failed: {e}") return {'error': str(e)} @staticmethod def generate_risk_analysis(): """ Generate focused risk analysis with actionable insights. Identifies high-risk loans and clients requiring attention. """ cache_key = OptimizedReportGenerator._get_cache_key('risk_analysis') cached_data = OptimizedReportGenerator._get_cached_data(cache_key) if cached_data: return cached_data try: # High-risk loans query high_risk_query = """ SELECT l.id, l.loan_number, l.principal_amount, l.due_date, DATEDIFF(CURDATE(), l.due_date) as days_overdue, u.first_name, u.last_name, u.phone_number FROM loans_loan l JOIN users_customuser u ON l.borrower_id = u.id WHERE l.status = 'active' AND l.due_date < CURDATE() ORDER BY days_overdue DESC LIMIT 20 """ high_risk_loans = OptimizedReportGenerator._execute_raw_query(high_risk_query) # Risk summary query risk_summary_query = """ SELECT COUNT(CASE WHEN DATEDIFF(CURDATE(), due_date) BETWEEN 1 AND 7 THEN 1 END) as overdue_1_7_days, COUNT(CASE WHEN DATEDIFF(CURDATE(), due_date) BETWEEN 8 AND 30 THEN 1 END) as overdue_8_30_days, COUNT(CASE WHEN DATEDIFF(CURDATE(), due_date) > 30 THEN 1 END) as overdue_30_plus_days, COALESCE(SUM(CASE WHEN due_date < CURDATE() THEN principal_amount - amount_paid END), 0) as total_at_risk FROM loans_loan WHERE status = 'active' AND due_date < CURDATE() """ risk_summary = OptimizedReportGenerator._execute_raw_query(risk_summary_query) analysis = { 'high_risk_loans': high_risk_loans, 'risk_summary': risk_summary[0] if risk_summary else {}, 'generated_at': timezone.now().isoformat() } OptimizedReportGenerator._cache_data(cache_key, analysis, OptimizedReportGenerator.CACHE_TIMEOUT_SHORT) return analysis except Exception as e: logger.error(f"Risk analysis generation failed: {e}") return {'error': str(e)} @staticmethod def generate_executive_summary_pdf(): """Generate optimized executive summary PDF""" try: buffer = io.BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4, topMargin=20*mm, bottomMargin=20*mm) styles = getSampleStyleSheet() # Custom styles for professional appearance title_style = ParagraphStyle( 'CustomTitle', parent=styles['Title'], fontSize=18, textColor=colors.HexColor('#2563eb'), alignment=TA_CENTER, spaceAfter=20 ) elements = [] # Get data summary_data = OptimizedReportGenerator.generate_executive_summary() if 'error' in summary_data: elements.append(Paragraph("Error generating report", styles['Normal'])) doc.build(elements) buffer.seek(0) return buffer # Title and date elements.append(Paragraph("Executive Summary Report", title_style)) elements.append(Paragraph(f"Generated: {timezone.now().strftime('%B %d, %Y')}", styles['Normal'])) elements.append(Spacer(1, 20)) # Key Metrics Table kpi_data = [ ["Metric", "Value"], ["Total Loans", f"{summary_data['total_loans']:,}"], ["Active Loans", f"{summary_data['active_loans']:,}"], ["Total Disbursed", f"KES {summary_data['total_disbursed']:,.2f}"], ["Collection Rate", f"{summary_data['collection_rate']:.1f}%"], ["Default Rate", f"{summary_data['default_rate']:.1f}%"] ] kpi_table = Table(kpi_data, colWidths=[60*mm, 60*mm]) kpi_table.setStyle(TableStyle([ ('BACKGROUND', (0, 0), (-1, 0), colors.grey), ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('GRID', (0, 0), (-1, -1), 1, colors.black) ])) elements.append(kpi_table) doc.build(elements) buffer.seek(0) return buffer except Exception as e: logger.error(f"PDF generation failed: {e}") buffer = io.BytesIO() return buffer @staticmethod def generate_portfolio_analysis_csv(date_range_days=30): """Generate optimized portfolio analysis CSV""" try: output = io.StringIO() writer = csv.writer(output) # Header writer.writerow(['Portfolio Analysis Report']) writer.writerow(['Generated:', timezone.now().strftime('%Y-%m-%d %H:%M:%S')]) writer.writerow([]) # Executive Summary summary_data = OptimizedReportGenerator.generate_executive_summary(date_range_days) if 'error' not in summary_data: writer.writerow(['Executive Summary']) writer.writerow(['Metric', 'Value']) writer.writerow(['Total Loans', summary_data['total_loans']]) writer.writerow(['Active Loans', summary_data['active_loans']]) writer.writerow(['Total Disbursed (KES)', f"{summary_data['total_disbursed']:,.2f}"]) writer.writerow(['Collection Rate (%)', f"{summary_data['collection_rate']:.2f}"]) writer.writerow(['Default Rate (%)', f"{summary_data['default_rate']:.2f}"]) return output.getvalue() except Exception as e: logger.error(f"CSV generation failed: {e}") output = io.StringIO() writer = csv.writer(output) writer.writerow(['Error generating report:', str(e)]) return output.getvalue() @staticmethod def generate_dashboard_data(): """Generate optimized dashboard data for web interface""" cache_key = OptimizedReportGenerator._get_cache_key('dashboard_data') cached_data = OptimizedReportGenerator._get_cached_data(cache_key) if cached_data: return cached_data try: # Get core metrics summary = OptimizedReportGenerator.generate_executive_summary(30) trends = OptimizedReportGenerator.generate_loan_performance_trends(6) risk = OptimizedReportGenerator.generate_risk_analysis() # Prepare dashboard data dashboard_data = { 'kpis': { 'total_loans': summary.get('total_loans', 0), 'active_loans': summary.get('active_loans', 0), 'total_disbursed': summary.get('total_disbursed', 0), 'collection_rate': summary.get('collection_rate', 0), 'default_rate': summary.get('default_rate', 0) }, 'trends': { 'months': trends.get('months', []), 'loan_counts': trends.get('loan_counts', []) }, 'risk_alerts': { 'overdue_loans': summary.get('overdue_loans', 0), 'high_risk_count': len(risk.get('high_risk_loans', [])) }, 'generated_at': timezone.now().isoformat() } OptimizedReportGenerator._cache_data(cache_key, dashboard_data, OptimizedReportGenerator.CACHE_TIMEOUT_SHORT) return dashboard_data except Exception as e: logger.error(f"Dashboard data generation failed: {e}") return {'error': str(e)} @staticmethod def generate_loan_statement(loan_id, user=None): """Generate comprehensive loan statement""" try: loan = Loan.objects.select_related('borrower', 'application__loan_product').get(id=loan_id) repayments = Repayment.objects.filter(loan=loan).order_by('payment_date') # Log the statement generation if user: log_report_generation(user, 'loan_statement', {'loan_id': loan_id}) # Calculate current amounts using stored values current_interest = loan.interest_amount current_processing_fee = loan.processing_fee current_total = loan.total_amount current_outstanding = current_total - loan.amount_paid statement_data = { 'loan': { 'loan_number': loan.loan_number, 'borrower_name': loan.borrower.get_full_name(), 'borrower_phone': loan.borrower.phone_number, 'principal_amount': float(loan.principal_amount), 'interest_rate': float(loan.get_display_interest_rate()), 'interest_amount': float(current_interest), 'processing_fee': float(current_processing_fee), 'total_amount': float(current_total), 'loan_term': loan.duration_days, 'disbursement_date': loan.disbursement_date.strftime('%Y-%m-%d') if loan.disbursement_date else None, 'due_date': loan.due_date.strftime('%Y-%m-%d') if loan.due_date else None, 'status': loan.status, 'amount_paid': float(loan.amount_paid), 'balance': float(current_outstanding) }, 'repayments': [ { 'date': rep.payment_date.strftime('%Y-%m-%d'), 'amount': float(rep.amount), 'method': rep.payment_method, 'reference': rep.receipt_number, 'balance_after': float(loan.principal_amount - (Repayment.objects.filter( loan=loan, payment_date__lte=rep.payment_date ).aggregate(total=Sum('amount'))['total'] or 0)) } for rep in repayments ], 'summary': { 'total_repayments': repayments.count(), 'total_paid': float(repayments.aggregate(total=Sum('amount'))['total'] or 0), 'outstanding_balance': float(current_outstanding), 'days_overdue': (timezone.now().date() - loan.due_date.date()).days if loan.due_date and loan.due_date.date() < timezone.now().date() else 0 }, 'generated_at': timezone.now().isoformat() } return statement_data except Exception as e: if user: log_report_generation(user, 'loan_statement', {'loan_id': loan_id}, False, str(e)) logger.error(f"Loan statement generation failed: {e}") return {'error': str(e)} @staticmethod def _get_client_credit_score(client): """Get client credit score from LoanScoring model""" try: from reports.models import LoanScoring credit_score = LoanScoring.objects.get(user=client) return credit_score.total_score except: return 0 @staticmethod def generate_client_statement(client_id, user=None): """Generate comprehensive client statement with all loans""" try: client = CustomUser.objects.get(id=client_id) loans = Loan.objects.filter(borrower=client).order_by('-created_at') # Log the statement generation if user: log_report_generation(user, 'client_statement', {'client_id': client_id}) statement_data = { 'client': { 'name': client.get_full_name(), 'phone': client.phone_number, 'email': client.email, 'id_number': client.id_number, 'address': client.physical_address, 'monthly_income': float(client.monthly_income) if client.monthly_income else 0, 'credit_score': OptimizedReportGenerator._get_client_credit_score(client), 'member_since': client.date_joined.strftime('%Y-%m-%d') }, 'loans': [ { 'loan_number': loan.loan_number, 'principal_amount': float(loan.principal_amount), 'disbursement_date': loan.disbursement_date.strftime('%Y-%m-%d') if loan.disbursement_date else None, 'due_date': loan.due_date.strftime('%Y-%m-%d') if loan.due_date else None, 'status': loan.status, 'amount_paid': float(loan.amount_paid), 'balance': float(loan.principal_amount - loan.amount_paid) } for loan in loans ], 'summary': { 'total_loans': loans.count(), 'active_loans': loans.filter(status='active').count(), 'completed_loans': loans.filter(status='completed').count(), 'total_borrowed': float(loans.aggregate(total=Sum('principal_amount'))['total'] or 0), 'total_repaid': float(loans.aggregate(total=Sum('amount_paid'))['total'] or 0), 'outstanding_balance': float(loans.filter(status='active').aggregate( total=Sum(F('principal_amount') - F('amount_paid')) )['total'] or 0) }, 'generated_at': timezone.now().isoformat() } return statement_data except Exception as e: if user: log_report_generation(user, 'client_statement', {'client_id': client_id}, False, str(e)) logger.error(f"Client statement generation failed: {e}") return {'error': str(e)} @staticmethod def generate_collection_statement(date_from=None, date_to=None, user=None): """Generate collection statement for a date range""" try: if not date_from: date_from = timezone.now().date() - timedelta(days=30) if not date_to: date_to = timezone.now().date() # Log the statement generation if user: log_report_generation(user, 'collection_statement', { 'date_from': str(date_from), 'date_to': str(date_to) }) repayments = Repayment.objects.filter( payment_date__range=[date_from, date_to] ).select_related('loan__borrower').order_by('-payment_date') statement_data = { 'period': { 'from': date_from.strftime('%Y-%m-%d'), 'to': date_to.strftime('%Y-%m-%d') }, 'collections': [ { 'date': rep.payment_date.strftime('%Y-%m-%d'), 'loan_number': rep.loan.loan_number, 'borrower_name': rep.loan.borrower.get_full_name(), 'amount': float(rep.amount), 'method': rep.payment_method, 'reference': rep.receipt_number } for rep in repayments ], 'summary': { 'total_collections': repayments.count(), 'total_amount': float(repayments.aggregate(total=Sum('amount'))['total'] or 0), 'by_method': { method: float(repayments.filter(payment_method=method).aggregate( total=Sum('amount') )['total'] or 0) for method in repayments.values_list('payment_method', flat=True).distinct() } }, 'generated_at': timezone.now().isoformat() } return statement_data except Exception as e: if user: log_report_generation(user, 'collection_statement', { 'date_from': str(date_from), 'date_to': str(date_to) }, False, str(e)) logger.error(f"Collection statement generation failed: {e}") return {'error': str(e)} @staticmethod def generate_loan_statement_pdf(loan_id, user=None): """Generate loan statement PDF""" try: statement_data = OptimizedReportGenerator.generate_loan_statement(loan_id, user) if 'error' in statement_data: return None buffer = io.BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4, topMargin=20*mm, bottomMargin=20*mm) styles = getSampleStyleSheet() # Custom styles title_style = ParagraphStyle( 'CustomTitle', parent=styles['Title'], fontSize=16, textColor=colors.HexColor('#2563eb'), alignment=TA_CENTER, spaceAfter=20 ) elements = [] loan = statement_data['loan'] # Header elements.append(Paragraph("LOAN STATEMENT", title_style)) elements.append(Paragraph(f"Generated: {timezone.now().strftime('%B %d, %Y at %I:%M %p')}", styles['Normal'])) elements.append(Spacer(1, 20)) # Loan Details loan_details = [ ["Loan Number:", loan['loan_number']], ["Borrower:", loan['borrower_name']], ["Phone:", loan['borrower_phone']], ["Principal Amount:", f"KES {loan['principal_amount']:,.2f}"], ["Interest Rate:", f"{loan['interest_rate']:.2f}%"], ["Loan Term:", f"{loan['loan_term']} days"], ["Disbursement Date:", loan['disbursement_date'] or 'N/A'], ["Due Date:", loan['due_date'] or 'N/A'], ["Status:", loan['status'].upper()], ["Amount Paid:", f"KES {loan['amount_paid']:,.2f}"], ["Outstanding Balance:", f"KES {loan['balance']:,.2f}"] ] loan_table = Table(loan_details, colWidths=[50*mm, 80*mm]) loan_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('BACKGROUND', (0, 0), (0, -1), colors.lightgrey) ])) elements.append(loan_table) elements.append(Spacer(1, 20)) # Repayment History if statement_data['repayments']: elements.append(Paragraph("REPAYMENT HISTORY", styles['Heading2'])) elements.append(Spacer(1, 10)) repayment_data = [["Date", "Amount", "Method", "Reference", "Balance After"]] for rep in statement_data['repayments']: repayment_data.append([ rep['date'], f"KES {rep['amount']:,.2f}", rep['method'], rep['reference'] or 'N/A', f"KES {rep['balance_after']:,.2f}" ]) repayment_table = Table(repayment_data, colWidths=[25*mm, 30*mm, 25*mm, 30*mm, 30*mm]) repayment_table.setStyle(TableStyle([ ('BACKGROUND', (0, 0), (-1, 0), colors.grey), ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), ('ALIGN', (0, 0), (-1, -1), 'CENTER'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 1, colors.black) ])) elements.append(repayment_table) doc.build(elements) buffer.seek(0) return buffer except Exception as e: if user: log_report_generation(user, 'loan_statement_pdf', {'loan_id': loan_id}, False, str(e)) logger.error(f"Loan statement PDF generation failed: {e}") return None @staticmethod def generate_client_statement_pdf(client_id, user=None): """Generate client statement PDF""" try: statement_data = OptimizedReportGenerator.generate_client_statement(client_id, user) if 'error' in statement_data: return None buffer = io.BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4, topMargin=20*mm, bottomMargin=20*mm) styles = getSampleStyleSheet() # Custom styles title_style = ParagraphStyle( 'CustomTitle', parent=styles['Title'], fontSize=16, textColor=colors.HexColor('#2563eb'), alignment=TA_CENTER, spaceAfter=20 ) elements = [] client = statement_data['client'] # Header elements.append(Paragraph("CLIENT STATEMENT", title_style)) elements.append(Paragraph(f"Generated: {timezone.now().strftime('%B %d, %Y at %I:%M %p')}", styles['Normal'])) elements.append(Spacer(1, 20)) # Client Details client_details = [ ["Client Name:", client['name']], ["Phone:", client['phone']], ["Email:", client['email'] or 'N/A'], ["ID Number:", client['id_number'] or 'N/A'], ["Address:", client['address'] or 'N/A'], ["Monthly Income:", f"KES {client['monthly_income']:,.2f}"], ["Credit Score:", str(client['credit_score']) if client['credit_score'] else 'N/A'], ["Member Since:", client['member_since']] ] client_table = Table(client_details, colWidths=[50*mm, 80*mm]) client_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('BACKGROUND', (0, 0), (0, -1), colors.lightgrey) ])) elements.append(client_table) elements.append(Spacer(1, 20)) # Loan Summary summary = statement_data['summary'] summary_data = [ ["Total Loans:", str(summary['total_loans'])], ["Active Loans:", str(summary['active_loans'])], ["Completed Loans:", str(summary['completed_loans'])], ["Total Borrowed:", f"KES {summary['total_borrowed']:,.2f}"], ["Total Repaid:", f"KES {summary['total_repaid']:,.2f}"], ["Outstanding Balance:", f"KES {summary['outstanding_balance']:,.2f}"] ] summary_table = Table(summary_data, colWidths=[50*mm, 80*mm]) summary_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('BACKGROUND', (0, 0), (0, -1), colors.lightblue) ])) elements.append(Paragraph("LOAN SUMMARY", styles['Heading2'])) elements.append(Spacer(1, 10)) elements.append(summary_table) elements.append(Spacer(1, 20)) # Loan History if statement_data['loans']: elements.append(Paragraph("LOAN HISTORY", styles['Heading2'])) elements.append(Spacer(1, 10)) loan_data = [["Loan Number", "Principal", "Disbursed", "Due Date", "Status", "Balance"]] for loan in statement_data['loans']: loan_data.append([ loan['loan_number'], f"KES {loan['principal_amount']:,.2f}", loan['disbursement_date'] or 'N/A', loan['due_date'] or 'N/A', loan['status'].upper(), f"KES {loan['balance']:,.2f}" ]) loan_table = Table(loan_data, colWidths=[30*mm, 25*mm, 25*mm, 25*mm, 20*mm, 25*mm]) loan_table.setStyle(TableStyle([ ('BACKGROUND', (0, 0), (-1, 0), colors.grey), ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), ('ALIGN', (0, 0), (-1, -1), 'CENTER'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 1, colors.black) ])) elements.append(loan_table) doc.build(elements) buffer.seek(0) return buffer except Exception as e: if user: log_report_generation(user, 'client_statement_pdf', {'client_id': client_id}, False, str(e)) logger.error(f"Client statement PDF generation failed: {e}") return None @staticmethod def generate_payment_receipt_pdf(receipt, thermal_printer=False): """Generate professional payment receipt PDF with enhanced design""" try: buffer = io.BytesIO() if thermal_printer: # Thermal printer dimensions (80mm width, ~3.15 inches) pagesize = (3.15*inch, 11*inch) # Width x Height doc = SimpleDocTemplate( buffer, pagesize=pagesize, rightMargin=10, leftMargin=10, topMargin=10, bottomMargin=10 ) else: doc = SimpleDocTemplate( buffer, pagesize=letter, rightMargin=50, leftMargin=50, topMargin=50, bottomMargin=50 ) styles = getSampleStyleSheet() elements = [] if thermal_printer: # Thermal printer styling - smaller fonts and compact layout header_style = ParagraphStyle( 'CompanyHeader', parent=styles['Heading1'], fontSize=12, spaceAfter=3, alignment=TA_CENTER, textColor=colors.black ) elements.append(Paragraph("HAVEN GRAZURI ", header_style)) subtitle_style = ParagraphStyle( 'CompanySubtitle', parent=styles['Normal'], fontSize=8, spaceAfter=10, alignment=TA_CENTER, textColor=colors.black ) elements.append(Paragraph("Microfinance Solutions", subtitle_style)) else: # Regular printer styling header_style = ParagraphStyle( 'CompanyHeader', parent=styles['Heading1'], fontSize=24, spaceAfter=5, alignment=TA_CENTER, textColor=colors.HexColor('#1f2937') ) elements.append(Paragraph("HAVEN GRAZURI ", header_style)) subtitle_style = ParagraphStyle( 'CompanySubtitle', parent=styles['Normal'], fontSize=12, spaceAfter=20, alignment=TA_CENTER, textColor=colors.HexColor('#6b7280') ) elements.append(Paragraph("Microfinance Solutions", subtitle_style)) # Receipt title with background if thermal_printer: title_style = ParagraphStyle( 'ReceiptTitle', parent=styles['Heading1'], fontSize=10, spaceAfter=10, spaceBefore=5, alignment=TA_CENTER, textColor=colors.black ) elements.append(Paragraph("PAYMENT RECEIPT", title_style)) elements.append(Spacer(1, 5)) else: title_style = ParagraphStyle( 'ReceiptTitle', parent=styles['Heading1'], fontSize=20, spaceAfter=30, spaceBefore=20, alignment=TA_CENTER, textColor=colors.white, backColor=colors.HexColor('#059669'), borderPadding=10 ) elements.append(Paragraph("PAYMENT RECEIPT", title_style)) elements.append(Spacer(1, 20)) # Receipt number and date header if thermal_printer: # Compact layout for thermal printer header_data = [ ['Receipt #:', receipt.receipt_number], ['Date:', receipt.payment_date.strftime('%d/%m/%Y %H:%M')], ] header_table = Table(header_data, colWidths=[0.8*inch, 2.2*inch]) header_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('TOPPADDING', (0, 0), (-1, -1), 2), ('BOTTOMPADDING', (0, 0), (-1, -1), 2), ])) else: header_data = [ ['Receipt Number:', receipt.receipt_number, 'Date:', receipt.payment_date.strftime('%B %d, %Y')], ] header_table = Table(header_data, colWidths=[1.5*inch, 2*inch, 1*inch, 1.5*inch]) header_table.setStyle(TableStyle([ ('BACKGROUND', (0, 0), (-1, -1), colors.HexColor('#f3f4f6')), ('TEXTCOLOR', (0, 0), (-1, -1), colors.HexColor('#1f2937')), ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (2, 0), (2, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTNAME', (3, 0), (3, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 11), ('GRID', (0, 0), (-1, -1), 1, colors.HexColor('#d1d5db')), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('TOPPADDING', (0, 0), (-1, -1), 8), ('BOTTOMPADDING', (0, 0), (-1, -1), 8), ])) elements.append(header_table) if thermal_printer: elements.append(Spacer(1, 5)) # Compact section title for thermal printer borrower_title = ParagraphStyle( 'SectionTitle', parent=styles['Heading2'], fontSize=8, spaceAfter=3, textColor=colors.black, alignment=TA_CENTER ) elements.append(Paragraph("--- BORROWER INFO ---", borrower_title)) else: elements.append(Spacer(1, 20)) # Regular section title borrower_title = ParagraphStyle( 'SectionTitle', parent=styles['Heading2'], fontSize=14, spaceAfter=10, textColor=colors.HexColor('#1f2937'), borderWidth=1, borderColor=colors.HexColor('#059669'), borderPadding=5, backColor=colors.HexColor('#ecfdf5') ) elements.append(Paragraph("BORROWER INFORMATION", borrower_title)) borrower_data = [ ['Name:', receipt.borrower.get_full_name()], ['Phone:', getattr(receipt.borrower, 'phone_number', 'N/A')], ] if not thermal_printer: borrower_data.append(['Email:', getattr(receipt.borrower, 'email', 'N/A')]) if thermal_printer: borrower_table = Table(borrower_data, colWidths=[0.8*inch, 2.2*inch]) borrower_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('TOPPADDING', (0, 0), (-1, -1), 1), ('BOTTOMPADDING', (0, 0), (-1, -1), 1), ])) else: borrower_table = Table(borrower_data, colWidths=[1.5*inch, 4.5*inch]) borrower_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6), ('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#e5e7eb')), ])) elements.append(borrower_table) if thermal_printer: elements.append(Spacer(1, 5)) else: elements.append(Spacer(1, 20)) # Loan Information Section if thermal_printer: elements.append(Paragraph("--- LOAN INFO ---", borrower_title)) else: elements.append(Paragraph("LOAN INFORMATION", borrower_title)) # Safely get loan information loan_number = 'N/A' loan_product = 'N/A' principal_amount = 'N/A' if receipt.loan: loan_number = receipt.loan.loan_number if hasattr(receipt.loan, 'application') and receipt.loan.application: if hasattr(receipt.loan.application, 'loan_product') and receipt.loan.application.loan_product: loan_product = receipt.loan.application.loan_product.name if hasattr(receipt.loan.application, 'loan_amount'): principal_amount = f"KES {receipt.loan.application.loan_amount:,.2f}" elif hasattr(receipt.loan.application, 'amount'): principal_amount = f"KES {receipt.loan.application.amount:,.2f}" if thermal_printer: # Compact loan data for thermal printer loan_data = [ ['Loan #:', loan_number], ['Amount:', f"KES {receipt.amount_paid:,.2f}"], ] loan_table = Table(loan_data, colWidths=[0.8*inch, 2.2*inch]) loan_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('TOPPADDING', (0, 0), (-1, -1), 1), ('BOTTOMPADDING', (0, 0), (-1, -1), 1), ])) else: loan_data = [ ['Loan Number:', loan_number], ['Loan Product:', loan_product], ['Principal Amount:', principal_amount], ['Previous Balance:', f"KES {getattr(receipt, 'previous_balance', 0):,.2f}"], ['New Balance:', f"KES {getattr(receipt, 'new_balance', 0):,.2f}"], ] loan_table = Table(loan_data, colWidths=[1.5*inch, 4.5*inch]) loan_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6), ('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#e5e7eb')), # Highlight new balance ('BACKGROUND', (0, 4), (-1, 4), colors.HexColor('#fef3c7')), ('FONTNAME', (1, 4), (1, 4), 'Helvetica-Bold'), ])) elements.append(loan_table) if thermal_printer: elements.append(Spacer(1, 5)) else: elements.append(Spacer(1, 20)) # Payment Details Section if thermal_printer: elements.append(Paragraph("--- PAYMENT ---", borrower_title)) else: elements.append(Paragraph("PAYMENT DETAILS", borrower_title)) # Payment method formatting payment_method = receipt.payment_method.upper() if payment_method == 'MPESA': payment_method = 'M-PESA' elif payment_method == 'BANK': payment_method = 'Bank Transfer' if thermal_printer: # Compact payment data for thermal printer payment_data = [ ['Method:', payment_method], ['Ref:', getattr(receipt, 'payment_reference', receipt.receipt_number) if hasattr(receipt, 'payment_reference') else receipt.receipt_number], ] payment_table = Table(payment_data, colWidths=[0.8*inch, 2.2*inch]) payment_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('TOPPADDING', (0, 0), (-1, -1), 1), ('BOTTOMPADDING', (0, 0), (-1, -1), 1), ])) else: payment_data = [ ['Amount Paid:', f"KES {receipt.amount_paid:,.2f}"], ['Payment Method:', payment_method], ['Transaction Reference:', getattr(receipt, 'payment_reference', receipt.receipt_number) if hasattr(receipt, 'payment_reference') else receipt.receipt_number], ['Payment Date:', receipt.payment_date.strftime('%B %d, %Y at %I:%M %p')], ['Processed By:', getattr(receipt, 'processed_by', 'System')], ] payment_table = Table(payment_data, colWidths=[1.5*inch, 4.5*inch]) payment_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTNAME', (1, 0), (1, -1), 'Helvetica'), ('FONTSIZE', (0, 0), (-1, -1), 10), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6), ('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#e5e7eb')), # Highlight amount paid ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#dcfce7')), ('FONTNAME', (1, 0), (1, 0), 'Helvetica-Bold'), ('FONTSIZE', (1, 0), (1, 0), 12), ])) elements.append(payment_table) if thermal_printer: elements.append(Spacer(1, 10)) # Compact status for thermal printer status_style = ParagraphStyle( 'StatusStyle', parent=styles['Normal'], fontSize=8, alignment=TA_CENTER, textColor=colors.black ) elements.append(Paragraph("PAYMENT CONFIRMED", status_style)) elements.append(Spacer(1, 5)) # Compact footer for thermal printer footer_style = ParagraphStyle( 'Footer', parent=styles['Normal'], fontSize=6, alignment=TA_CENTER, textColor=colors.black, spaceAfter=2 ) elements.append(Paragraph("Thank you!", footer_style)) elements.append(Paragraph("Computer generated receipt", footer_style)) elements.append(Spacer(1, 3)) # Compact contact for thermal printer contact_style = ParagraphStyle( 'Contact', parent=styles['Normal'], fontSize=5, alignment=TA_CENTER, textColor=colors.black ) elements.append(Paragraph("info@branchbusinessadvance.co.ke", contact_style)) else: elements.append(Spacer(1, 30)) # Status and verification status_style = ParagraphStyle( 'StatusStyle', parent=styles['Normal'], fontSize=12, alignment=TA_CENTER, textColor=colors.HexColor('#059669'), backColor=colors.HexColor('#dcfce7'), borderWidth=1, borderColor=colors.HexColor('#059669'), borderPadding=10 ) elements.append(Paragraph("SUCCESS: PAYMENT CONFIRMED AND PROCESSED", status_style)) elements.append(Spacer(1, 30)) # Footer with terms and contact footer_style = ParagraphStyle( 'Footer', parent=styles['Normal'], fontSize=9, alignment=TA_CENTER, textColor=colors.HexColor('#6b7280'), spaceAfter=5 ) elements.append(Paragraph("Thank you for your payment!", footer_style)) elements.append(Paragraph("This is a computer-generated receipt and does not require a signature.", footer_style)) elements.append(Spacer(1, 10)) # Contact information contact_style = ParagraphStyle( 'Contact', parent=styles['Normal'], fontSize=8, alignment=TA_CENTER, textColor=colors.HexColor('#9ca3af') ) elements.append(Paragraph("For inquiries: info@branchbusinessadvance.co.ke | +254 700 000 000", contact_style)) elements.append(Paragraph("Visit us: www.branchbusinessadvance.co.ke", contact_style)) # Build PDF doc.build(elements) buffer.seek(0) return buffer except Exception as e: logger.error(f"Enhanced payment receipt PDF generation failed: {e}") return None @staticmethod def generate_excel_report(report_type, data, user=None): """Generate Excel report for any report type""" try: buffer = io.BytesIO() workbook = xlsxwriter.Workbook(buffer) # Define formats header_format = workbook.add_format({ 'bold': True, 'bg_color': '#4472C4', 'font_color': 'white', 'border': 1 }) cell_format = workbook.add_format({ 'border': 1, 'align': 'left' }) number_format = workbook.add_format({ 'border': 1, 'num_format': '#,##0.00' }) if report_type == 'executive_summary': worksheet = workbook.add_worksheet('Executive Summary') # Headers worksheet.write('A1', 'Executive Summary Report', header_format) worksheet.write('A2', f'Generated: {timezone.now().strftime("%Y-%m-%d %H:%M:%S")}', cell_format) # Data row = 4 worksheet.write(row, 0, 'Metric', header_format) worksheet.write(row, 1, 'Value', header_format) metrics = [ ('Total Loans', data.get('total_loans', 0)), ('Active Loans', data.get('active_loans', 0)), ('Total Disbursed (KES)', data.get('total_disbursed', 0)), ('Collection Rate (%)', data.get('collection_rate', 0)), ('Default Rate (%)', data.get('default_rate', 0)) ] for metric, value in metrics: row += 1 worksheet.write(row, 0, metric, cell_format) if isinstance(value, (int, float)): worksheet.write(row, 1, value, number_format) else: worksheet.write(row, 1, value, cell_format) elif report_type == 'loan_statement': worksheet = workbook.add_worksheet('Loan Statement') # Loan details loan = data['loan'] worksheet.write('A1', 'Loan Statement', header_format) worksheet.write('A2', f'Loan Number: {loan["loan_number"]}', cell_format) worksheet.write('A3', f'Borrower: {loan["borrower_name"]}', cell_format) # Repayments if data['repayments']: row = 5 headers = ['Date', 'Amount', 'Method', 'Reference', 'Balance After'] for col, header in enumerate(headers): worksheet.write(row, col, header, header_format) for repayment in data['repayments']: row += 1 worksheet.write(row, 0, repayment['date'], cell_format) worksheet.write(row, 1, repayment['amount'], number_format) worksheet.write(row, 2, repayment['method'], cell_format) worksheet.write(row, 3, repayment['reference'] or 'N/A', cell_format) worksheet.write(row, 4, repayment['balance_after'], number_format) # Log the generation if user: log_report_generation(user, f'{report_type}_excel', {'report_type': report_type}) workbook.close() buffer.seek(0) return buffer except Exception as e: if user: log_report_generation(user, f'{report_type}_excel', {'report_type': report_type}, False, str(e)) logger.error(f"Excel report generation failed: {e}") return None # Utility functions for report generation def get_report_data(report_type, **kwargs): """Unified interface for getting report data""" generators = { 'executive_summary': OptimizedReportGenerator.generate_executive_summary, 'performance_trends': OptimizedReportGenerator.generate_loan_performance_trends, 'risk_analysis': OptimizedReportGenerator.generate_risk_analysis, 'dashboard_data': OptimizedReportGenerator.generate_dashboard_data, 'loan_statement': OptimizedReportGenerator.generate_loan_statement, 'client_statement': OptimizedReportGenerator.generate_client_statement, 'collection_statement': OptimizedReportGenerator.generate_collection_statement, } generator = generators.get(report_type) if generator: # Filter out user parameter for methods that don't need it methods_without_user = ['risk_analysis', 'dashboard_data', 'executive_summary', 'performance_trends'] if report_type in methods_without_user: # Remove user from kwargs for these methods filtered_kwargs = {k: v for k, v in kwargs.items() if k != 'user'} return generator(**filtered_kwargs) else: return generator(**kwargs) else: return {'error': f'Unknown report type: {report_type}'} def generate_report_file(report_type, format_type='pdf', **kwargs): """Generate report file in specified format""" try: user = kwargs.get('user') if format_type == 'pdf': if report_type == 'executive_summary': return OptimizedReportGenerator.generate_executive_summary_pdf() elif report_type == 'loan_statement': loan_id = kwargs.get('loan_id') return OptimizedReportGenerator.generate_loan_statement_pdf(loan_id, user) elif report_type == 'client_statement': client_id = kwargs.get('client_id') return OptimizedReportGenerator.generate_client_statement_pdf(client_id, user) elif format_type == 'excel': data = get_report_data(report_type, **kwargs) if 'error' not in data: return OptimizedReportGenerator.generate_excel_report(report_type, data, user) elif format_type == 'csv': if report_type == 'portfolio_analysis': return OptimizedReportGenerator.generate_portfolio_analysis_csv(**kwargs) else: # Convert any report to CSV format data = get_report_data(report_type, **kwargs) if 'error' not in data: return convert_to_csv(data, report_type, user) return None except Exception as e: logger.error(f"Report file generation failed: {e}") return None def convert_to_csv(data, report_type, user=None): """Convert report data to CSV format""" try: output = io.StringIO() writer = csv.writer(output) # Header writer.writerow([f'{report_type.replace("_", " ").title()} Report']) writer.writerow(['Generated:', timezone.now().strftime('%Y-%m-%d %H:%M:%S')]) writer.writerow([]) if report_type == 'executive_summary': writer.writerow(['Metric', 'Value']) writer.writerow(['Total Loans', data.get('total_loans', 0)]) writer.writerow(['Active Loans', data.get('active_loans', 0)]) writer.writerow(['Total Disbursed (KES)', f"{data.get('total_disbursed', 0):,.2f}"]) writer.writerow(['Collection Rate (%)', f"{data.get('collection_rate', 0):.2f}"]) writer.writerow(['Default Rate (%)', f"{data.get('default_rate', 0):.2f}"]) elif report_type == 'loan_statement': loan = data['loan'] writer.writerow(['Loan Details']) writer.writerow(['Loan Number', loan['loan_number']]) writer.writerow(['Borrower', loan['borrower_name']]) writer.writerow(['Principal Amount (KES)', f"{loan['principal_amount']:,.2f}"]) writer.writerow(['Status', loan['status']]) writer.writerow([]) if data['repayments']: writer.writerow(['Repayment History']) writer.writerow(['Date', 'Amount', 'Method', 'Reference', 'Balance After']) for rep in data['repayments']: writer.writerow([ rep['date'], f"{rep['amount']:,.2f}", rep['method'], rep['reference'] or 'N/A', f"{rep['balance_after']:,.2f}" ]) # Log the generation if user: log_report_generation(user, f'{report_type}_csv', {'report_type': report_type}) return output.getvalue() except Exception as e: if user: log_report_generation(user, f'{report_type}_csv', {'report_type': report_type}, False, str(e)) logger.error(f"CSV conversion failed: {e}") return None # Quick report generation functions for common use cases def quick_executive_summary(branch_id=None): """Generate quick executive summary for dashboard""" return OptimizedReportGenerator.generate_executive_summary(30, branch_id) def quick_risk_report(): """Generate quick risk report for immediate attention""" return OptimizedReportGenerator.generate_risk_analysis() def quick_portfolio_summary(portfolio_manager_id=None): """Generate quick portfolio summary for managers""" try: cache_key = f"quick_portfolio_{portfolio_manager_id or 'all'}" cached_data = OptimizedReportGenerator._get_cached_data(cache_key) if cached_data: return cached_data # Get portfolio data from users.models import CustomUser if portfolio_manager_id: manager = CustomUser.objects.get(id=portfolio_manager_id, role='portfolio_manager') clients = manager.managed_clients.all() else: clients = CustomUser.objects.filter(role='borrower') # Calculate portfolio metrics total_clients = clients.count() active_loans = Loan.objects.filter( application__applicant__in=clients, status__in=['approved', 'disbursed', 'active'] ).count() total_portfolio_value = Loan.objects.filter( application__applicant__in=clients, status__in=['approved', 'disbursed', 'active'] ).aggregate( total=Sum('application__loan_amount') )['total'] or 0 overdue_loans = Loan.objects.filter( application__applicant__in=clients, status='overdue' ).count() summary = { 'total_clients': total_clients, 'active_loans': active_loans, 'total_portfolio_value': float(total_portfolio_value), 'overdue_loans': overdue_loans, 'portfolio_at_risk': (overdue_loans / active_loans * 100) if active_loans > 0 else 0, 'generated_at': timezone.now().isoformat() } # Cache for 10 minutes OptimizedReportGenerator._set_cached_data(cache_key, summary, 600) return summary except Exception as e: logger.error(f"Quick portfolio summary generation failed: {e}") return {'error': str(e)} def generate_bulk_receipts(repayment_ids, user=None): """Generate multiple receipts in bulk for efficiency""" try: receipts = [] failed_receipts = [] for repayment_id in repayment_ids: try: repayment = Repayment.objects.get(id=repayment_id) receipt_pdf = OptimizedReportGenerator.generate_payment_receipt_pdf(repayment) if receipt_pdf: receipts.append({ 'repayment_id': repayment_id, 'receipt_number': getattr(repayment, 'receipt_number', f'RCP-{repayment_id:06d}'), 'pdf_data': receipt_pdf }) else: failed_receipts.append(repayment_id) except Exception as e: logger.error(f"Failed to generate receipt for repayment {repayment_id}: {e}") failed_receipts.append(repayment_id) # Log bulk operation if user: log_report_generation( user, 'bulk_receipts', { 'total_requested': len(repayment_ids), 'successful': len(receipts), 'failed': len(failed_receipts) } ) return { 'receipts': receipts, 'failed_receipts': failed_receipts, 'success_count': len(receipts), 'failure_count': len(failed_receipts) } except Exception as e: logger.error(f"Bulk receipt generation failed: {e}") return {'error': str(e)} def generate_monthly_report_package(month=None, year=None, user=None): """Generate comprehensive monthly report package""" try: if not month: month = timezone.now().month if not year: year = timezone.now().year # Generate all monthly reports reports = {} # Executive summary reports['executive_summary'] = OptimizedReportGenerator.generate_executive_summary(30) # Performance trends reports['performance_trends'] = OptimizedReportGenerator.generate_loan_performance_trends() # Risk analysis reports['risk_analysis'] = OptimizedReportGenerator.generate_risk_analysis() # Dashboard data reports['dashboard_data'] = OptimizedReportGenerator.generate_dashboard_data() # Generate PDFs pdf_reports = {} pdf_reports['executive_summary'] = OptimizedReportGenerator.generate_executive_summary_pdf() package = { 'month': month, 'year': year, 'generated_at': timezone.now().isoformat(), 'data_reports': reports, 'pdf_reports': pdf_reports, 'summary': { 'total_reports': len(reports), 'pdf_reports': len(pdf_reports), 'generation_time': timezone.now().isoformat() } } # Log package generation if user: log_report_generation( user, 'monthly_package', {'month': month, 'year': year} ) return package except Exception as e: logger.error(f"Monthly report package generation failed: {e}") return {'error': str(e)} # Performance monitoring for reports def get_report_performance_metrics(): """Get performance metrics for report generation""" try: from django.core.cache import cache # Get cache statistics cache_stats = { 'cache_hits': cache.get('report_cache_hits', 0), 'cache_misses': cache.get('report_cache_misses', 0), 'total_requests': cache.get('report_total_requests', 0) } # Calculate hit rate if cache_stats['total_requests'] > 0: cache_stats['hit_rate'] = (cache_stats['cache_hits'] / cache_stats['total_requests']) * 100 else: cache_stats['hit_rate'] = 0 # Get recent report generation logs recent_logs = AuditLog.objects.filter( action='generate_report', timestamp__gte=timezone.now() - timedelta(hours=24) ).count() return { 'cache_statistics': cache_stats, 'reports_generated_24h': recent_logs, 'last_updated': timezone.now().isoformat() } except Exception as e: logger.error(f"Performance metrics retrieval failed: {e}") return {'error': str(e)} # Legacy compatibility class AdvancedReportGenerator(OptimizedReportGenerator): """Legacy compatibility class""" pass