📈 Reporting Automation: Create Dynamic Reports Automatically

Reporting automation transforms the tedious, error-prone process of manual report creation into a streamlined, accurate, and timely delivery of business intelligence - it's the difference between spending days compiling data and having insights delivered to your inbox every morning. Like having a team of analysts working 24/7, automated reporting systems gather data from multiple sources, perform complex calculations, generate visualizations, and distribute polished reports to stakeholders automatically. Whether you're creating financial reports, operational dashboards, compliance documentation, or executive briefings, mastering reporting automation is essential for data-driven decision making. Let's explore the comprehensive world of automated reporting! 📊

The Reporting Automation Architecture

Think of reporting automation as creating an intelligent information factory - raw data flows in from various sources, gets processed, analyzed, and transformed into meaningful insights, then packaged into beautiful reports and delivered to the right people at the right time. Using data pipelines, visualization libraries, template engines, and distribution systems, modern reporting automation handles everything from simple metrics to complex analytical reports. Understanding data integration, report design, and delivery mechanisms is crucial for successful implementation!

Real-World Scenario: The Enterprise Reporting Platform 🎯

You're building a comprehensive reporting platform for a global enterprise that generates 500+ reports daily across departments, pulls data from 20+ sources including databases, APIs, and files, creates financial reports requiring penny-perfect accuracy, produces operational dashboards with real-time metrics, generates regulatory compliance reports for multiple jurisdictions, delivers personalized reports to 1,000+ stakeholders, supports multiple formats (PDF, Excel, HTML, PowerBI), schedules reports hourly, daily, weekly, and monthly, and provides self-service reporting capabilities. Your platform must handle large data volumes, ensure data accuracy, maintain performance under load, and scale with business growth. Let's build a comprehensive reporting automation framework!

# Comprehensive Reporting Automation Framework
# pip install pandas numpy matplotlib seaborn plotly
# pip install jinja2 weasyprint pdfkit reportlab
# pip install openpyxl xlsxwriter python-docx
# pip install sqlalchemy pymongo requests
# pip install schedule croniter celery
# pip install streamlit dash bokeh

import os
import json
import logging
from typing import Dict, List, Any, Optional, Union, Tuple
from dataclasses import dataclass, field, asdict
from datetime import datetime, date, timedelta
from pathlib import Path
from enum import Enum, auto
import hashlib
import tempfile
import base64
import io

# Data processing
import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text

# Visualization
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots

# Report generation
from jinja2 import Environment, FileSystemLoader, Template
from reportlab.lib import colors
from reportlab.lib.pagesizes import letter, A4
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, Image, PageBreak
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
import pdfkit

# Excel generation
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font, Fill, PatternFill, Border, Side, Alignment
from openpyxl.chart import BarChart, PieChart, LineChart, Reference
from openpyxl.utils import get_column_letter
import xlsxwriter

# Document generation
from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH

# Email
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

# Scheduling
import schedule
from croniter import croniter

# ==================== Report Models ====================

class ReportType(Enum):
    """Types of reports."""
    FINANCIAL = auto()
    OPERATIONAL = auto()
    EXECUTIVE = auto()
    COMPLIANCE = auto()
    ANALYTICAL = auto()
    DASHBOARD = auto()
    AD_HOC = auto()

class ReportFormat(Enum):
    """Report output formats."""
    PDF = auto()
    EXCEL = auto()
    HTML = auto()
    WORD = auto()
    CSV = auto()
    JSON = auto()
    POWERBI = auto()
    TABLEAU = auto()

class DeliveryMethod(Enum):
    """Report delivery methods."""
    EMAIL = auto()
    FTP = auto()
    API = auto()
    S3 = auto()
    SHAREPOINT = auto()
    WEB_PORTAL = auto()
    SLACK = auto()

@dataclass
class ReportDefinition:
    """Definition of a report."""
    id: str
    name: str
    description: str
    type: ReportType
    format: ReportFormat
    
    # Data configuration
    data_sources: List[Dict[str, Any]]
    parameters: Dict[str, Any] = field(default_factory=dict)
    filters: Dict[str, Any] = field(default_factory=dict)
    
    # Schedule
    schedule: Optional[str] = None  # Cron expression
    
    # Distribution
    recipients: List[str] = field(default_factory=list)
    delivery_method: DeliveryMethod = DeliveryMethod.EMAIL
    
    # Template
    template: Optional[str] = None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary."""
        data = asdict(self)
        data['type'] = self.type.name
        data['format'] = self.format.name
        data['delivery_method'] = self.delivery_method.name
        return data

@dataclass
class ReportInstance:
    """Instance of a generated report."""
    id: str
    definition_id: str
    generated_at: datetime
    file_path: Optional[str] = None
    status: str = "pending"
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

# ==================== Data Connectors ====================

class DataConnector:
    """Base class for data connectors."""
    
    def connect(self, config: Dict[str, Any]):
        """Connect to data source."""
        raise NotImplementedError
    
    def fetch_data(self, query: str, parameters: Optional[Dict] = None) -> pd.DataFrame:
        """Fetch data from source."""
        raise NotImplementedError
    
    def close(self):
        """Close connection."""
        pass

class SQLConnector(DataConnector):
    """SQL database connector."""
    
    def __init__(self):
        self.engine = None
        self.logger = logging.getLogger(__name__)
    
    def connect(self, config: Dict[str, Any]):
        """Connect to SQL database."""
        connection_string = config.get('connection_string')
        if not connection_string:
            # Build connection string
            db_type = config.get('type', 'postgresql')
            host = config.get('host', 'localhost')
            port = config.get('port', 5432)
            database = config.get('database')
            username = config.get('username')
            password = config.get('password')
            
            connection_string = f"{db_type}://{username}:{password}@{host}:{port}/{database}"
        
        self.engine = create_engine(connection_string)
        self.logger.info(f"Connected to database")
    
    def fetch_data(self, query: str, parameters: Optional[Dict] = None) -> pd.DataFrame:
        """Execute SQL query and return DataFrame."""
        if not self.engine:
            raise ConnectionError("Not connected to database")
        
        return pd.read_sql(query, self.engine, params=parameters)
    
    def close(self):
        """Close database connection."""
        if self.engine:
            self.engine.dispose()

class APIConnector(DataConnector):
    """REST API connector."""
    
    def __init__(self):
        self.base_url = None
        self.headers = {}
        self.session = None
        
    def connect(self, config: Dict[str, Any]):
        """Connect to API."""
        import requests
        
        self.base_url = config.get('base_url')
        self.headers = config.get('headers', {})
        
        # Authentication
        auth_type = config.get('auth_type')
        if auth_type == 'bearer':
            self.headers['Authorization'] = f"Bearer {config.get('token')}"
        elif auth_type == 'api_key':
            self.headers['X-API-Key'] = config.get('api_key')
        
        self.session = requests.Session()
        self.session.headers.update(self.headers)
    
    def fetch_data(self, endpoint: str, parameters: Optional[Dict] = None) -> pd.DataFrame:
        """Fetch data from API."""
        if not self.session:
            raise ConnectionError("Not connected to API")
        
        url = f"{self.base_url}/{endpoint}"
        response = self.session.get(url, params=parameters)
        response.raise_for_status()
        
        data = response.json()
        
        # Convert to DataFrame
        if isinstance(data, list):
            return pd.DataFrame(data)
        elif isinstance(data, dict):
            if 'data' in data:
                return pd.DataFrame(data['data'])
            else:
                return pd.DataFrame([data])
        
        return pd.DataFrame()

class FileConnector(DataConnector):
    """File-based data connector."""
    
    def __init__(self):
        self.base_path = None
    
    def connect(self, config: Dict[str, Any]):
        """Setup file connector."""
        self.base_path = Path(config.get('base_path', '.'))
    
    def fetch_data(self, file_path: str, parameters: Optional[Dict] = None) -> pd.DataFrame:
        """Read data from file."""
        full_path = self.base_path / file_path
        
        if file_path.endswith('.csv'):
            return pd.read_csv(full_path, **parameters if parameters else {})
        elif file_path.endswith(('.xlsx', '.xls')):
            return pd.read_excel(full_path, **parameters if parameters else {})
        elif file_path.endswith('.json'):
            return pd.read_json(full_path, **parameters if parameters else {})
        elif file_path.endswith('.parquet'):
            return pd.read_parquet(full_path, **parameters if parameters else {})
        
        raise ValueError(f"Unsupported file type: {file_path}")

# ==================== Report Generators ====================

class ReportGenerator:
    """Base report generator."""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def generate(self, data: pd.DataFrame, definition: ReportDefinition) -> str:
        """Generate report and return file path."""
        raise NotImplementedError

class PDFReportGenerator(ReportGenerator):
    """Generate PDF reports."""
    
    def generate(self, data: pd.DataFrame, definition: ReportDefinition) -> str:
        """Generate PDF report."""
        # Create temporary file
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.pdf')
        filename = temp_file.name
        temp_file.close()
        
        # Create PDF document
        doc = SimpleDocTemplate(filename, pagesize=letter)
        story = []
        styles = getSampleStyleSheet()
        
        # Add title
        title = Paragraph(definition.name, styles['Title'])
        story.append(title)
        story.append(Spacer(1, 12))
        
        # Add description
        description = Paragraph(definition.description, styles['Normal'])
        story.append(description)
        story.append(Spacer(1, 12))
        
        # Add generated timestamp
        timestamp = Paragraph(
            f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            styles['Normal']
        )
        story.append(timestamp)
        story.append(Spacer(1, 24))
        
        # Add summary statistics
        if not data.empty:
            summary = self._create_summary_table(data)
            story.append(summary)
            story.append(Spacer(1, 12))
            
            # Add data table
            data_table = self._create_data_table(data.head(50))  # Limit rows for PDF
            story.append(data_table)
            
            # Add charts
            story.append(PageBreak())
            charts = self._create_charts(data, filename)
            for chart in charts:
                story.append(chart)
                story.append(Spacer(1, 12))
        
        # Build PDF
        doc.build(story)
        
        self.logger.info(f"Generated PDF report: {filename}")
        return filename
    
    def _create_summary_table(self, data: pd.DataFrame) -> Table:
        """Create summary statistics table."""
        summary_data = [['Metric', 'Value']]
        summary_data.append(['Total Records', f"{len(data):,}"])
        summary_data.append(['Columns', str(len(data.columns))])
        
        # Add numeric column statistics
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        for col in numeric_cols[:5]:  # Limit to 5 columns
            summary_data.append([f"{col} (avg)", f"{data[col].mean():.2f}"])
        
        table = Table(summary_data)
        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, 0), 14),
            ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
            ('BACKGROUND', (0, 1), (-1, -1), colors.beige),
            ('GRID', (0, 0), (-1, -1), 1, colors.black),
        ]))
        
        return table
    
    def _create_data_table(self, data: pd.DataFrame) -> Table:
        """Create data table."""
        # Prepare table data
        table_data = [data.columns.tolist()]
        for _, row in data.iterrows():
            table_data.append([str(val) for val in row.values])
        
        table = Table(table_data)
        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, 0), 10),
            ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
            ('BACKGROUND', (0, 1), (-1, -1), colors.beige),
            ('GRID', (0, 0), (-1, -1), 1, colors.black),
            ('FONTSIZE', (0, 1), (-1, -1), 8),
        ]))
        
        return table
    
    def _create_charts(self, data: pd.DataFrame, base_filename: str) -> List[Image]:
        """Create charts for the report."""
        charts = []
        
        # Create bar chart for numeric columns
        numeric_cols = data.select_dtypes(include=[np.number]).columns[:3]
        if len(numeric_cols) > 0:
            fig, ax = plt.subplots(figsize=(8, 4))
            data[numeric_cols].mean().plot(kind='bar', ax=ax)
            ax.set_title('Average Values')
            ax.set_ylabel('Value')
            
            chart_path = base_filename.replace('.pdf', '_bar.png')
            plt.savefig(chart_path, dpi=100, bbox_inches='tight')
            plt.close()
            
            charts.append(Image(chart_path, width=6*inch, height=3*inch))
        
        return charts

class ExcelReportGenerator(ReportGenerator):
    """Generate Excel reports."""
    
    def generate(self, data: pd.DataFrame, definition: ReportDefinition) -> str:
        """Generate Excel report."""
        # Create temporary file
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.xlsx')
        filename = temp_file.name
        temp_file.close()
        
        # Create Excel writer
        with pd.ExcelWriter(filename, engine='xlsxwriter') as writer:
            # Write main data
            data.to_excel(writer, sheet_name='Data', index=False)
            
            # Get workbook and worksheet
            workbook = writer.book
            worksheet = writer.sheets['Data']
            
            # Format header
            header_format = workbook.add_format({
                'bold': True,
                'text_wrap': True,
                'valign': 'top',
                'fg_color': '#D7E4BD',
                'border': 1
            })
            
            # Write header with formatting
            for col_num, value in enumerate(data.columns.values):
                worksheet.write(0, col_num, value, header_format)
            
            # Auto-fit columns
            for i, col in enumerate(data.columns):
                column_width = max(data[col].astype(str).str.len().max(), len(col)) + 2
                worksheet.set_column(i, i, min(column_width, 50))
            
            # Add summary sheet
            summary_df = self._create_summary_df(data)
            summary_df.to_excel(writer, sheet_name='Summary', index=False)
            
            # Add chart sheet
            self._add_charts(writer, data)
            
            # Add metadata sheet
            metadata_df = pd.DataFrame([{
                'Report': definition.name,
                'Generated': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'Records': len(data),
                'Type': definition.type.name
            }])
            metadata_df.to_excel(writer, sheet_name='Metadata', index=False)
        
        self.logger.info(f"Generated Excel report: {filename}")
        return filename
    
    def _create_summary_df(self, data: pd.DataFrame) -> pd.DataFrame:
        """Create summary DataFrame."""
        summary = []
        
        # Basic statistics
        summary.append({
            'Metric': 'Total Records',
            'Value': len(data)
        })
        
        summary.append({
            'Metric': 'Total Columns',
            'Value': len(data.columns)
        })
        
        # Numeric column statistics
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            summary.extend([
                {'Metric': f'{col} - Mean', 'Value': data[col].mean()},
                {'Metric': f'{col} - Min', 'Value': data[col].min()},
                {'Metric': f'{col} - Max', 'Value': data[col].max()},
                {'Metric': f'{col} - Std Dev', 'Value': data[col].std()}
            ])
        
        return pd.DataFrame(summary)
    
    def _add_charts(self, writer, data: pd.DataFrame):
        """Add charts to Excel."""
        workbook = writer.book
        
        # Create charts worksheet
        chart_sheet = workbook.add_worksheet('Charts')
        
        # Create a column chart
        chart = workbook.add_chart({'type': 'column'})
        
        # Configure chart (simplified example)
        chart.set_title({'name': 'Data Analysis'})
        chart.set_x_axis({'name': 'Categories'})
        chart.set_y_axis({'name': 'Values'})
        
        # Insert chart
        chart_sheet.insert_chart('B2', chart)

class HTMLReportGenerator(ReportGenerator):
    """Generate HTML reports."""
    
    def __init__(self):
        super().__init__()
        self.template_env = Environment(loader=FileSystemLoader('templates'))
    
    def generate(self, data: pd.DataFrame, definition: ReportDefinition) -> str:
        """Generate HTML report."""
        # Create temporary file
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.html')
        filename = temp_file.name
        temp_file.close()
        
        # Generate visualizations
        charts = self._create_interactive_charts(data)
        
        # Prepare template data
        template_data = {
            'title': definition.name,
            'description': definition.description,
            'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'data_table': data.to_html(classes='table table-striped', index=False),
            'summary': self._create_summary_html(data),
            'charts': charts,
            'row_count': len(data),
            'column_count': len(data.columns)
        }
        
        # Render template
        if definition.template:
            template = self.template_env.get_template(definition.template)
        else:
            template = self._get_default_template()
        
        html = template.render(**template_data)
        
        # Save to file
        with open(filename, 'w') as f:
            f.write(html)
        
        self.logger.info(f"Generated HTML report: {filename}")
        return filename
    
    def _create_interactive_charts(self, data: pd.DataFrame) -> List[str]:
        """Create interactive Plotly charts."""
        charts = []
        
        # Create bar chart for numeric columns
        numeric_cols = data.select_dtypes(include=[np.number]).columns[:5]
        if len(numeric_cols) > 0:
            fig = go.Figure()
            
            for col in numeric_cols:
                fig.add_trace(go.Bar(
                    name=col,
                    x=['Mean', 'Min', 'Max'],
                    y=[data[col].mean(), data[col].min(), data[col].max()]
                ))
            
            fig.update_layout(
                title='Statistical Summary',
                barmode='group',
                template='plotly_white'
            )
            
            charts.append(fig.to_html(full_html=False, include_plotlyjs='cdn'))
        
        # Create time series if date column exists
        date_cols = data.select_dtypes(include=['datetime64']).columns
        if len(date_cols) > 0 and len(numeric_cols) > 0:
            fig = go.Figure()
            
            for col in numeric_cols[:3]:
                fig.add_trace(go.Scatter(
                    x=data[date_cols[0]],
                    y=data[col],
                    mode='lines',
                    name=col
                ))
            
            fig.update_layout(
                title='Time Series Analysis',
                xaxis_title='Date',
                yaxis_title='Value',
                template='plotly_white'
            )
            
            charts.append(fig.to_html(full_html=False, include_plotlyjs='cdn'))
        
        return charts
    
    def _create_summary_html(self, data: pd.DataFrame) -> str:
        """Create HTML summary."""
        summary = "
"
        summary += "
Data Summary
"
        summary += f"
Total Records: {len(data):,}
"
        summary += f"
Columns: {', '.join(data.columns)}
"
        
        # Add statistics for numeric columns
        numeric_cols = data.select_dtypes(include=[np.number]).columns
        if len(numeric_cols) > 0:
            summary += "
Numeric Column Statistics
"
            summary += ""
            summary += ""
            summary += ""
            
            for col in numeric_cols:
                summary += f""
                summary += f""
                summary += f""
                summary += f""
                summary += f""
                summary += f""
                summary += f""
            
            summary += "
Column
Mean
Min
Max
Std Dev
{col}
{data[col].mean():.2f}
{data[col].min():.2f}
{data[col].max():.2f}
{data[col].std():.2f}
"
        
        summary += "
"
        return summary
    
    def _get_default_template(self) -> Template:
        """Get default HTML template."""
        template_str = """
        


        
        
            
            
            
        
        
            

                

                    
{{ title }}
                    
{{ description }}
                    
Generated: {{ generated_at }}
                
                
                {{ summary | safe }}
                
                {% for chart in charts %}
                

                    {{ chart | safe }}
                
                {% endfor %}
                
                

                    
Data Table ({{ row_count }} records)
                    {{ data_table | safe }}