Mini project automation suite

This mini project builds a small automation suite: watch a folder for new CSV files, validate and summarise them, write a JSON report, and email a digest when something arrives. No database – files on disk carry the state. This is the kind of glue script that saves a team hours once it runs reliably on a schedule.

What we are building

• Drop zone folder for incoming .csv sales exports
• Processor validates rows and computes totals
• Processed files move to archive/; failures go to failed/
• Report written to reports/latest.json
• Optional SMTP email with a plain-text summary
• CLI entry point for cron or manual runs

Project layout

sales_automation/
├── pyproject.toml
├── inbox/
├── archive/
├── failed/
├── reports/
└── src/sales_automation/
    ├── __init__.py
    ├── processor.py
    ├── reporter.py
    ├── mailer.py
    └── cli.py

CSV processor

import csv
import shutil
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

@dataclass
class Summary:
    source: str
    row_count: int
    total_revenue: float
    processed_at: str

class SalesProcessor:
    def __init__(self, inbox: Path, archive: Path, failed: Path) -> None:
        self.inbox = inbox
        self.archive = archive
        self.failed = failed
        for folder in (inbox, archive, failed):
            folder.mkdir(parents=True, exist_ok=True)

    def process_all(self) -> list[Summary]:
        summaries = []
        for path in sorted(self.inbox.glob("*.csv")):
            try:
                summaries.append(self._process_file(path))
            except Exception:
                shutil.move(path, self.failed / path.name)
                raise
        return summaries

    def _process_file(self, path: Path) -> Summary:
        total = 0.0
        rows = 0
        with path.open(newline="", encoding="utf-8") as handle:
            reader = csv.DictReader(handle)
            for row in reader:
                qty = int(row["qty"])
                price = float(row["unit_price"])
                if qty < 0 or price < 0:
                    raise ValueError(f"invalid row in {path.name}")
                total += qty * price
                rows += 1
        shutil.move(path, self.archive / path.name)
        return Summary(
            source=path.name,
            row_count=rows,
            total_revenue=round(total, 2),
            processed_at=datetime.now(timezone.utc).isoformat(),
        )

Report writer

import json
import tempfile
from dataclasses import asdict
from pathlib import Path

def write_report(path: Path, summaries: list) -> None:
    payload = {
        "files_processed": len(summaries),
        "summaries": [asdict(s) for s in summaries],
        "grand_total": round(sum(s.total_revenue for s in summaries), 2),
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", delete=False, dir=path.parent) as tmp:
        json.dump(payload, tmp, indent=2)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
    return payload

Email digest

Use environment variables for SMTP credentials. In development, point at Mailpit or Mailhog and skip real delivery.

import os
import smtplib
from email.message import EmailMessage

def send_digest(report: dict, to_address: str) -> None:
    host = os.environ.get("SMTP_HOST")
    if not host:
        return  # email disabled

    lines = [
        f"Files processed: {report['files_processed']}",
        f"Grand total: £{report['grand_total']:,.2f}",
        "",
    ]
    for item in report["summaries"]:
        lines.append(f"- {item['source']}: {item['row_count']} rows, £{item['total_revenue']:,.2f}")

    message = EmailMessage()
    message["Subject"] = "Sales import digest"
    message["From"] = os.environ["SMTP_FROM"]
    message["To"] = to_address
    message.set_content("n".join(lines))

    with smtplib.SMTP(host, int(os.environ.get("SMTP_PORT", "587"))) as smtp:
        smtp.starttls()
        smtp.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
        smtp.send_message(message)

CLI entry point

import argparse
import logging
from pathlib import Path
from sales_automation.processor import SalesProcessor
from sales_automation.reporter import write_report
from sales_automation.mailer import send_digest

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger(__name__)

def main() -> None:
    parser = argparse.ArgumentParser(description="Process sales CSV drops")
    parser.add_argument("--root", type=Path, default=Path("."))
    parser.add_argument("--email-to", default="")
    args = parser.parse_args()

    processor = SalesProcessor(
        inbox=args.root / "inbox",
        archive=args.root / "archive",
        failed=args.root / "failed",
    )
    summaries = processor.process_all()
    if not summaries:
        log.info("No files to process")
        return

    report = write_report(args.root / "reports" / "latest.json", summaries)
    log.info("Processed %s file(s), total £%.2f", len(summaries), report["grand_total"])

    if args.email_to:
        send_digest(report, args.email_to)
        log.info("Digest sent to %s", args.email_to)

if __name__ == "__main__":
    main()

Scheduling

Add a cron entry or systemd timer that runs the CLI every few minutes. Ensure only one instance runs at a time – a simple file lock or flock wrapper prevents overlapping imports.

# crontab example - every 5 minutes:
# */5 * * * * cd /opt/sales_automation && /opt/venv/bin/sales-import --email-to ops@example.com

Hardening ideas

• Log to a file with rotation, not just stdout
• Alert on files landing in failed/
• Tests with sample CSV fixtures in tests/fixtures/
• Move long imports to a background worker if files grow huge

This project stitches together the pipeline thinking from tutorial 50, background job ideas from tutorial 70, and plain file I/O you already know. It is boring in the best way – predictable folders, explicit logs, email when something worth noting happens.