Workflow Automation with Python Scripts — Automate Repetitive Tasks

from pathlib import Path
import shutil
from datetime import datetime, timedelta

def cleanup_old_logs(log_dir, days_old=30):
    """Remove log files older than specified days."""
    log_path = Path(log_dir)
    cutoff = datetime.now() - timedelta(days=days_old)
    removed_count = 0
    freed_bytes = 0

    for log_file in log_path.glob("*.log"):
        mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
        if mtime < cutoff:
            freed_bytes += log_file.stat().st_size
            log_file.unlink()
            removed_count += 1

    return removed_count, freed_bytes

removed, freed = cleanup_old_logs("/var/log/myapp")
print(f"Cleaned {removed} files ({freed / 1048576:.2f} MB freed)")

from pathlib import Path

def bulk_rename(directory, prefix, extension=None):
    """Rename all files in directory with a prefix and sequential number."""
    path = Path(directory)
    files = list(path.iterdir())

    if extension:
        files = [f for f in files if f.suffix == extension]

    for index, file in enumerate(sorted(files), start=1):
        new_name = f"{prefix}_{index:03d}{file.suffix}"
        file.rename(path / new_name)
        print(f"Renamed: {file.name} -> {new_name}")

# Usage: prefix all images with "screenshot"
bulk_rename("./screenshots", "screenshot", ".png")

import requests
import json
from pathlib import Path

API_KEY = "your-api-key-here"
BASE_URL = "https://api.example.com/v1"

def fetch_and_save_alerts(status="open", output_file="alerts.json"):
    """Fetch open alerts from an API and save to a JSON file."""
    headers = {"Authorization": f"Bearer {API_KEY}"}
    params = {"status": status, "limit": 100}

    response = requests.get(
        f"{BASE_URL}/alerts",
        headers=headers,
        params=params,
        timeout=30
    )
    response.raise_for_status()

    alerts = response.json()
    output = Path(output_file)
    output.write_text(json.dumps(alerts, indent=2))

    print(f"Saved {len(alerts)} {status} alerts to {output_file}")

    # Count by severity
    severity_counts = {}
    for alert in alerts:
        sev = alert.get("severity", "unknown")
        severity_counts[sev] = severity_counts.get(sev, 0) + 1

    for sev, count in sorted(severity_counts.items()):
        print(f"  {sev}: {count}")

fetch_and_save_alerts()

pip install schedule

import schedule
import time
from pathlib import Path

def backup_database():
    """Simulate database backup."""
    timestamp = time.strftime("%Y%m%d_%H%M%S")
    backup_file = Path(f"/backups/db_dump_{timestamp}.sql")
    # In real code: subprocess.run(["pg_dump", ...])
    backup_file.write_text("-- simulated database dump")
    print(f"Backup created: {backup_file.name}")

def check_disk_space():
    """Check disk usage and alert if low."""
    import shutil
    usage = shutil.disk_usage("/")
    percent_used = (usage.used / usage.total) * 100
    print(f"Disk usage: {percent_used:.1f}%")
    if percent_used > 90:
        print("WARNING: Disk usage above 90%!")

def send_daily_report():
    """Generate and send daily summary."""
    print("Generating daily report...")
    # In real code: generate report and email it

# Schedule jobs
schedule.every().day.at("02:00").do(backup_database)
schedule.every().hour.do(check_disk_space)
schedule.every().day.at("08:00").do(send_daily_report)

print("Scheduler started. Press Ctrl+C to stop.")
while True:
    schedule.run_pending()
    time.sleep(60)

import smtplib
from email.message import EmailMessage
from pathlib import Path

SMTP_SERVER = "smtp.example.com"
SMTP_PORT = 587
SMTP_USER = "alerts@example.com"
SMTP_PASS = "your-password"

def send_alert(subject, body, attachment_path=None):
    """Send an email alert, optionally with an attachment."""
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = SMTP_USER
    msg["To"] = "team@example.com"
    msg.set_content(body)

    if attachment_path:
        path = Path(attachment_path)
        if path.exists():
            with open(path, "rb") as f:
                data = f.read()
            msg.add_attachment(
                data,
                maintype="application",
                subtype="octet-stream",
                filename=path.name
            )

    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SMTP_USER, SMTP_PASS)
        server.send_message(msg)
    print(f"Alert sent: {subject}")

send_alert(
    "Deployment Complete - v2.3.1",
    "The deployment of version 2.3.1 finished successfully.\n\n"
    "Changes:\n- Fixed login timeout issue\n- Updated dependencies\n- Added new API endpoint"
)

from pathlib import Path
import gzip
import shutil
from datetime import datetime, timedelta

LOG_DIR = Path("/var/log/myapp")
ARCHIVE_DIR = Path("/var/log/myapp/archive")
RETENTION_DAYS = 7
ARCHIVE_RETENTION = 90

def rotate_logs():
    ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)
    cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)
    archive_cutoff = datetime.now() - timedelta(days=ARCHIVE_RETENTION)
    stats = {"compressed": 0, "deleted": 0, "size_saved": 0}

    for log_file in LOG_DIR.glob("*.log"):
        mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
        if mtime < cutoff:
            archive_path = ARCHIVE_DIR / f"{log_file.stem}.gz"
            with open(log_file, "rb") as f_in:
                with gzip.open(archive_path, "wb") as f_out:
                    shutil.copyfileobj(f_in, f_out)
            stats["size_saved"] += log_file.stat().st_size
            stats["compressed"] += 1
            log_file.unlink()

    for archive in ARCHIVE_DIR.glob("*.gz"):
        mtime = datetime.fromtimestamp(archive.stat().st_mtime)
        if mtime < archive_cutoff:
            archive.unlink()
            stats["deleted"] += 1

    print(f"Compressed {stats['compressed']} logs")
    print(f"Deleted {stats['deleted']} old archives")
    print(f"Freed {stats['size_saved'] / 1048576:.2f} MB")

if __name__ == "__main__":
    rotate_logs()