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}')