Automating Files, Logs and Scheduled Reports

An analysis you run once is a project. An analysis you run every Monday morning is a process, and processes should not depend on someone remembering to open a notebook. Automation means taking the steps you already know how to do (find the files, load them, summarise, save a report, tell somebody) and arranging them so that they run by themselves, reliably, and tell you clearly when something goes wrong.

This lesson ties together much of the course. We use pathlib and shutil to organise and combine files, the logging module to leave a trail, argparse to turn a script into a proper command-line tool with options, subprocess to run it the way a scheduler would, email to compose a message with an attachment, and finally the scheduling tools (cron and Windows Task Scheduler) that start it at the right time. Along the way you will meet the habits that make automation trustworthy: scripts that can be safely re-run, that exit with a clear status and that never carry passwords in the code.

Everything runs in a temporary folder inside the lesson, so you can experiment without touching your real files.

Worked examples

Example 1: Split one big file into monthly files

Question: A system exports everything in one file. How do we create one file per month?

import pandas as pd
from pathlib import Path

orders = pd.read_csv("orders.csv", parse_dates=["order_date"])
inbox = Path("inbox")
inbox.mkdir(exist_ok=True)

for month, group in orders.groupby(orders["order_date"].dt.to_period("M")):
    group.to_csv(inbox / f"orders_{month}.csv", index=False)

files = sorted(inbox.glob("orders_*.csv"))
print(len(files))
print([f.name for f in files[:3]], "...", files[-1].name)
print(sum(1 for _ in inbox.iterdir()))

18
['orders_2025-01.csv', 'orders_2025-02.csv', 'orders_2025-03.csv'] ... orders_2026-06.csv
18

How it works: Path("inbox") is an object representing a folder, and mkdir(exist_ok=True) creates it without complaining if it exists. The / operator joins path parts, which works on every operating system. We loop over month groups and write each to a CSV named after its month. glob("orders_*.csv") finds files by pattern and sorted gives a predictable order.

Reading the result: We created 18 monthly files, from orders_2025-01.csv to orders_2026-06.csv. The folder contains 18 items.

Example 2: Combine files and check nothing was lost

Question: How do we reload all the monthly files and prove we have everything?

parts = []
for f in sorted(inbox.glob("orders_*.csv")):
    part = pd.read_csv(f, parse_dates=["order_date"])
    part["source_file"] = f.name
    parts.append(part)

combined = pd.concat(parts, ignore_index=True)
print(combined.shape)
print("rows match original:", len(combined) == len(orders))
print("order IDs unique:", combined["order_id"].is_unique)
print(combined["source_file"].value_counts().sort_index().head(3).to_dict())

(6000, 10)
rows match original: True
order IDs unique: True
{'orders_2025-01.csv': 66, 'orders_2025-02.csv': 67, 'orders_2025-03.csv': 109}

How it works: Read each file, tag it with its file name (so you can trace any row back to its source), collect the pieces in a list and concat them once. Concatenating inside a loop is slow, because it copies the data each time. The two checks at the end are reconciliation: the row count equals the original, and the IDs are unique (no file was loaded twice). Automation should always verify its own output.

Reading the result: The combined table has 6000 rows from 18 files; the count matches the original (6000).

Example 3: Archive processed files with shutil

Question: After processing, how do we move files out of the inbox so they are not processed twice?

import shutil

archive = Path("archive")
archive.mkdir(exist_ok=True)

moved = []
for f in sorted(inbox.glob("orders_2025-0[1-3].csv")):
    shutil.move(str(f), archive / f.name)
    moved.append(f.name)

print(moved)
print(len(list(inbox.glob("*.csv"))), "left in inbox;", len(list(archive.glob("*.csv"))), "in archive")

zip_path = shutil.make_archive("archive_q1_2025", "zip", root_dir=archive)
print(Path(zip_path).name, Path(zip_path).stat().st_size > 500)

['orders_2025-01.csv', 'orders_2025-02.csv', 'orders_2025-03.csv']
15 left in inbox; 3 in archive
archive_q1_2025.zip True

How it works: shutil.move(src, dst) moves a file (it also works across drives), shutil.copy2 copies it with timestamps preserved and shutil.make_archive creates a zip of a folder. Moving finished files to an archive folder is the simplest way to mark them as done: the next run only sees new files. (Deleting is riskier, because it cannot be undone.)

Reading the result: We moved 3 files into the archive and zipped them as archive_q1_2025.zip. The inbox now holds 15 files and the archive 3.

Example 4: Logging instead of print

Question: How do we record what a script did, with severity levels and a log file?

import logging
import sys

logger = logging.getLogger("freshbasket")
logger.setLevel(logging.DEBUG)
logger.handlers.clear()

console = logging.StreamHandler(sys.stdout)
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter("%(levelname)-8s %(message)s"))

logfile = logging.FileHandler("run.log", encoding="utf-8")
logfile.setLevel(logging.DEBUG)
logfile.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))

logger.addHandler(console)
logger.addHandler(logfile)

logger.debug("loading %d files", len(parts))
logger.info("combined %d rows", len(combined))
logger.warning("12 customers have never ordered")
logger.error("could not reach the email server")

lines = Path("run.log").read_text(encoding="utf-8").splitlines()
print(len(lines), "lines in the log file;", "DEBUG" in lines[0])

INFO     combined 6000 rows
WARNING  12 customers have never ordered
ERROR    could not reach the email server
4 lines in the log file; True

How it works: The logging module attaches handlers to a logger. Each handler has its own minimum level and format: here the console shows INFO and above in a short format, while the file keeps everything (including DEBUG) with timestamps. Levels in increasing severity are DEBUG, INFO, WARNING, ERROR and CRITICAL. Using %s-style arguments (logger.info("rows: %d", n)) is lazy: the message is only built if a handler will use it. When a scheduled job fails at 3 a.m., the log file is the only witness.

Reading the result: The console showed 3 messages (INFO, WARNING and ERROR) while the log file recorded all 4 lines including the DEBUG one, each with a timestamp.

Example 5: A command-line script with argparse

Question: How do we turn our code into a tool that others (and schedulers) can run with options?

script = '''
import argparse
import sys
from pathlib import Path
import pandas as pd

def main(argv=None):
    parser = argparse.ArgumentParser(description="FreshBasket monthly order summary")
    parser.add_argument("--month", required=True, help="month as YYYY-MM")
    parser.add_argument("--orders", default="orders.csv", help="path to orders.csv")
    parser.add_argument("--top", type=int, default=3, help="how many cities to list")
    args = parser.parse_args(argv)

    path = Path(args.orders)
    if not path.exists():
        print(f"error: {path} not found", file=sys.stderr)
        return 1

    orders = pd.read_csv(path, parse_dates=["order_date"])
    stores = pd.read_csv(path.parent / "stores.csv")
    df = orders.merge(stores[["store_id", "city"]], on="store_id")
    month = df[df["order_date"].dt.strftime("%Y-%m") == args.month]
    if month.empty:
        print(f"error: no orders in {args.month}", file=sys.stderr)
        return 2

    delivered = month[month["order_status"] == "Delivered"]
    print(f"Month {args.month}: {len(month)} orders, {len(delivered)} delivered")
    top = delivered["city"].value_counts().head(args.top)
    for city, n in top.items():
        print(f"  {city}: {n}")
    return 0

if __name__ == "__main__":
    sys.exit(main())
'''
Path("weekly_report.py").write_text(script.lstrip(), encoding="utf-8")
print(Path("weekly_report.py").read_text().count("\n"), "lines written")

34 lines written

How it works: argparse defines the options the script accepts (--month, --orders, --top) with types, defaults and help text, then parse_args reads them from the command line. The pattern if __name__ == "__main__": sys.exit(main()) runs main only when the file is executed directly, and main returns an exit code: 0 for success, non-zero for failure. Errors go to stderr (file=sys.stderr) so they do not mix with normal output. We write the script to a file here so that the next examples can run it exactly as a scheduler would.

Reading the result: The script file has 34 lines. It defines three options and two failure exit codes (1 for a missing file, 2 for a month with no data).

Example 6: Run it like a scheduler and read the exit code

Question: What happens in the successful and failing cases?

import subprocess

def run(*args):
    result = subprocess.run([sys.executable, "weekly_report.py", *args], capture_output=True, text=True)
    return result.returncode, result.stdout.strip(), result.stderr.strip()

ok = run("--month", "2026-06", "--top", "2")
print(ok[0]); print(ok[1])

missing = run("--month", "2026-06", "--orders", "nope.csv")
print(missing[0], missing[2])

empty = run("--month", "2030-01")
print(empty[0], empty[2])

helptext = subprocess.run([sys.executable, "weekly_report.py", "--help"], capture_output=True, text=True).stdout
print(helptext.splitlines()[0])

0
Month 2026-06: 630 orders, 562 delivered
  Delhi: 115
  Bengaluru: 102
1 error: nope.csv not found
2 error: no orders in 2030-01
usage: weekly_report.py [-h] --month MONTH [--orders ORDERS] [--top TOP]

How it works: subprocess.run launches the script in a separate process, just like cron or Task Scheduler does, capturing its output and return code. Schedulers and CI systems judge success by that code: 0 is success and anything else is failure, which can trigger an alert. Because we chose distinct codes (1 and 2), whoever monitors the job knows why it failed. --help is generated automatically by argparse.

Reading the result: The normal run returned code 0 and printed 3 lines. The missing file returned 1 ("error: nope.csv not found") and the empty month returned 2 ("error: no orders in 2030-01").

Example 7: Idempotent runs

Question: What if the job runs twice by accident, or after a crash?

def build_summary(month, out_dir="reports"):
    out = Path(out_dir)
    out.mkdir(exist_ok=True)
    target = out / f"summary_{month}.csv"
    if target.exists():
        return target, "skipped (already built)"

    m = orders[orders["order_date"].dt.strftime("%Y-%m") == month]
    summary = m["order_status"].value_counts().rename_axis("status").reset_index(name="orders")

    tmp = target.with_suffix(".tmp")
    summary.to_csv(tmp, index=False)
    tmp.replace(target)                      # atomic rename: no half-written files
    return target, "built"

print(build_summary("2026-06")[1])
print(build_summary("2026-06")[1])
print(sorted(p.name for p in Path("reports").iterdir()))

built
skipped (already built)
['summary_2026-06.csv']

How it works: A job is idempotent if running it twice gives the same result as running it once. Here, if the output exists the function skips the work. We also write to a temporary file and then rename it: a rename is atomic, so a crash can never leave a half-written report that looks finished. These two habits (skip what is done, never leave partial output) make recovery trivial: just run the job again.

Reading the result: The first call reported "built" and the second "skipped (already built)", leaving exactly 1 file in the folder.

Example 8: Compose an email with an attachment

Question: How do we prepare the weekly email for a manager?

from email.message import EmailMessage

msg = EmailMessage()
msg["Subject"] = "FreshBasket monthly summary - June 2026"
msg["From"] = "reports@example.com"
msg["To"] = "operations@example.com"
msg.set_content("Hello,\n\nThe June summary is attached.\n\nRegards,\nReporting bot")

csv_bytes = Path("reports/summary_2026-06.csv").read_bytes()
msg.add_attachment(csv_bytes, maintype="text", subtype="csv", filename="summary_2026-06.csv")

print(msg["Subject"])
print([part.get_content_type() for part in msg.walk()])
print([part.get_filename() for part in msg.iter_attachments()])
print(len(msg.as_bytes()) > 300)

FreshBasket monthly summary - June 2026
['multipart/mixed', 'text/plain', 'text/csv']
['summary_2026-06.csv']
True

import smtplib, os

with smtplib.SMTP_SSL("smtp.example.com", 465) as server:
    server.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
    server.send_message(msg)

How it works: EmailMessage builds a standards-compliant message: headers, a plain-text body and an attachment. Building a message is safe and testable, so we run it. The second block, which is not executed here, shows how it would be sent with smtplib: the credentials come from environment variables, never from the code. Many companies use an email API instead of SMTP; the message-building step stays the same idea. Test email code by printing the message first, and send to yourself before sending to a manager.

Reading the result: The message "FreshBasket monthly summary - June 2026" has 1 attachment, summary_2026-06.csv, and the content types ['multipart/mixed', 'text/plain', 'text/csv'].

Example 9: Calculate the next scheduled run

Question: If a job runs every Monday at 08:00, when does it run next?

from datetime import datetime, timedelta

def next_weekly_run(now, weekday=0, hour=8):
    """weekday: Monday=0. Returns the next datetime at that weekday and hour."""
    days_ahead = (weekday - now.weekday()) % 7
    candidate = (now + timedelta(days=days_ahead)).replace(hour=hour, minute=0, second=0, microsecond=0)
    if candidate <= now:
        candidate += timedelta(days=7)
    return candidate

for stamp in ["2026-10-25 15:30", "2026-10-26 07:59", "2026-10-26 08:00"]:
    now = datetime.strptime(stamp, "%Y-%m-%d %H:%M")
    print(stamp, "->", next_weekly_run(now).strftime("%a %d %b %Y %H:%M"))

2026-10-25 15:30 -> Mon 26 Oct 2026 08:00
2026-10-26 07:59 -> Mon 26 Oct 2026 08:00
2026-10-26 08:00 -> Mon 02 Nov 2026 08:00

# cron: minute hour day-of-month month day-of-week   command
# every Monday at 08:00
0 8 * * 1  cd /opt/reports && /usr/bin/python3 weekly_report.py --month "$(date +\%Y-\%m)" >> run.log 2>&1

# Windows Task Scheduler (command line)
schtasks /Create /SC WEEKLY /D MON /ST 08:00 /TN "FreshBasket report" /TR "python C:\reports\weekly_report.py --month 2026-06"

How it works: Schedulers use a compact syntax: a cron line has five time fields (minute, hour, day of month, month, day of week) followed by the command; Windows Task Scheduler offers the same in a graphical tool or via schtasks. Cloud platforms (GitHub Actions, Airflow, cloud schedulers) use cron expressions too. The function above is our own reasoning about "next Monday at 8": note the edge case that at exactly 08:00 on Monday the next run is a week away.

Reading the result: From Sunday 25 October the next run is Mon 26 Oct 08:00, and at exactly Monday 08:00 it moves to Mon 02 Nov 08:00.

Example 10: Settings in a config file

Question: How do we keep paths, recipients and thresholds out of the code?

import json

DEFAULTS = {"orders_path": "orders.csv", "top_cities": 3, "recipients": [], "alert_if_cancel_rate_above": 0.08}

def load_config(path):
    config = dict(DEFAULTS)
    p = Path(path)
    if p.exists():
        config.update(json.loads(p.read_text(encoding="utf-8")))
    unknown = set(config) - set(DEFAULTS)
    if unknown:
        raise ValueError(f"unknown settings: {sorted(unknown)}")
    return config

Path("settings.json").write_text(json.dumps({"top_cities": 5, "recipients": ["operations@example.com"]}, indent=2), encoding="utf-8")
cfg = load_config("settings.json")
print(cfg)

try:
    Path("bad.json").write_text('{"top_citys": 5}')
    load_config("bad.json")
except ValueError as e:
    print("caught:", e)

rate = (orders["order_status"] == "Cancelled").mean()
print(f"cancel rate {rate:.2%}; alert:", rate > cfg["alert_if_cancel_rate_above"])

{'orders_path': 'orders.csv', 'top_cities': 5, 'recipients': ['operations@example.com'], 'alert_if_cancel_rate_above': 0.08}
caught: unknown settings: ['top_citys']
cancel rate 6.17%; alert: False

How it works: A JSON config file with sensible defaults lets non-programmers change recipients or thresholds without touching code, and lets you run the same script in test and production. Validating the keys catches typos (top_citys) immediately instead of silently ignoring them. The last lines use a threshold from the config to decide whether to raise an alert, a simple but real automation rule. Passwords and tokens still belong in environment variables or a secrets manager, never in this file.

Reading the result: The loaded settings ask for 5 cities and 1 recipient. The overall cancellation rate is 6.17%, below the 8% alert threshold, so no alert is raised.

Common mistakes

1. Hard-coded absolute paths like C:\Users\me\Desktop\data.csv. Use relative paths, Path(__file__).parent or configuration.

2. Scripts that are not safe to re-run, producing duplicate rows or half-written files after a crash.

3. Silent failures. A job that swallows exceptions and exits with 0 will fail unnoticed. Log errors and return a non-zero exit code.

4. Credentials in the code, where they leak into version control and screenshots.

5. Testing only the happy path. Try missing files, empty inputs, a locked file and a month with no data. In the example below, check=True turns the script's failure exit code (2, no orders that month) into an exception.

subprocess.run([sys.executable, "weekly_report.py", "--month", "2030-01"], check=True, capture_output=True)

Traceback (most recent call last):
  File "example.py", line 1, in <module>
    subprocess.run([sys.executable, "weekly_report.py", "--month", "2030-01"], check=True, capture_output=True)
    ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "example.py", line 578, in run
    raise CalledProcessError(retcode, process.args,
                             output=stdout, stderr=stderr)
subprocess.CalledProcessError: Command '['/usr/local/bin/python3', 'weekly_report.py', '--month', '2030-01']' returned non-zero exit status 2.

Practice questions

Practice 1 (Easy): List the biggest files

List the three largest files in the inbox folder with their size in kilobytes.

sizes = sorted(((f.stat().st_size / 1024, f.name) for f in inbox.glob("*.csv")), reverse=True)[:3]
for kb, name in sizes:
    print(f"{name}: {kb:.1f} KB")

orders_2026-06.csv: 39.2 KB
orders_2026-04.csv: 38.6 KB
orders_2026-05.csv: 37.7 KB

Practice 2 (Medium): A processing function with logging

Write process_file(path) that reads a monthly CSV, logs how many rows it has, logs a warning if more than 7% of orders are cancelled, and returns the cancel rate. Run it over the files left in the inbox and report the worst month.

def process_file(path):
    df = pd.read_csv(path)
    rate = (df["order_status"] == "Cancelled").mean()
    logger.info("%s: %d rows, cancel rate %.1f%%", path.name, len(df), rate * 100)
    if rate > 0.07:
        logger.warning("%s: cancel rate above 7%%", path.name)
    return rate

rates = {f.name: process_file(f) for f in sorted(inbox.glob("orders_*.csv"))}
worst = max(rates, key=rates.get)
print(worst, round(rates[worst], 3))

orders_2025-06.csv 0.088

Practice 3 (Hard): Process only new files

Write process_new(inbox, archive) that processes every CSV in the inbox, moves it to the archive afterwards, and does nothing on a second call. Verify by calling it twice and counting.

def process_new(inbox_dir, archive_dir):
    inbox_dir, archive_dir = Path(inbox_dir), Path(archive_dir)
    archive_dir.mkdir(exist_ok=True)
    done = []
    for f in sorted(inbox_dir.glob("orders_*.csv")):
        df = pd.read_csv(f)
        done.append((f.name, len(df)))
        shutil.move(str(f), archive_dir / f.name)
    return done

first = process_new(inbox, archive)
second = process_new(inbox, archive)
print(len(first), len(second), len(list(archive.glob("*.csv"))))
print(sum(n for _, n in first))

15 0 18
5758

Key takeaways

• pathlib and shutil handle files portably; reconcile row counts after combining files.
• Use logging with levels and a file instead of print in unattended jobs.
• argparse plus if __name__ == "__main__" turns code into a command-line tool, and exit codes tell schedulers what happened.
• Make jobs idempotent: skip finished work, write to a temp file and rename.
• Build emails with EmailMessage; keep credentials in environment variables.
• Schedule with cron, Task Scheduler or a cloud scheduler, and monitor the job, because a silent failure is the worst failure.