from sqlalchemy.orm import Session from typing import List, Dict, Any, Optional from io import BytesIO from datetime import datetime import uuid from reportlab.lib.pagesizes import A4, letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak, Image from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import inch, cm from reportlab.lib import colors from reportlab.graphics.shapes import Drawing from reportlab.graphics.charts.barcharts import VerticalBarChart from reportlab.graphics.charts.piecharts import Pie from reportlab.graphics.charts.linecharts import HorizontalLineChart from app.models.main.simulation import TblSimulation from app.models.main.group import TblGroup from app.models.main.student import TblStudent from app.locale.messages import Messages from .schemas import ( KPIMetric, FinancialMetrics, OperationalMetrics, StudentPerformance, GroupPerformance, SimulationSummary, ComprehensiveReportRequest, PDFReportResponse ) class PDFReportService: def __init__(self, db: Session): self.db = db self.styles = getSampleStyleSheet() self._setup_custom_styles() def _setup_custom_styles(self): """Setup custom paragraph styles for reports""" self.title_style = ParagraphStyle( 'CustomTitle', parent=self.styles['Heading1'], fontSize=18, spaceAfter=20, alignment=1, fontName='Helvetica-Bold', textColor=colors.darkblue ) self.section_style = ParagraphStyle( 'SectionHeader', parent=self.styles['Heading2'], fontSize=14, spaceAfter=12, spaceBefore=15, fontName='Helvetica-Bold', textColor=colors.darkgreen ) self.kpi_style = ParagraphStyle( 'KPIStyle', parent=self.styles['Normal'], fontSize=10, fontName='Helvetica-Bold', textColor=colors.darkred ) def calculate_financial_metrics(self, group_id: int) -> FinancialMetrics: """Calculate financial KPIs for a group""" # This would integrate with your existing module services # For now, returning sample calculations # Get data from various modules total_investment = self._get_total_investment(group_id) revenue = self._get_estimated_revenue(group_id) costs = self._get_total_costs(group_id) gross_profit = revenue - costs net_profit = gross_profit * 0.85 # Assuming 15% additional expenses return FinancialMetrics( revenue=revenue, gross_profit=gross_profit, net_profit=net_profit, gross_margin_percentage=(gross_profit / revenue * 100) if revenue > 0 else 0, net_margin_percentage=(net_profit / revenue * 100) if revenue > 0 else 0, roi=(net_profit / total_investment * 100) if total_investment > 0 else 0, total_investment=total_investment, cash_flow=net_profit * 0.9 # Simplified cash flow ) def calculate_operational_metrics(self, group_id: int) -> OperationalMetrics: """Calculate operational KPIs for a group""" # Sample operational metrics calculation return OperationalMetrics( total_stores=self._get_planned_stores(group_id), sales_per_sqft=self._get_sales_per_sqft(group_id), inventory_turnover=12.0, # Sample value customer_footfall=self._get_estimated_footfall(group_id), average_transaction_value=self._get_avg_transaction_value(group_id), conversion_rate=15.5 # Sample conversion rate ) def _get_total_investment(self, group_id: int) -> float: """Get total investment from Module 1 capital expenses""" from app.api.module_reports_6_to_10.service import Module1ReportService try: module1_service = Module1ReportService(self.db) report = module1_service.generate_module1_report(group_id) return report.capital_expenses_format_a.total + report.capital_expenses_format_b.total except: return 1000000.0 # Default value def _get_estimated_revenue(self, group_id: int) -> float: """Get estimated revenue from Module 2 sales estimates""" from app.api.module_reports_6_to_10.service import Module2ReportService try: module2_service = Module2ReportService(self.db) report = module2_service.generate_module2_report(group_id) return (report.sales_estimate_format_a.sales_value + report.sales_estimate_format_b.sales_value) * 12 # Annual except: return 5000000.0 # Default value def _get_total_costs(self, group_id: int) -> float: """Calculate total operational costs""" # This would sum up costs from various modules return 3500000.0 # Sample value def _get_planned_stores(self, group_id: int) -> int: """Get planned number of stores from Module 3""" try: from app.api.module_reports_6_to_10.service import Module3ReportService module3_service = Module3ReportService(self.db) report = module3_service.generate_module3_report(group_id) return report.target_stores except: return 5 # Default value def _get_sales_per_sqft(self, group_id: int) -> float: """Calculate sales per square foot""" return 2500.0 # Sample value def _get_estimated_footfall(self, group_id: int) -> int: """Get estimated customer footfall""" return 15000 # Sample monthly footfall def _get_avg_transaction_value(self, group_id: int) -> float: """Get average transaction value""" return 850.0 # Sample ATV def generate_kpi_dashboard_chart(self, metrics: FinancialMetrics) -> Drawing: """Generate KPI dashboard chart""" drawing = Drawing(400, 200) # Create bar chart for key financial metrics chart = VerticalBarChart() chart.x = 50 chart.y = 50 chart.height = 125 chart.width = 300 chart.data = [[ metrics.revenue / 1000000, # Convert to millions metrics.gross_profit / 1000000, metrics.net_profit / 1000000, metrics.total_investment / 1000000 ]] chart.categoryAxis.categoryNames = ['Revenue', 'Gross Profit', 'Net Profit', 'Investment'] chart.valueAxis.valueMin = 0 chart.bars[0].fillColor = colors.lightblue drawing.add(chart) return drawing def generate_comprehensive_report(self, request: ComprehensiveReportRequest) -> BytesIO: """Generate comprehensive PDF report with KPIs and metrics""" buffer = BytesIO() doc = SimpleDocTemplate( buffer, pagesize=A4, leftMargin=50, rightMargin=50, topMargin=50, bottomMargin=50 ) story = [] # Report Header story.append(Paragraph("Retail Simulation Performance Report", self.title_style)) story.append(Paragraph(f"Generated on: {datetime.now().strftime('%B %d, %Y at %I:%M %p')}", self.styles['Normal'])) story.append(Spacer(1, 20)) if request.group_id: # Group-specific report story.extend(self._generate_group_report_content(request.group_id, request.include_charts)) elif request.simulation_id: # Simulation-wide report story.extend(self._generate_simulation_report_content(request.simulation_id, request.include_charts)) doc.build(story) buffer.seek(0) return buffer def _generate_group_report_content(self, group_id: int, include_charts: bool = True) -> List: """Generate content for group-specific report""" story = [] # Get group information group = self.db.query(TblGroup).filter(TblGroup.group_id == group_id).first() if not group: story.append(Paragraph("Group not found", self.styles['Normal'])) return story # Group Overview story.append(Paragraph("Group Overview", self.section_style)) group_info_data = [ ['Group Name', group.group_name or 'N/A'], ['Group ID', str(group.group_id)], ['Category', group.category or 'N/A'], ['Members', str(group.number_of_members or 0)] ] group_table = Table(group_info_data, colWidths=[2*inch, 3*inch]) group_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('BACKGROUND', (0, 0), (0, -1), colors.lightgrey) ])) story.append(group_table) story.append(Spacer(1, 20)) # Financial Performance story.append(Paragraph("Financial Performance & KPIs", self.section_style)) financial_metrics = self.calculate_financial_metrics(group_id) operational_metrics = self.calculate_operational_metrics(group_id) # Key Financial KPIs Table financial_kpis = [ ['KPI', 'Value', 'Status'], ['Total Revenue', f"Rs {financial_metrics.revenue:,.0f}", self._get_kpi_status(financial_metrics.revenue, 4000000)], ['Gross Profit', f"Rs {financial_metrics.gross_profit:,.0f}", self._get_kpi_status(financial_metrics.gross_profit, 1500000)], ['Net Profit', f"Rs {financial_metrics.net_profit:,.0f}", self._get_kpi_status(financial_metrics.net_profit, 800000)], ['Gross Margin %', f"{financial_metrics.gross_margin_percentage:.1f}%", self._get_kpi_status(financial_metrics.gross_margin_percentage, 25)], ['Net Margin %', f"{financial_metrics.net_margin_percentage:.1f}%", self._get_kpi_status(financial_metrics.net_margin_percentage, 15)], ['ROI %', f"{financial_metrics.roi:.1f}%", self._get_kpi_status(financial_metrics.roi, 20)], ['Total Investment', f"Rs {financial_metrics.total_investment:,.0f}", 'Baseline'] ] financial_table = Table(financial_kpis, colWidths=[2.5*inch, 1.5*inch, 1*inch]) financial_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('BACKGROUND', (0, 0), (-1, 0), colors.lightblue), ('FONTSIZE', (0, 0), (-1, -1), 9) ])) story.append(financial_table) story.append(Spacer(1, 15)) # Operational KPIs Table story.append(Paragraph("Operational Performance & KPIs", self.section_style)) operational_kpis = [ ['KPI', 'Value', 'Target', 'Status'], ['Total Stores Planned', str(operational_metrics.total_stores), '5-10', self._get_kpi_status(operational_metrics.total_stores, 5)], ['Sales per Sq.Ft', f"Rs {operational_metrics.sales_per_sqft:,.0f}", 'Rs 2,000+', self._get_kpi_status(operational_metrics.sales_per_sqft, 2000)], ['Inventory Turnover', f"{operational_metrics.inventory_turnover:.1f}x", '10x+', self._get_kpi_status(operational_metrics.inventory_turnover, 10)], ['Monthly Footfall', f"{operational_metrics.customer_footfall:,}", '12,000+', self._get_kpi_status(operational_metrics.customer_footfall, 12000)], ['Avg Transaction Value', f"Rs {operational_metrics.average_transaction_value:.0f}", 'Rs 800+', self._get_kpi_status(operational_metrics.average_transaction_value, 800)], ['Conversion Rate', f"{operational_metrics.conversion_rate:.1f}%", '15%+', self._get_kpi_status(operational_metrics.conversion_rate, 15)] ] operational_table = Table(operational_kpis, colWidths=[2*inch, 1.2*inch, 1*inch, 0.8*inch]) operational_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('BACKGROUND', (0, 0), (-1, 0), colors.lightgreen), ('FONTSIZE', (0, 0), (-1, -1), 9) ])) story.append(operational_table) story.append(Spacer(1, 20)) # Module Completion Status story.append(Paragraph("Module Completion Status", self.section_style)) module_status = self._get_module_completion_status(group_id) module_table = Table(module_status, colWidths=[3*inch, 1.5*inch, 1.5*inch]) module_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('BACKGROUND', (0, 0), (-1, 0), colors.lightyellow), ('FONTSIZE', (0, 0), (-1, -1), 9) ])) story.append(module_table) story.append(Spacer(1, 20)) # Performance Analysis story.append(Paragraph("Performance Analysis & Recommendations", self.section_style)) analysis = self._generate_performance_analysis(financial_metrics, operational_metrics) for point in analysis: story.append(Paragraph(f"• {point}", self.styles['Normal'])) story.append(PageBreak()) # Detailed Module Reports Summary story.append(Paragraph("Detailed Module Analysis", self.section_style)) story.extend(self._generate_module_summary(group_id)) return story def _generate_simulation_report_content(self, simulation_id: int, include_charts: bool = True) -> List: """Generate content for simulation-wide report""" story = [] # Get simulation information simulation = self.db.query(TblSimulation).filter(TblSimulation.simulation_id == simulation_id).first() if not simulation: story.append(Paragraph("Simulation not found", self.styles['Normal'])) return story # Simulation Overview story.append(Paragraph("Simulation Overview", self.section_style)) # Get all groups in simulation groups = self.db.query(TblGroup).filter(TblGroup.simulation_code == simulation.simulation_code).all() sim_info_data = [ ['Simulation Name', simulation.simulation_name or 'N/A'], ['Simulation ID', str(simulation.simulation_id)], ['Total Groups', str(len(groups))], ['Total Members', str(sum(g.number_of_members or 0 for g in groups))], ['Start Date', simulation.created_at.strftime('%Y-%m-%d') if simulation.created_at else 'N/A'] ] sim_table = Table(sim_info_data, colWidths=[2*inch, 3*inch]) sim_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('BACKGROUND', (0, 0), (0, -1), colors.lightgrey) ])) story.append(sim_table) story.append(Spacer(1, 20)) # Group Performance Comparison story.append(Paragraph("Group Performance Leaderboard", self.section_style)) group_performances = [] for group in groups: financial_metrics = self.calculate_financial_metrics(group.group_id) operational_metrics = self.calculate_operational_metrics(group.group_id) # Calculate overall score score = self._calculate_overall_score(financial_metrics, operational_metrics) group_performances.append({ 'group_name': group.group_name or f'Group {group.group_id}', 'participants': group.number_of_members or 0, 'score': score, 'roi': financial_metrics.roi, 'revenue': financial_metrics.revenue }) # Sort by score group_performances.sort(key=lambda x: x['score'], reverse=True) # Create leaderboard table leaderboard_data = [['Rank', 'Group Name', 'Members', 'Overall Score', 'ROI %', 'Revenue']] for i, perf in enumerate(group_performances, 1): leaderboard_data.append([ str(i), perf['group_name'], str(perf['participants']), f"{perf['score']:.1f}", f"{perf['roi']:.1f}%", f"Rs {perf['revenue']:,.0f}" ]) leaderboard_table = Table(leaderboard_data, colWidths=[0.5*inch, 2*inch, 0.8*inch, 1*inch, 0.8*inch, 1.4*inch]) leaderboard_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'CENTER'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('BACKGROUND', (0, 0), (-1, 0), colors.gold), ('FONTSIZE', (0, 0), (-1, -1), 9), # Highlight top 3 ('BACKGROUND', (0, 1), (-1, 1), colors.lightgreen), # 1st place ('BACKGROUND', (0, 2), (-1, 2), colors.lightblue) if len(group_performances) > 1 else None, # 2nd place ('BACKGROUND', (0, 3), (-1, 3), colors.lightyellow) if len(group_performances) > 2 else None # 3rd place ])) story.append(leaderboard_table) story.append(Spacer(1, 20)) # Simulation Statistics story.append(Paragraph("Simulation Statistics", self.section_style)) # Calculate aggregate statistics total_revenue = sum(self.calculate_financial_metrics(g.group_id).revenue for g in groups) avg_roi = sum(self.calculate_financial_metrics(g.group_id).roi for g in groups) / len(groups) if groups else 0 completion_rate = self._calculate_simulation_completion_rate(simulation_id) stats_data = [ ['Total Revenue (All Groups)', f"Rs {total_revenue:,.0f}"], ['Average ROI', f"{avg_roi:.1f}%"], ['Completion Rate', f"{completion_rate:.1f}%"], ['Best Performing Group', group_performances[0]['group_name'] if group_performances else 'N/A'], ['Highest ROI', f"{max(p['roi'] for p in group_performances):.1f}%" if group_performances else 'N/A'] ] stats_table = Table(stats_data, colWidths=[3*inch, 2*inch]) stats_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('BACKGROUND', (0, 0), (0, -1), colors.lightcyan) ])) story.append(stats_table) return story def _get_kpi_status(self, value: float, target: float) -> str: """Determine KPI status based on value vs target""" if value >= target: return "✓ Good" elif value >= target * 0.8: return "⚠ Fair" else: return "✗ Poor" def _get_module_completion_status(self, group_id: int) -> List[List[str]]: """Get module completion status for a group""" modules = [ "Module 1: Brand, Customers & Stores", "Module 2: Location Analysis", "Module 3: Network Planning", "Module 4: Category Management", "Module 5: Vendor Management", "Module 6: Pricing & Promotions", "Module 7: Marketing & Communications", "Module 8: Services & Utilities", "Module 9: Backend & Supply Chain" ] status_data = [['Module', 'Status', 'Completion %']] for i, module in enumerate(modules, 1): # This would check actual completion status from database # For now, using sample data completion = min(100, max(0, 85 + (i * 2))) # Sample completion rates status = "Completed" if completion == 100 else "In Progress" if completion > 50 else "Not Started" status_data.append([module, status, f"{completion}%"]) return status_data def _generate_performance_analysis(self, financial: FinancialMetrics, operational: OperationalMetrics) -> List[str]: """Generate performance analysis points""" analysis = [] if financial.roi > 20: analysis.append("Excellent ROI performance indicates strong business model and execution") elif financial.roi > 10: analysis.append("Good ROI performance with room for optimization in cost management") else: analysis.append("ROI below expectations - review pricing strategy and cost structure") if financial.gross_margin_percentage > 25: analysis.append("Strong gross margins indicate effective pricing and supplier management") else: analysis.append("Gross margins need improvement - consider category mix optimization") if operational.sales_per_sqft > 2000: analysis.append("Excellent space utilization with high sales per square foot") else: analysis.append("Space productivity can be improved through better merchandising") if operational.conversion_rate > 15: analysis.append("Good customer conversion rate indicates effective store operations") else: analysis.append("Focus on improving customer experience to increase conversion") return analysis def _generate_module_summary(self, group_id: int) -> List: """Generate summary of all modules for the group""" story = [] # This would integrate with existing module report services # For brevity, showing structure for key modules try: from app.api.module_reports_6_to_10.service import Module1ReportService, Module2ReportService # Module 1 Summary story.append(Paragraph("Module 1: Brand & Store Investment Summary", self.styles['Heading3'])) module1_service = Module1ReportService(self.db) module1_data = module1_service.generate_module1_report(group_id) story.append(Paragraph(f"Brand Name: {module1_data.brand_details.brand_name}", self.styles['Normal'])) story.append(Paragraph(f"Total Investment Format A: Rs {module1_data.capital_expenses_format_a.total:,.0f}", self.styles['Normal'])) story.append(Paragraph(f"Total Investment Format B: Rs {module1_data.capital_expenses_format_b.total:,.0f}", self.styles['Normal'])) story.append(Spacer(1, 10)) # Module 2 Summary story.append(Paragraph("Module 2: Location & Sales Analysis", self.styles['Heading3'])) module2_service = Module2ReportService(self.db) module2_data = module2_service.generate_module2_report(group_id) story.append(Paragraph(f"Estimated Market Share: {module2_data.estimated_market_share:.1f}%", self.styles['Normal'])) story.append(Paragraph(f"Sales Estimate Format A: Rs {module2_data.sales_estimate_format_a.sales_value:,.0f}", self.styles['Normal'])) story.append(Paragraph(f"Sales Estimate Format B: Rs {module2_data.sales_estimate_format_b.sales_value:,.0f}", self.styles['Normal'])) story.append(Spacer(1, 10)) except Exception as e: story.append(Paragraph(f"Module data unavailable: {str(e)}", self.styles['Normal'])) return story def _calculate_overall_score(self, financial: FinancialMetrics, operational: OperationalMetrics) -> float: """Calculate overall performance score""" # Weighted scoring system roi_score = min(100, financial.roi * 2) # ROI weight: 40% margin_score = min(100, financial.gross_margin_percentage * 2) # Margin weight: 30% sales_score = min(100, operational.sales_per_sqft / 50) # Sales/sqft weight: 20% conversion_score = min(100, operational.conversion_rate * 4) # Conversion weight: 10% overall_score = (roi_score * 0.4 + margin_score * 0.3 + sales_score * 0.2 + conversion_score * 0.1) return overall_score def _calculate_simulation_completion_rate(self, simulation_id: int) -> float: """Calculate overall completion rate for simulation""" # This would check actual module completion data return 78.5 # Sample completion rate def generate_student_report(self, student_id: int) -> BytesIO: """Generate individual student performance report""" buffer = BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4, leftMargin=50, rightMargin=50, topMargin=50, bottomMargin=50) story = [] # Get student information student = self.db.query(TblStudent).filter(TblStudent.student_id == student_id).first() if not student: story.append(Paragraph("Student not found", self.styles['Normal'])) doc.build(story) buffer.seek(0) return buffer # Student Report Header story.append(Paragraph("Individual Student Performance Report", self.title_style)) story.append(Spacer(1, 20)) # Student Information student_info = [ ['Student Name', f"{student.first_name or ''} {student.last_name or ''}" or 'N/A'], ['Student ID', str(student.student_id)], ['Group Code', student.group_code or 'N/A'], ['Login Status', 'Active' if student.login_status == '1' else 'Inactive'] ] student_table = Table(student_info, colWidths=[2*inch, 3*inch]) student_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('BACKGROUND', (0, 0), (0, -1), colors.lightgrey) ])) story.append(student_table) story.append(Spacer(1, 20)) # If student belongs to a group, show group performance if student.group_code: # Get group by group_code group = self.db.query(TblGroup).filter(TblGroup.group_code == student.group_code).first() if group: story.append(Paragraph("Group Performance Summary", self.section_style)) financial_metrics = self.calculate_financial_metrics(group.group_id) group_performance = [ ['Metric', 'Value'], ['Revenue', f"Rs {financial_metrics.revenue:,.0f}"], ['Net Profit', f"Rs {financial_metrics.net_profit:,.0f}"], ['ROI', f"{financial_metrics.roi:.1f}%"], ['Gross Margin', f"{financial_metrics.gross_margin_percentage:.1f}%"] ] perf_table = Table(group_performance, colWidths=[2*inch, 2*inch]) perf_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'), ('BACKGROUND', (0, 0), (-1, 0), colors.lightblue) ])) story.append(perf_table) doc.build(story) buffer.seek(0) return buffer def create_report_response(self, buffer: BytesIO, report_type: str) -> PDFReportResponse: """Create standardized report response""" report_id = str(uuid.uuid4()) file_name = f"{report_type}_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf" return PDFReportResponse( report_id=report_id, file_name=file_name, file_size=len(buffer.getvalue()), generation_time=datetime.now(), report_type=report_type, status="completed" )