Task Automation

Automate the following: whenever a new CSV file is added to the ~/data/incoming/
directory, validate the CSV headers, transform the data into JSON, and move the
result to ~/data/processed/. Log any files that fail validation to ~/data/errors/.

#!/usr/bin/env python3
"""File watcher that processes incoming CSVs into JSON.

Usage: python csv_watcher.py
Watches: ~/data/incoming/ for new .csv files
Outputs: ~/data/processed/*.json and ~/data/errors/error.log
"""

import os
import csv
import json
import time
import logging
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

INCOMING = Path.home() / "data" / "incoming"
PROCESSED = Path.home() / "data" / "processed"
ERRORS = Path.home() / "data" / "errors"
REQUIRED_HEADERS = ["id", "name", "email", "amount"]

logging.basicConfig(
    filename=ERRORS / "error.log",
    level=logging.ERROR,
    format="%(asctime)s - %(message)s"
)

for d in [INCOMING, PROCESSED, ERRORS]:
    d.mkdir(parents=True, exist_ok=True)


class CSVHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.src_path.endswith(".csv"):
            return
        filepath = Path(event.src_path)
        try:
            with open(filepath, "r") as f:
                reader = csv.DictReader(f)
                headers = reader.fieldnames or []
                missing = set(REQUIRED_HEADERS) - set(headers)
                if missing:
                    raise ValueError(f"Missing headers: {missing}")
                rows = list(reader)

            output = PROCESSED / filepath.with_suffix(".json").name
            with open(output, "w") as f:
                json.dump(rows, f, indent=2)

            filepath.unlink()  # Remove processed file
            print(f"Processed: {filepath.name} -> {output.name}")
        except Exception as e:
            logging.error(f"Failed to process {filepath.name}: {e}")
            filepath.rename(ERRORS / filepath.name)


if __name__ == "__main__":
    observer = Observer()
    observer.schedule(CSVHandler(), str(INCOMING), recursive=False)
    observer.start()
    print(f"Watching {INCOMING} for new CSV files...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

0 */6 * * * /usr/bin/python3 /opt/scripts/health_check.py >> /var/log/health_check.log 2>&1

#!/usr/bin/env python3
"""Poll API health endpoint and alert Slack on degraded services.

Runs every 6 hours via cron. Exits 0 on success, 1 on alert sent, 2 on script error.
"""

import os
import json
import urllib.request

HEALTH_URL = "https://api.example.com/health"
SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]


def check_health():
    req = urllib.request.Request(HEALTH_URL, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read())
    return data  # e.g., {"services": {"auth": "ok", "payments": "degraded", "db": "ok"}}


def send_slack_alert(degraded_services):
    service_list = "\n".join(f"- *{name}*: {status}" for name, status in degraded_services)
    payload = json.dumps({
        "text": f":warning: *Service Health Alert*\n{service_list}"
    }).encode()
    req = urllib.request.Request(
        SLACK_WEBHOOK,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST"
    )
    urllib.request.urlopen(req)


if __name__ == "__main__":
    health = check_health()
    degraded = [
        (name, status)
        for name, status in health.get("services", {}).items()
        if status != "ok"
    ]
    if degraded:
        send_slack_alert(degraded)
        print(f"Alert sent for {len(degraded)} degraded service(s)")
        exit(1)
    else:
        print("All services healthy")
        exit(0)