DuckDB End-to-End Automated Report Pipeline: From Data Source to Email Delivery

pip install duckdb openpyxl pandas

import duckdb
print(duckdb.__version__)  # Should print the version number

import duckdb
import pandas as pd
from pathlib import Path

# Connect to a persistent DuckDB database (data saved to disk)
con = duckdb.connect('sales_automation.duckdb')

# Batch-read all CSV files (auto-merge, no need to read individually)
con.execute("""
    CREATE TABLE all_sales AS
    SELECT * FROM read_csv_auto('sales_data/*.csv', hive_partitioning=true)
""")

# Preview the data structure
result = con.execute("DESCRIBE all_sales").fetchdf()
print(result)

monthly_summary = con.execute("""
    SELECT 
        DATE_TRUNC('month', sale_date) AS month,
        SUM(revenue) AS total_revenue,
        COUNT(DISTINCT customer_id) AS active_customers,
        AVG(order_value) AS avg_order_value,
        SUM(CASE WHEN region = 'East' THEN revenue ELSE 0 END) AS east_region_revenue
    FROM all_sales
    GROUP BY DATE_TRUNC('month', sale_date)
    ORDER BY month DESC
""").fetchdf()

print(monthly_summary)

top_products = con.execute("""
    SELECT 
        product_name,
        category,
        SUM(quantity) AS total_qty,
        SUM(revenue) AS total_revenue,
        RANK() OVER (ORDER BY SUM(revenue) DESC) AS revenue_rank
    FROM all_sales
    GROUP BY product_name, category
    HAVING SUM(revenue) > 1000
    ORDER BY total_revenue DESC
    LIMIT 20
""").fetchdf()

customer_segment = con.execute("""
    SELECT 
        customer_id,
        COUNT(*) AS order_count,
        SUM(revenue) AS total_spend,
        MAX(sale_date) AS last_purchase,
        CASE 
            WHEN SUM(revenue) > 10000 THEN 'VIP'
            WHEN SUM(revenue) > 5000 THEN 'High-Value'
            WHEN SUM(revenue) > 1000 THEN 'Regular'
            ELSE 'Potential'
        END AS segment
    FROM all_sales
    GROUP BY customer_id
    ORDER BY total_spend DESC
""").fetchdf()

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment

def generate_excel_report(df_dict, output_path='report.xlsx'):
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        for sheet_name, df in df_dict.items():
            df.to_excel(writer, sheet_name=sheet_name, index=False)
            
            # Style the header row
            ws = writer.sheets[sheet_name]
            header_fill = PatternFill(start_color='4472C4', end_color='4472C4', fill_type='solid')
            header_font = Font(bold=True, color='FFFFFF')
            
            for cell in ws[1]:
                cell.fill = header_fill
                cell.font = header_font
                cell.alignment = Alignment(horizontal='center')
            
            # Auto-fit column widths
            for column in ws.columns:
                max_length = 0
                column_letter = column[0].column_letter
                for cell in column:
                    try:
                        if len(str(cell.value)) > max_length:
                            max_length = len(str(cell.value))
                    except:
                        pass
                adjusted_width = min(max_length + 2, 50)
                ws.column_dimensions[column_letter].width = adjusted_width
    
    print(f'Report generated: {output_path}')

# Generate multi-sheet report
report_data = {
    'Monthly Summary': monthly_summary,
    'Top Products': top_products,
    'Customer Segments': customer_segment
}

generate_excel_report(report_data, 'weekly_report.xlsx')

def generate_text_report(df_dict, output_path='report.txt'):
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('=' * 60 + '\n')
        f.write('Weekly Data Report\n')
        f.write('=' * 60 + '\n\n')
        
        for sheet_name, df in df_dict.items():
            f.write(f'\n[{sheet_name}]\n')
            f.write(df.to_string(index=False))
            f.write('\n' + '-' * 60 + '\n')
    
    print(f'Text report generated: {output_path}')

import schedule
import time
from datetime import datetime

def run_weekly_report():
    print(f'{datetime.now()} - Starting weekly report...')
    
    # Reload latest data
    con.execute("DROP TABLE IF EXISTS all_sales")
    con.execute("""
        CREATE TABLE all_sales AS
        SELECT * FROM read_csv_auto('sales_data/*.csv', hive_partitioning=true)
    """)
    
    # Execute analysis queries
    monthly_summary = con.execute("""
        SELECT DATE_TRUNC('month', sale_date) AS month,
               SUM(revenue) AS total_revenue, COUNT(DISTINCT customer_id) AS active_customers
        FROM all_sales GROUP BY DATE_TRUNC('month', sale_date)
        ORDER BY month DESC
    """).fetchdf()
    
    # Output report
    report_data = {'Monthly Summary': monthly_summary}
    generate_excel_report(report_data, f'report_{datetime.now().strftime("%Y%m%d")}.xlsx')
    print('Weekly report generated!')

# Run every Monday at 9:00 AM
schedule.every().monday.at("09:00").do(run_weekly_report)

while True:
    schedule.run_pending()
    time.sleep(60)

crontab -e

name: Weekly Report
on:
  schedule:
    - cron: '0 9 * * 1'  # Monday 9:00 UTC
  workflow_dispatch:

jobs:
  build-report:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install duckdb openpyxl pandas
      - name: Generate report
        run: python3 weekly_report.py
      - name: Upload artifact
        uses: actions/upload-artifact@v3
        with:
          name: weekly-report
          path: report_*.xlsx

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

def send_email_report(to_email, report_path, subject='Weekly Report'):
    msg = MIMEMultipart()
    msg['From'] = '[email protected]'
    msg['To'] = to_email
    msg['Subject'] = subject
    
    body = 'Please find this week\'s data report attached.'
    msg.attach(MIMEText(body, 'plain', 'utf-8'))
    
    with open(report_path, 'rb') as f:
        attachment = MIMEBase('application', 'octet-stream')
        attachment.set_payload(f.read())
        encoders.encode_base64(attachment)
        attachment.add_header('Content-Disposition', f'attachment; filename={report_path}')
        msg.attach(attachment)
    
    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('[email protected]', 'your_password')
        server.send_message(msg)
    
    print(f'Email sent to {to_email}')