Spaces:
Running
Running
| import io | |
| from datetime import datetime | |
| from reportlab.lib.pagesizes import letter | |
| from reportlab.lib import colors | |
| from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle | |
| from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak, HRFlowable | |
| def build_pdf(ai_report_text, summary_data, flagged_df): | |
| """ | |
| Generate the AML Compliance Monitoring PDF using ReportLab | |
| Returns PDF file as bytes | |
| """ | |
| buffer = io.BytesIO() | |
| doc = SimpleDocTemplate(buffer, pagesize=letter, rightMargin=72, leftMargin=72, topMargin=72, bottomMargin=18) | |
| styles = getSampleStyleSheet() | |
| title_style = ParagraphStyle( | |
| "CoverTitle", | |
| parent=styles['Heading1'], | |
| fontName="Helvetica-Bold", | |
| fontSize=24, | |
| textColor=colors.HexColor("#0a1628"), | |
| alignment=1, # Center | |
| spaceAfter=20 | |
| ) | |
| subtitle_style = ParagraphStyle( | |
| "CoverSubtitle", | |
| parent=styles['Heading2'], | |
| fontName="Helvetica", | |
| fontSize=16, | |
| textColor=colors.HexColor("#e63946"), | |
| alignment=1, | |
| spaceAfter=40 | |
| ) | |
| normal_center = ParagraphStyle( | |
| "NormalCenter", | |
| parent=styles['Normal'], | |
| fontName="Helvetica", | |
| fontSize=12, | |
| alignment=1, | |
| spaceAfter=10 | |
| ) | |
| confidential_style = ParagraphStyle( | |
| "Confidential", | |
| parent=styles['Normal'], | |
| fontName="Helvetica-Bold", | |
| fontSize=12, | |
| textColor=colors.HexColor("#e63946"), | |
| alignment=1, | |
| spaceBefore=100 | |
| ) | |
| # Body styles | |
| section_header_style = ParagraphStyle( | |
| "SectionHeader", | |
| parent=styles['Heading2'], | |
| fontName="Helvetica-Bold", | |
| fontSize=13, | |
| textColor=colors.HexColor("#e63946"), | |
| spaceBefore=15, | |
| spaceAfter=10 | |
| ) | |
| body_style = ParagraphStyle( | |
| "ReportBody", | |
| parent=styles['Normal'], | |
| fontName="Helvetica", | |
| fontSize=10, | |
| textColor=colors.HexColor("#1a1a2e"), | |
| leading=14 # line spacing 1.4 (10 * 1.4) | |
| ) | |
| story = [] | |
| # ------------------ | |
| # Page 1: Cover Page | |
| # ------------------ | |
| story.append(Spacer(1, 100)) | |
| story.append(Paragraph("AML COMPLIANCE MONITORING REPORT", title_style)) | |
| story.append(Paragraph("AML Shield — Powered by AI", subtitle_style)) | |
| story.append(HRFlowable(width="80%", thickness=2, color=colors.HexColor("#e63946"), spaceBefore=20, spaceAfter=20)) | |
| story.append(Paragraph(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", normal_center)) | |
| story.append(Paragraph(f"Dataset: {summary_data.get('filename', 'Unknown')}", normal_center)) | |
| story.append(Paragraph(f"Date Range: {summary_data.get('date_range', 'Unknown')}", normal_center)) | |
| story.append(Paragraph(f"Total Transactions: {summary_data.get('total_transactions', 0)}", normal_center)) | |
| story.append(Paragraph("CONFIDENTIAL — For Internal Use Only.<br/>This report contains sensitive AML analysis.", confidential_style)) | |
| story.append(PageBreak()) | |
| # ------------------ | |
| # Pages 2+: AI Report Body | |
| # ------------------ | |
| sections = [ | |
| "1. EXECUTIVE SUMMARY", | |
| "2. KEY FINDINGS", | |
| "3. HIGH RISK TRANSACTIONS", | |
| "4. CUSTOMER RISK ASSESSMENT (KYC)", | |
| "5. REGULATORY IMPLICATIONS", | |
| "6. RECOMMENDATIONS", | |
| "7. CONCLUSION" | |
| ] | |
| # Simple parser to split by sections | |
| # Find all instances of these headers in the text and format them | |
| text = ai_report_text | |
| # Just a basic approach: we find the positions of headers, and slice | |
| # However, text may not perfectly match these if AI varies slightly. | |
| # We will just split paragraphs and check if they start with a number + section name logic | |
| paragraphs = text.replace('\r\n', '\n').split('\n\n') | |
| for p in paragraphs: | |
| p = p.strip() | |
| if not p: continue | |
| # Check if it looks like a header (starts with number and is short, or matches our section list) | |
| is_header = False | |
| for sec in sections: | |
| if p.upper().startswith(sec) or p.startswith(f"**{sec}"): | |
| is_header = True | |
| p_clean = p.replace("**", "") # Remove markdown bold | |
| story.append(HRFlowable(width="100%", thickness=1, color=colors.lightgrey, spaceBefore=10, spaceAfter=5)) | |
| story.append(Paragraph(p_clean, section_header_style)) | |
| break | |
| if not is_header: | |
| # Clean up markdown specifics for body | |
| p_clean = p.replace("**", "") | |
| # Markdown bullet points to text representation | |
| if p_clean.startswith("- ") or p_clean.startswith("* "): | |
| p_clean = p_clean.replace("- ", r"• ", 1) | |
| p_clean = p_clean.replace("* ", r"• ", 1) | |
| story.append(Paragraph(p_clean, body_style)) | |
| story.append(Spacer(1, 10)) | |
| story.append(PageBreak()) | |
| # ------------------ | |
| # Final Page: Top 20 Flagged Transactions Table | |
| # ------------------ | |
| story.append(Paragraph("Top 20 Flagged Transactions", section_header_style)) | |
| story.append(Spacer(1, 15)) | |
| if not flagged_df.empty: | |
| top_20 = flagged_df.sort_values(by='risk_score', ascending=False).head(20) | |
| table_data = [] | |
| headers = ['Transaction ID', 'Customer', 'Amount', 'Type', 'Risk Score', 'Risk Level', 'Rules Triggered'] | |
| table_data.append(headers) | |
| for idx, row in top_20.iterrows(): | |
| rules = row['rule_flags'] | |
| rule_str = ", ".join(rules) if isinstance(rules, list) else str(rules) | |
| amount = f"${row['amount']:,.2f}" | |
| row_data = [ | |
| str(row['transaction_id']), | |
| str(row['customer_id']), | |
| amount, | |
| str(row['transaction_type']), | |
| f"{row['risk_score']:.1f}", | |
| str(row['risk_level']), | |
| rule_str | |
| ] | |
| table_data.append(row_data) | |
| # Table styling | |
| ts = TableStyle([ | |
| ('BACKGROUND', (0,0), (-1,0), colors.HexColor("#0a1628")), | |
| ('TEXTCOLOR', (0,0), (-1,0), colors.white), | |
| ('ALIGN', (0,0), (-1,-1), 'LEFT'), | |
| ('FONTNAME', (0,0), (-1,0), 'Helvetica-Bold'), | |
| ('FONTSIZE', (0,0), (-1,0), 10), | |
| ('BOTTOMPADDING', (0,0), (-1,0), 12), | |
| ('GRID', (0,0), (-1,-1), 1, colors.black), | |
| ('FONTNAME', (0,1), (-1,-1), 'Helvetica'), | |
| ('FONTSIZE', (0,1), (-1,-1), 8), | |
| ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), | |
| ]) | |
| # Add alternating row colors based on risk | |
| for i, row in enumerate(top_20.itertuples(), start=1): | |
| if row.risk_level == 'High': | |
| ts.add('BACKGROUND', (0,i), (-1,i), colors.HexColor("#ffe0e0")) | |
| elif row.risk_level == 'Medium': | |
| ts.add('BACKGROUND', (0,i), (-1,i), colors.HexColor("#fff9c4")) | |
| t = Table(table_data, repeatRows=1) | |
| t.setStyle(ts) | |
| story.append(t) | |
| else: | |
| story.append(Paragraph("No flagged transactions to display.", body_style)) | |
| doc.build(story) | |
| pdf_bytes = buffer.getvalue() | |
| buffer.close() | |
| return pdf_bytes | |