import os
from datetime import datetime
from typing import Optional
from reportlab.lib import colors
from reportlab.lib.pagesizes import letter
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from reportlab.platypus import (
SimpleDocTemplate,
Paragraph,
Spacer,
Table,
TableStyle,
Image,
Preformatted,
HRFlowable,
PageBreak,
)
def _build_styles():
styles = getSampleStyleSheet()
styles.add(
ParagraphStyle(
name="HeroTitle",
parent=styles["Title"],
fontName="Helvetica-Bold",
fontSize=20,
leading=26,
textColor=colors.HexColor("#111827"),
)
)
styles.add(
ParagraphStyle(
name="SectionHeader",
parent=styles["Heading2"],
fontName="Helvetica-Bold",
fontSize=14,
spaceAfter=6,
textColor=colors.HexColor("#111827"),
)
)
styles.add(
ParagraphStyle(
name="ReportBody",
parent=styles["BodyText"],
fontSize=11,
leading=16,
)
)
styles.add(
ParagraphStyle(
name="MetaLabel",
parent=styles["BodyText"],
fontSize=8,
textColor=colors.HexColor("#6b7280"),
leading=10,
)
)
styles.add(
ParagraphStyle(
name="QueryText",
parent=styles["BodyText"],
fontSize=12,
leading=16,
textColor=colors.HexColor("#111827"),
)
)
styles.add(
ParagraphStyle(
name="MetricValue",
parent=styles["BodyText"],
fontName="Helvetica-Bold",
fontSize=13,
textColor=colors.HexColor("#111827"),
)
)
styles.add(
ParagraphStyle(
name="MetricLabel",
parent=styles["BodyText"],
fontSize=9,
textColor=colors.HexColor("#6b7280"),
)
)
styles.add(
ParagraphStyle(
name="CaptionSmall",
parent=styles["BodyText"],
fontSize=9,
textColor=colors.HexColor("#6b7280"),
)
)
styles.add(
ParagraphStyle(
name="InsightBullet",
parent=styles["BodyText"],
leftIndent=14,
bulletIndent=6,
bulletFontName="Helvetica",
bulletFontSize=10,
leading=16,
)
)
styles.add(
ParagraphStyle(
name="CodeBlock",
parent=styles["BodyText"],
fontName="Courier",
fontSize=9,
leading=14,
backColor=colors.whitesmoke,
)
)
return styles
def _format_insights(insights: str, styles) -> list:
blocks = []
lines = [line.strip() for line in (insights or "").splitlines() if line.strip()]
if not lines:
return [Paragraph("No insights provided.", styles["ReportBody"])]
for line in lines:
if line[0] in {"-", "*", "•"}:
text = line.lstrip("-•* ").strip()
blocks.append(Paragraph(text, styles["InsightBullet"], bulletText="•"))
else:
blocks.append(Paragraph(line, styles["ReportBody"]))
blocks.append(Spacer(1, 4))
return blocks
def _build_data_table(data_sample):
if not data_sample:
return None
columns = list(data_sample[0].keys())
rows = [columns]
for row in data_sample[:10]:
rows.append([str(row.get(col, "")) for col in columns])
table = Table(rows, repeatRows=1)
table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#111827")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, 0), 10),
("BOTTOMPADDING", (0, 0), (-1, 0), 6),
("BACKGROUND", (0, 1), (-1, -1), colors.HexColor("#f9fafb")),
("TEXTCOLOR", (0, 1), (-1, -1), colors.HexColor("#111827")),
("GRID", (0, 0), (-1, -1), 0.25, colors.HexColor("#e5e7eb")),
]
)
)
return table
def _build_query_callout(query: str, styles):
display = query or "No question provided"
content = [
[Paragraph("BUSINESS QUESTION", styles["MetaLabel"])],
[Paragraph(display, styles["QueryText"])]
]
table = Table(content, colWidths=[6.5 * inch])
table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), colors.HexColor("#f3f4ff")),
("BOX", (0, 0), (-1, -1), 0.6, colors.HexColor("#c7d2fe")),
("INNERPADDING", (0, 0), (-1, -1), 8),
]
)
)
return table
def _build_metric_cards(stats: dict, styles):
if not stats:
return None
data = []
row = []
for label, value in stats.items():
cell = [
Paragraph(label, styles["MetricLabel"]),
Paragraph(value, styles["MetricValue"]),
]
row.append(cell)
data.append(row)
col_width = (6.5 * inch) / len(stats)
table = Table(data, colWidths=[col_width] * len(stats))
table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), colors.HexColor("#f9fafb")),
("BOX", (0, 0), (-1, -1), 0.5, colors.HexColor("#e5e7eb")),
("INNERPADDING", (0, 0), (-1, -1), 8),
("VALIGN", (0, 0), (-1, -1), "TOP"),
]
)
)
return table
def _format_value(value):
if value is None:
return "—"
if isinstance(value, float):
return f"{value:,.2f}"
if isinstance(value, int):
return f"{value:,}"
return str(value)
def _render_trend_section(trend_analysis, styles):
blocks = [Paragraph("Trend Diagnostics", styles["SectionHeader"])]
if not trend_analysis:
blocks.append(Paragraph("Not enough chronological data to compute a trend.", styles["ReportBody"]))
return blocks
blocks.append(Paragraph(trend_analysis.get("summary", ""), styles["ReportBody"]))
stats = [
["Starting value", _format_value(trend_analysis.get("start"))],
["Ending value", _format_value(trend_analysis.get("end"))],
["Slope", _format_value(trend_analysis.get("slope"))],
]
if trend_analysis.get("change_pct") is not None:
stats.append(["% Change", f"{trend_analysis['change_pct']:+.1f}%"])
table = Table(stats, colWidths=[2.3 * inch, 4 * inch])
table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, -1), colors.HexColor("#f9fafb")),
("BOX", (0, 0), (-1, -1), 0.4, colors.HexColor("#e5e7eb")),
("INNERPADDING", (0, 0), (-1, -1), 6),
("FONTNAME", (0, 0), (-1, -1), "Helvetica"),
]
)
)
blocks.extend([Spacer(1, 6), table])
return blocks
def _render_anomaly_section(anomaly_analysis, styles):
blocks = [Paragraph("Anomaly Highlights", styles["SectionHeader"])]
if not anomaly_analysis:
blocks.append(Paragraph("No statistical anomalies detected across the observed period.", styles["ReportBody"]))
return blocks
blocks.append(Paragraph(anomaly_analysis.get("summary", ""), styles["ReportBody"]))
table_rows = [["Period", "Value", "z-score"]]
for entry in anomaly_analysis.get("anomalies", [])[:8]:
table_rows.append([
entry.get("period", "?"),
_format_value(entry.get("value")),
f"{entry.get('z_score', 0):+.2f}",
])
table = Table(table_rows, repeatRows=1, colWidths=[2 * inch, 2.5 * inch, 1.5 * inch])
table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#111827")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("BACKGROUND", (0, 1), (-1, -1), colors.HexColor("#f3f4f6")),
("GRID", (0, 0), (-1, -1), 0.25, colors.HexColor("#d1d5db")),
]
)
)
blocks.extend([Spacer(1, 6), table])
return blocks
def _render_forecast_section(forecast_analysis, styles):
blocks = [Paragraph("Forecast", styles["SectionHeader"])]
if not forecast_analysis or not forecast_analysis.get("forecasts"):
blocks.append(Paragraph("Forecasting unavailable for this dataset.", styles["ReportBody"]))
return blocks
blocks.append(Paragraph(forecast_analysis.get("summary", ""), styles["ReportBody"]))
method = forecast_analysis.get("method", "N/A")
blocks.append(Spacer(1, 4))
blocks.append(Paragraph(f"Method: {method}", styles["CaptionSmall"]))
table_rows = [["Period", "Forecast", "Lower Bound", "Upper Bound"]]
for entry in forecast_analysis.get("forecasts", []):
table_rows.append([
entry.get("period", "?"),
_format_value(entry.get("value")),
_format_value(entry.get("lower_bound")),
_format_value(entry.get("upper_bound")),
])
table = Table(table_rows, repeatRows=1, colWidths=[1.5 * inch, 1.5 * inch, 1.5 * inch, 1.5 * inch])
table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#111827")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("BACKGROUND", (0, 1), (-1, -1), colors.HexColor("#f3f4f6")),
("GRID", (0, 0), (-1, -1), 0.25, colors.HexColor("#d1d5db")),
]
)
)
blocks.extend([Spacer(1, 6), table])
return blocks
def _render_statistical_tests_section(statistical_tests, styles):
blocks = [Paragraph("Statistical Tests", styles["SectionHeader"])]
if not statistical_tests or not statistical_tests.get("tests"):
blocks.append(Paragraph("No statistical tests performed.", styles["ReportBody"]))
return blocks
blocks.append(Paragraph(statistical_tests.get("summary", ""), styles["ReportBody"]))
blocks.append(Spacer(1, 8))
for test_name, test_data in statistical_tests.get("tests", {}).items():
test_title = test_data.get("test", test_name)
blocks.append(Paragraph(f"{test_title}", styles["ReportBody"]))
blocks.append(Spacer(1, 2))
details = []
if "comparison" in test_data:
details.append(f"Comparison: {test_data['comparison']}")
if "p_value" in test_data:
p_val = test_data["p_value"]
sig = "Yes" if test_data.get("significant", False) else "No"
details.append(f"p-value: {p_val:.4f} (Significant: {sig})")
if "summary" in test_data:
details.append(test_data["summary"])
for detail in details:
blocks.append(Paragraph(f"• {detail}", styles["ReportBody"]))
blocks.append(Spacer(1, 6))
return blocks
def generate_pdf_report(
report_path: str,
title: str,
query: str,
sql_query: str,
insights: str,
chart_image_path: str = None,
chart_summary: str = None,
trend_analysis: Optional[dict] = None,
anomaly_analysis: Optional[dict] = None,
forecast_analysis: Optional[dict] = None,
statistical_tests: Optional[dict] = None,
data_sample=None,
):
styles = _build_styles()
doc = SimpleDocTemplate(
report_path,
pagesize=letter,
leftMargin=0.75 * inch,
rightMargin=0.75 * inch,
topMargin=0.75 * inch,
bottomMargin=0.75 * inch,
)
story = []
generated_at = datetime.utcnow().strftime("%B %d, %Y %H:%M UTC")
row_count = len(data_sample or [])
column_count = len(data_sample[0]) if data_sample else 0
story.append(Paragraph(title, styles["HeroTitle"]))
story.append(Paragraph("Autonomous insight report", styles["CaptionSmall"]))
story.append(Spacer(1, 10))
story.append(_build_query_callout(query, styles))
story.append(Spacer(1, 12))
metrics = {
"Rows Returned": f"{row_count:,}" if row_count else "0",
"Columns": str(column_count),
"Generated": generated_at,
}
metric_table = _build_metric_cards(metrics, styles)
if metric_table:
story.append(metric_table)
story.append(Spacer(1, 18))
story.append(Paragraph("This auto-generated briefing captures the freshest SQL results, advanced analytics, and executive visuals from InsightPilot.", styles["ReportBody"]))
story.append(Spacer(1, 14))
story.append(PageBreak())
story.append(HRFlowable(width="100%", thickness=0.6, color=colors.HexColor("#e5e7eb")))
story.append(Spacer(1, 14))
story.append(Paragraph("SQL Used", styles["SectionHeader"]))
story.append(
Preformatted(sql_query.strip() or "No SQL generated", styles["CodeBlock"], maxLineLength=80)
)
story.append(Spacer(1, 12))
story.append(Paragraph("Insights", styles["SectionHeader"]))
for block in _format_insights(insights, styles):
story.append(block)
story.append(Spacer(1, 10))
table = _build_data_table(data_sample or [])
if table:
story.append(Paragraph("Data Preview", styles["SectionHeader"]))
story.append(table)
story.append(Spacer(1, 12))
for block in _render_trend_section(trend_analysis, styles):
story.append(block)
story.append(Spacer(1, 12))
for block in _render_anomaly_section(anomaly_analysis, styles):
story.append(block)
story.append(Spacer(1, 12))
for block in _render_forecast_section(forecast_analysis, styles):
story.append(block)
story.append(Spacer(1, 12))
for block in _render_statistical_tests_section(statistical_tests, styles):
story.append(block)
story.append(Spacer(1, 12))
if chart_image_path and os.path.exists(chart_image_path):
story.append(Paragraph("Visualization", styles["SectionHeader"]))
img = Image(chart_image_path)
img._restrictSize(6.5 * inch, 4.5 * inch)
img.hAlign = "CENTER"
story.append(img)
if chart_summary:
story.append(Spacer(1, 6))
story.append(Paragraph(chart_summary, styles["CaptionSmall"]))
elif chart_summary:
story.append(Paragraph("Visualization Summary", styles["SectionHeader"]))
story.append(Paragraph(chart_summary, styles["ReportBody"]))
doc.build(story)
return report_path