Capstone: end-to-end pipeline — CSV फ़ोल्डर → साफ़ करना → जोड़ना → फ़ॉर्मेटेड xlsx → ईमेल → शेड्यूल

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

यह मॉड्यूल 1 (टूल चुनना), मॉड्यूल 5 (pandas + openpyxl) और मॉड्यूल 6 (ईमेल + शेड्यूलिंग) को एक साथ लाता है।

1. इस काम के लिए Python क्यों?

मॉड्यूल 1 का डिसीज़न ट्री चलाएँ: बहुत-सी फ़ाइलें, बिना किसी के Excel खोले चलना है, फ़ाइल ईमेल करनी है, लॉग चाहिए → Python। Power Query CSV जोड़ सकता है, पर ईमेल नहीं कर सकता और बिना निगरानी नहीं चल सकता; VBA को Excel खुला और लॉग-इन यूज़र चाहिए।

2. प्रोजेक्ट की बनावट

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

सेटअप:

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. इनपुट का नियम

हर CSV में ये कॉलम होने चाहिए: Date, Salesperson, Region, Product, Amount, और तारीख़ें dd/mm/yyyy में। इसे लिखकर ब्रांच के साथ शेयर करें — ज़्यादातर पाइपलाइन फ़ेलियर इनपुट के अचानक बदलाव से होते हैं। सफ़ाई वाला स्टेप स्पेस हटाता है, कैपिटलाइज़ेशन ठीक करता है, डुप्लिकेट हटाता है और गलत तारीख़/रकम को फ़ेल होने की जगह Rejected Rows शीट में भेजता है।

4. ऑर्केस्ट्रेटर: pipeline.py

monthly_report.py और send_mail.py काम करते हैं; pipeline.py स्टेप क्रम से चलाता है, चार्ट जोड़ता है, तय करता है कि ईमेल करना है या नहीं, इनपुट archive करता है और सब लॉग करता है।

"""
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. डिज़ाइन के फ़ैसले (और क्यों)

6. कदम-दर-कदम जाँचें

a) बिना ईमेल ड्राई रन। मॉड्यूल 5 लेसन 6 की तीनों सैंपल CSV inbox/ में कॉपी करें और चलाएँ:

python pipeline.py --no-email

अपेक्षित लॉग:

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

रिपोर्ट खोलें: By Region में बार चार्ट के साथ South 96,000, North 92,000, West 61,000 दिखना चाहिए; Rejected Rows में "bad date" वाली रो और "NA" रकम वाली रो दिखनी चाहिए।

b) तुरंत दोबारा चलाएँ। अब inbox खाली है → "nothing to do", exit code 0।

c) जानबूझकर तोड़ें। inbox में सिर्फ़ गलत रो वाली CSV रखें → "No valid rows", exit code 1, फ़ाइल inbox में ही रहती है।

d) ईमेल जाँचें। environment variables सेट करें (Gmail App Password इस्तेमाल करें) और अपने आप को भेजें:

# 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. शेड्यूल करें

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%

SMTP_PASSWORD को .bat फ़ाइल की जगह user environment variable के रूप में रखें (Start → "Edit environment variables for your account")। फिर Task Scheduler टास्क बनाएँ (मॉड्यूल 6 लेसन 1): रोज़ सुबह 9:00, action = .bat, Start in = प्रोजेक्ट फ़ोल्डर, "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

(crontab आपके यूज़र अकाउंट तक निजी है, पर सिर्फ़ आपके पढ़ने लायक पासवर्ड फ़ाइल और भी बेहतर है।)

GitHub Actions? रिपोर्ट और ईमेल के लिए चलता है, पर GitHub की मशीन पर archive/ में भेजी फ़ाइलें रन के बाद गायब हो जाती हैं। इसे तब इस्तेमाल करें जब इनपुट डाउनलोड स्टेप (API, क्लाउड स्टोरेज) से आएँ, और archive स्टेप छोड़ दें — या archive को वापस repo में commit करें।

8. आख़िरी चेकलिस्ट

• बिल्कुल उसी शेड्यूल कमांड से नए टर्मिनल में चलता है
• शेड्यूलर में venv का Python और पूरे पाथ
• खाली inbox → exit 0; गलत डेटा → Rejected Rows; सब गलत → exit 1
• रिपोर्ट में 5 शीट, फ़ॉर्मेटेड हेडर, नंबर/तारीख़ फ़ॉर्मेट और एक चार्ट
• अटैचमेंट और सही टोटल के साथ ईमेल पहुँचता है
• प्रोसेस हुई CSV archive/<timestamp>/ में जाती हैं
• logs/pipeline.log हर रन समझाता है
• किसी कोड फ़ाइल में पासवर्ड नहीं
• एक पेज की README: क्या करता है, इनपुट फ़ॉर्मेट, कैसे चलाएँ, किसे फ़ोन करें

9. आगे बढ़ाने के आइडिया (एक-दो चुनें)

• फ़ेलियर अलर्ट: main() के except ब्लॉक में लॉग का आख़िरी हिस्सा खुद को ईमेल करें।
• महीने-दर-महीने: पिछले महीने की रिपोर्ट लोड करें और By Region में % बदलाव कॉलम जोड़ें।
• Config फ़ाइल: पाथ, प्राप्तकर्ता और reorder सीमा config.json में ले जाएँ।
• और आउटपुट: हर रीजन की एक फ़ाइल (मॉड्यूल 5 लेसन 3), हर रीजनल मैनेजर के ईमेल में अटैच।
• Google Sheets स्रोत: CSV की जगह Sheets API से डेटा लाएँ — या उल्टा, पूरा फ़्लो Apps Script (मॉड्यूल 4) में करें।
• टेस्ट: एक छोटी pytest फ़ाइल जो clean() को गड़बड़ रो देकर नतीजा जाँचे।

आम गलतियाँ

सिर्फ़ परफ़ेक्ट डेटा से जाँचना। रिपोर्ट सुरक्षित लिखे जाने से पहले archive करना। महीना या प्राप्तकर्ता कोड में फ़िक्स लिखना। हाथ से चलाना भरोसेमंद होने से पहले शेड्यूल करना। README न होना, जिससे कोई और इसे नहीं चला पाता।