""" CXInsights - Excel Export Exports analysis results to Excel format with multiple sheets. """ from datetime import datetime from pathlib import Path from src.aggregation.models import BatchAggregation from src.models.call_analysis import CallAnalysis # Try to import openpyxl, provide fallback try: from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter OPENPYXL_AVAILABLE = True except ImportError: OPENPYXL_AVAILABLE = False def export_to_excel( batch_id: str, aggregation: BatchAggregation, analyses: list[CallAnalysis], output_dir: Path, ) -> Path: """ Export results to Excel file. Creates workbook with sheets: - Summary: High-level metrics - Lost Sales Drivers: Driver frequencies and severity - Poor CX Drivers: Driver frequencies and severity - Call Details: Individual call results - Emergent Patterns: New patterns found Args: batch_id: Batch identifier aggregation: Aggregation results analyses: Individual call analyses output_dir: Output directory Returns: Path to Excel file """ if not OPENPYXL_AVAILABLE: raise ImportError( "openpyxl is required for Excel export. " "Install with: pip install openpyxl" ) output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) wb = Workbook() # Style definitions header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") header_alignment = Alignment(horizontal="center", vertical="center") # Sheet 1: Summary ws_summary = wb.active ws_summary.title = "Summary" _create_summary_sheet(ws_summary, batch_id, aggregation, analyses, header_font, header_fill) # Sheet 2: Lost Sales Drivers ws_lost_sales = wb.create_sheet("Lost Sales Drivers") _create_drivers_sheet( ws_lost_sales, aggregation.lost_sales_frequencies, aggregation.lost_sales_severities, header_font, header_fill, ) # Sheet 3: Poor CX Drivers ws_poor_cx = wb.create_sheet("Poor CX Drivers") _create_drivers_sheet( ws_poor_cx, aggregation.poor_cx_frequencies, aggregation.poor_cx_severities, header_font, header_fill, ) # Sheet 4: Call Details ws_calls = wb.create_sheet("Call Details") _create_calls_sheet(ws_calls, analyses, header_font, header_fill) # Sheet 5: Emergent Patterns if aggregation.emergent_patterns: ws_emergent = wb.create_sheet("Emergent Patterns") _create_emergent_sheet(ws_emergent, aggregation.emergent_patterns, header_font, header_fill) # Save workbook output_path = output_dir / f"{batch_id}_analysis.xlsx" wb.save(output_path) return output_path def _create_summary_sheet(ws, batch_id, aggregation, analyses, header_font, header_fill): """Create summary sheet.""" # Title ws["A1"] = "CXInsights Analysis Report" ws["A1"].font = Font(bold=True, size=16) ws.merge_cells("A1:D1") ws["A2"] = f"Batch ID: {batch_id}" ws["A3"] = f"Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}" # Metrics section row = 5 ws[f"A{row}"] = "Key Metrics" ws[f"A{row}"].font = Font(bold=True, size=12) metrics = [ ("Total Calls Analyzed", aggregation.total_calls_processed), ("Successful Analyses", aggregation.successful_analyses), ("Failed Analyses", aggregation.failed_analyses), ("Success Rate", f"{aggregation.successful_analyses / aggregation.total_calls_processed * 100:.1f}%" if aggregation.total_calls_processed > 0 else "N/A"), ] if aggregation.rca_tree: tree = aggregation.rca_tree metrics.extend([ ("", ""), ("Calls with Lost Sales Issues", tree.calls_with_lost_sales), ("Calls with Poor CX Issues", tree.calls_with_poor_cx), ("Calls with Both Issues", tree.calls_with_both), ]) for i, (label, value) in enumerate(metrics): ws[f"A{row + 1 + i}"] = label ws[f"B{row + 1 + i}"] = value # Top drivers section row = row + len(metrics) + 3 ws[f"A{row}"] = "Top Lost Sales Drivers" ws[f"A{row}"].font = Font(bold=True, size=12) for i, freq in enumerate(aggregation.lost_sales_frequencies[:5]): ws[f"A{row + 1 + i}"] = freq.driver_code ws[f"B{row + 1 + i}"] = f"{freq.call_rate:.1%}" row = row + 7 ws[f"A{row}"] = "Top Poor CX Drivers" ws[f"A{row}"].font = Font(bold=True, size=12) for i, freq in enumerate(aggregation.poor_cx_frequencies[:5]): ws[f"A{row + 1 + i}"] = freq.driver_code ws[f"B{row + 1 + i}"] = f"{freq.call_rate:.1%}" # Adjust column widths ws.column_dimensions["A"].width = 30 ws.column_dimensions["B"].width = 20 def _create_drivers_sheet(ws, frequencies, severities, header_font, header_fill): """Create driver analysis sheet.""" headers = [ "Rank", "Driver Code", "Occurrences", "Calls Affected", "Call Rate", "Avg Confidence", "Severity Score", "Impact Level", ] # Write headers for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.font = header_font cell.fill = header_fill # Create severity lookup sev_map = {s.driver_code: s for s in severities} # Write data for row, freq in enumerate(frequencies, 2): sev = sev_map.get(freq.driver_code) ws.cell(row=row, column=1, value=row - 1) ws.cell(row=row, column=2, value=freq.driver_code) ws.cell(row=row, column=3, value=freq.total_occurrences) ws.cell(row=row, column=4, value=freq.calls_affected) ws.cell(row=row, column=5, value=f"{freq.call_rate:.1%}") ws.cell(row=row, column=6, value=f"{freq.avg_confidence:.2f}") ws.cell(row=row, column=7, value=f"{sev.severity_score:.1f}" if sev else "N/A") ws.cell(row=row, column=8, value=sev.impact_level.value if sev else "N/A") # Adjust column widths for col in range(1, len(headers) + 1): ws.column_dimensions[get_column_letter(col)].width = 15 def _create_calls_sheet(ws, analyses, header_font, header_fill): """Create call details sheet.""" headers = [ "Call ID", "Outcome", "Status", "Lost Sales Drivers", "Poor CX Drivers", "Audio Duration (s)", ] # Write headers for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.font = header_font cell.fill = header_fill # Write data for row, analysis in enumerate(analyses, 2): lost_sales = ", ".join(d.driver_code for d in analysis.lost_sales_drivers) poor_cx = ", ".join(d.driver_code for d in analysis.poor_cx_drivers) ws.cell(row=row, column=1, value=analysis.call_id) ws.cell(row=row, column=2, value=analysis.outcome.value) ws.cell(row=row, column=3, value=analysis.status.value) ws.cell(row=row, column=4, value=lost_sales or "-") ws.cell(row=row, column=5, value=poor_cx or "-") ws.cell(row=row, column=6, value=analysis.observed.audio_duration_sec) # Adjust column widths ws.column_dimensions["A"].width = 15 ws.column_dimensions["B"].width = 20 ws.column_dimensions["C"].width = 12 ws.column_dimensions["D"].width = 40 ws.column_dimensions["E"].width = 40 ws.column_dimensions["F"].width = 18 def _create_emergent_sheet(ws, emergent_patterns, header_font, header_fill): """Create emergent patterns sheet.""" headers = [ "Proposed Label", "Occurrences", "Avg Confidence", "Sample Evidence", ] # Write headers for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.font = header_font cell.fill = header_fill # Write data for row, pattern in enumerate(emergent_patterns, 2): evidence = "; ".join(pattern.get("sample_evidence", [])[:2]) ws.cell(row=row, column=1, value=pattern.get("proposed_label", "N/A")) ws.cell(row=row, column=2, value=pattern.get("occurrences", 0)) ws.cell(row=row, column=3, value=f"{pattern.get('avg_confidence', 0):.2f}") ws.cell(row=row, column=4, value=evidence[:100] if evidence else "-") # Adjust column widths ws.column_dimensions["A"].width = 30 ws.column_dimensions["B"].width = 12 ws.column_dimensions["C"].width = 15 ws.column_dimensions["D"].width = 60