Capstone: end-to-end pipeline — CSV folder → clean → aggregate → formatted xlsx → email → schedule

 inbox/*.csv ─► load ─► clean ─► aggregate ─► formatted .xlsx ─► email ─► archive
                         │                                          │
                         └─► Rejected Rows sheet                    └─► logs/pipeline.log
                                    (runs daily via Task Scheduler / cron)

This brings together Module 1 (choosing the tool), Module 5 (pandas + openpyxl) and Module 6 (email + scheduling).

1. Why Python for this job?

Run the Module 1 decision tree: many files, must run with nobody opening Excel, emails a file, needs logs → Python. Power Query could combine the CSVs, but not email or run unattended; VBA needs Excel open and a logged-in user.

2. Project structure

sales-pipeline/
    .venv/                      ← virtual environment
    inbox/                      ← branches drop CSVs here
    reports/                    ← generated reports
    archive/                    ← processed CSVs, one folder per run
    logs/pipeline.log           ← created automatically
    monthly_report.py           ← from Python for Excel, Lesson 6 (load, clean, summarise, write, format)
    send_mail.py                ← from Scheduling & CI, Lesson 2
    pipeline.py                 ← the orchestrator (this lesson)
    run_pipeline.bat            ← Windows launcher for Task Scheduler
    requirements.txt            ← pandas, openpyxl

Setup:

cd sales-pipeline
python -m venv .venv
# Windows: .venv\Scripts\activate     Mac/Linux: source .venv/bin/activate
pip install pandas openpyxl
mkdir inbox reports archive

3. Input contract

Every CSV must have these columns: Date, Salesperson, Region, Product, Amount, with dates as dd/mm/yyyy. Write this down and share it with the branches — most pipeline failures are input surprises. The cleaning step trims spaces, fixes capitalisation, removes duplicates and sends bad dates/amounts to a Rejected Rows sheet instead of failing.

4. The orchestrator: pipeline.py

monthly_report.py and send_mail.py do the work; pipeline.py runs the steps in order, adds a chart, decides whether to email, archives inputs and logs everything.

"""
pipeline.py — ExcelWalaa Automation capstone
CSV folder -> clean -> aggregate -> formatted xlsx -> email -> archive (run it on a schedule)

Usage:
    python pipeline.py --inbox inbox --outdir reports --archive archive --to owner@example.com
    python pipeline.py --inbox inbox --outdir reports --archive archive --no-email   # for testing
"""
import argparse
import logging
import os
import shutil
import sys
from datetime import datetime
from pathlib import Path
from uuid import uuid4

from openpyxl import load_workbook
from openpyxl.chart import BarChart, Reference

from monthly_report import load_data, clean, summarise, write_report, format_report
from send_mail import send_file

log = logging.getLogger("pipeline")


def setup_logging(log_dir: Path) -> None:
    log_dir.mkdir(parents=True, exist_ok=True)
    fmt = "%(asctime)s | %(levelname)-7s | %(message)s"
    logging.basicConfig(level=logging.INFO, format=fmt,
                        handlers=[logging.FileHandler(log_dir / "pipeline.log", encoding="utf-8"),
                                  logging.StreamHandler(sys.stdout)])


def add_region_chart(path: Path) -> None:
    wb = load_workbook(path)
    ws = wb["By Region"]
    last = ws.max_row
    chart = BarChart()
    chart.title = "Sales by Region"
    chart.add_data(Reference(ws, min_col=2, min_row=1, max_row=last), titles_from_data=True)
    chart.set_categories(Reference(ws, min_col=1, min_row=2, max_row=last))
    chart.width, chart.height = 16, 8
    ws.add_chart(chart, "D2")
    wb.save(path)


def archive_files(files: list[Path], archive_root: Path, stamp: str) -> Path:
    target = archive_root / stamp
    target.mkdir(parents=True, exist_ok=False)
    for f in files:
        shutil.move(str(f), target / f.name)
    return target


def run(args: argparse.Namespace) -> int:
    inbox, outdir, archive = Path(args.inbox), Path(args.outdir), Path(args.archive)
    stamp = datetime.now().strftime("%Y-%m-%d_%H%M%S_%f") + "_" + uuid4().hex[:8]
    if not inbox.is_dir():
        raise FileNotFoundError(f"Inbox does not exist: {inbox}")

    files = sorted(inbox.glob("*.csv"))
    if not files:
        log.info("Inbox %s is empty - nothing to do.", inbox)
        return 0
    log.info("Found %d CSV file(s): %s", len(files), ", ".join(f.name for f in files))

    if not args.no_email and not (args.to and os.environ.get("SMTP_USER") and os.environ.get("SMTP_PASSWORD")):
        raise ValueError("Email requires --to, SMTP_USER and SMTP_PASSWORD; use --no-email for a local run")

    # Process exactly the snapshot that will be archived; later arrivals wait for the next run.
    raw = load_data(inbox, files=files)
    good, rejected = clean(raw)
    summaries = summarise(good)
    log.info("Rows read: %d | clean: %d | rejected: %d", len(raw), len(good), len(rejected))
    if good.empty:
        log.error("No valid rows - report not created. Check the Rejected rows in the source files.")
        return 1

    # 4: formatted workbook
    outdir.mkdir(parents=True, exist_ok=True)
    report = outdir / f"Sales_Report_{stamp}.xlsx"
    write_report(good, rejected, summaries, report)
    format_report(report)
    add_region_chart(report)
    log.info("Report saved: %s", report.resolve())

    # 5: email
    if args.no_email:
        log.info("Email skipped (--no-email).")
    else:
        total = good["Amount"].sum()
        body = (f"Hello,\n\nAttached is the sales report ({stamp}).\n"
                f"Clean rows: {len(good)} | Rejected rows: {len(rejected)}\n"
                f"Total sales: Rs {total:,.0f}\n\nThis email was sent automatically.")
        send_file(report, args.to, f"Sales report {stamp}", body)
        log.info("Email sent to %s", args.to)

    # 6: archive processed inputs so the next run starts clean
    moved_to = archive_files(files, archive, stamp)
    log.info("Inputs archived to %s", moved_to)
    return 0


def main() -> None:
    p = argparse.ArgumentParser(description="CSV -> Excel report -> email pipeline")
    p.add_argument("--inbox", default="inbox")
    p.add_argument("--outdir", default="reports")
    p.add_argument("--archive", default="archive")
    p.add_argument("--to", default=os.environ.get("REPORT_TO"))
    p.add_argument("--no-email", action="store_true")
    args = p.parse_args()

    setup_logging(Path("logs"))
    try:
        sys.exit(run(args))
    except Exception:
        log.exception("Pipeline failed")
        sys.exit(1)


if __name__ == "__main__":
    main()

5. Design decisions (and why)

6. Test it step by step

a) Dry run without email. Copy the three sample CSVs from Module 5 Lesson 6 into inbox/ and run:

python pipeline.py --no-email

Expected log:

INFO    | Found 3 CSV file(s): north.csv, south.csv, west.csv
INFO    | Rows read: 10 | clean: 7 | rejected: 2
INFO    | Report saved: .../reports/Sales_Report_2026-10-01_0900.xlsx
INFO    | Email skipped (--no-email).
INFO    | Inputs archived to archive/2026-10-01_0900

Open the report: By Region should show South 96,000, North 92,000, West 61,000 with a bar chart; Rejected Rows should show the "bad date" row and the "NA" amount row.

b) Run again immediately. The inbox is now empty → "nothing to do", exit code 0.

c) Break it on purpose. Put a CSV with only bad rows in the inbox → "No valid rows", exit code 1, file stays in the inbox.

d) Test the email. Set the environment variables (use a Gmail App Password) and send to yourself:

# Windows (PowerShell)
$env:SMTP_USER="reports@yourshop.com"; $env:SMTP_PASSWORD="app-password"
python pipeline.py --to you@example.com
# Mac/Linux
export SMTP_USER=reports@yourshop.com SMTP_PASSWORD=app-password
python pipeline.py --to you@example.com

7. Schedule it

Windows — run_pipeline.bat:

@echo off
cd /d "%~dp0"
if not exist logs mkdir logs
rem Set SMTP_USER, SMTP_PASSWORD and REPORT_TO in the scheduler account environment.
".venv\Scripts\python.exe" pipeline.py %* >> logs\scheduler.log 2>&1
exit /b %errorlevel%

Store SMTP_PASSWORD as a user environment variable (Start → "Edit environment variables for your account") rather than in the .bat file. Then create a Task Scheduler task (Module 6 Lesson 1): daily 9:00 AM, action = the .bat, Start in = the project folder, "Run task as soon as possible after a scheduled start is missed".

Mac/Linux — cron:

0 9 * * 1-6 cd /home/divyesh/sales-pipeline && SMTP_USER=reports@yourshop.com SMTP_PASSWORD='app-password' REPORT_TO=owner@yourshop.com .venv/bin/python pipeline.py >> logs/scheduler.log 2>&1

(The crontab is private to your user account, but a password file readable only by you is even better.)

GitHub Actions? It works for the report and email, but files moved into archive/ on GitHub's machine disappear after the run. Use it when the inputs come from a download step (API, cloud storage), and skip the archive step — or commit the archive back to the repo.

8. Final checklist

• Runs from a fresh terminal with the exact scheduled command
• Uses the venv Python and full paths in the scheduler
• Empty inbox → exit 0; bad data → Rejected Rows; all-bad → exit 1
• Report has 5 sheets, formatted headers, number/date formats and a chart
• Email arrives with the attachment and correct totals
• Processed CSVs move to archive/<timestamp>/
• logs/pipeline.log explains every run
• No passwords in any code file
• A one-page README: what it does, input format, how to run, who to call

9. Extensions (pick one or two)

• Failure alert: in main()'s except block, email the log tail to yourself.
• Month-over-month: load last month's report and add a % change column to By Region.
• Config file: move paths, recipients and the reorder threshold into config.json.
• More outputs: one file per region (Module 5 Lesson 3) attached to each regional manager's email.
• Google Sheets source: fetch the data with the Sheets API instead of CSVs — or flip it around and do the whole flow in Apps Script (Module 4).
• Tests: a small pytest file that feeds clean() messy rows and checks the result.

Common mistakes

Testing only with perfect data. Archiving before the report is safely written. Hard-coding the month or the recipient. Scheduling before the manual run is reliable. No README, so nobody else can run it.