I got tired of writing the same utility scripts from scratch for every project. So I built a reusable collection.
Here are 5 scripts I use almost daily. They're all standalone (no heavy frameworks), tested, and ready to copy-paste.
My Downloads folder is a war zone. This script sorts files by type into clean folders โ and logs every move so you can undo it.
#!/usr/bin/env python3
"""Organize files by extension. Supports dry-run and undo."""
import json, shutil, sys
from pathlib import Path
CATEGORIES = {
"Images": {".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".heic"},
"Documents": {".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx"},
"Code": {".py", ".js", ".ts", ".go", ".rs", ".java", ".html", ".css"},
"Archives": {".zip", ".tar", ".gz", ".rar", ".7z"},
"Videos": {".mp4", ".mov", ".avi", ".mkv", ".webm"},
"Audio": {".mp3", ".wav", ".flac", ".aac", ".ogg"},
}
EXT_MAP = {ext: cat for cat, exts in CATEGORIES.items() for ext in exts}
def organize(directory, dry_run=False):
path = Path(directory)
log = []
for f in path.iterdir():
if not f.is_file() or f.name.startswith('.'):
continue
category = EXT_MAP.get(f.suffix.lower(), "Other")
dest = path / category
if not dry_run:
dest.mkdir(exist_ok=True)
shutil.move(str(f), str(dest / f.name))
log.append({"from": str(f), "to": str(dest / f.name)})
print(f"{'[DRY] ' if dry_run else ''} {f.name} โ {category}/")
if not dry_run:
(path / ".organize_log.json").write_text(json.dumps(log, indent=2))
print(f"\nโ Moved {len(log)} files. Run with --undo to restore.")
def undo(directory):
log_file = Path(directory) / ".organize_log.json"
for entry in json.loads(log_file.read_text()):
shutil.move(entry["to"], entry["from"])
log_file.unlink()
print("โ Restored all files.")
if __name__ == "__main__":
if "--undo" in sys.argv: undo(sys.argv[-1])
elif "--dry-run" in sys.argv: organize(sys.argv[1], dry_run=True)
else: organize(sys.argv[1])
python file_organizer.py ~/Downloads # organize
python file_organizer.py ~/Downloads --dry-run # preview first
python file_organizer.py --undo ~/Downloads # restore everything
The undo feature alone has saved me multiple times when I accidentally ran it on the wrong folder.
Pandas is great, but sometimes you just need to quickly filter, sort, or merge CSVs without installing a 400MB library.
#!/usr/bin/env python3
"""CSV operations without pandas: info, filter, convert to JSON."""
import csv, json, sys, operator
from pathlib import Path
def read_csv(filepath):
with open(filepath, newline="", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
return reader.fieldnames or [], list(reader)
def info(filepath):
headers, rows = read_csv(filepath)
print(f"File: {filepath}")
print(f"Rows: {len(rows):,} | Columns: {len(headers)}")
print(f"Headers: {', '.join(headers)}")
for h in headers:
values = [r[h] for r in rows if r[h]]
print(f" {h}: {len(set(values))} unique, {len(rows)-len(values)} empty")
def filter_rows(filepath, condition):
ops = {">=": operator.ge, "<=": operator.le, ">": operator.gt,
"<": operator.lt, "==": operator.eq, "!=": operator.ne}
for sym, func in sorted(ops.items(), key=lambda x: -len(x[0])):
if sym in condition:
field, value = [s.strip().strip("'\"") for s in condition.split(sym, 1)]
break
else:
raise ValueError(f"Can't parse: {condition}")
headers, rows = read_csv(filepath)
results = []
for row in rows:
cell = row.get(field, "")
try:
if func(float(cell), float(value)): results.append(row)
except ValueError:
if func(cell, value): results.append(row)
writer = csv.DictWriter(sys.stdout, fieldnames=headers)
writer.writeheader()
writer.writerows(results)
print(f"\n# {len(results)}/{len(rows)} rows matched", file=sys.stderr)
if __name__ == "__main__":
cmd, filepath = sys.argv[1], sys.argv[2]
if cmd == "info": info(filepath)
elif cmd == "filter": filter_rows(filepath, sys.argv[3])
elif cmd == "json":
_, rows = read_csv(filepath)
print(json.dumps(rows, indent=2))
python csv_toolkit.py info customers.csv
python csv_toolkit.py filter sales.csv "revenue > 10000"
python csv_toolkit.py json data.csv | jq '.[] | .email'
The full version adds dedup, merge, sort, split, and select โ but this covers 80% of my daily needs.
Track product prices on any website. Stores history in JSON, pings you on Telegram or Slack when prices drop.
#!/usr/bin/env python3
"""Monitor product prices and alert on drops."""
import json, re, os
from pathlib import Path
from datetime import datetime
import httpx # pip install httpx
DB = Path("price_monitor.json")
def load(): return json.loads(DB.read_text()) if DB.exists() else {"products": {}, "history": {}}
def save(db): DB.write_text(json.dumps(db, indent=2))
def extract_price(text):
cleaned = re.sub(r'[^\d.,]', '', text.strip())
if ',' in cleaned and '.' in cleaned: cleaned = cleaned.replace(',', '')
elif ',' in cleaned: cleaned = cleaned.replace(',', '.')
try: return float(cleaned)
except: return None
def add(url, selector, name):
db = load()
db["products"][name] = {"url": url, "selector": selector}
db["history"].setdefault(name, [])
save(db)
print(f"โ Tracking '{name}'")
def check():
from selectolax.parser import HTMLParser # pip install selectolax
db = load()
for name, prod in db["products"].items():
try:
r = httpx.get(prod["url"], headers={"User-Agent": "Mozilla/5.0"}, follow_redirects=True)
node = HTMLParser(r.text).css_first(prod["selector"])
if not node: print(f"โ {name}: selector not found"); continue
price = extract_price(node.text())
if price is None: print(f"โ {name}: can't parse"); continue
history = db["history"].get(name, [])
prev = history[-1]["price"] if history else None
history.append({"price": price, "date": datetime.now().isoformat()})
db["history"][name] = history[-100:]
arrow = "โ" if prev and price < prev else "โ" if prev and price > prev else "โ"
print(f"{arrow} {name}: ${price:.2f}" + (f" (was ${prev:.2f})" if prev else ""))
if prev and price < prev:
token = os.environ.get("TELEGRAM_BOT_TOKEN")
chat = os.environ.get("TELEGRAM_CHAT_ID")
if token and chat:
httpx.post(f"https://api.telegram.org/bot{token}/sendMessage",
json={"chat_id": chat, "text": f"๐ {name}: ${prev:.2f} โ ${price:.2f}\n{prod['url']}"})
except Exception as e:
print(f"โ {name}: {e}")
save(db)
I run check() via cron every 4 hours. Works on any site with a CSS-selectable price element.
Need to process 20 PDFs or research papers? This script summarizes them using any OpenAI-compatible API.
#!/usr/bin/env python3
"""Summarize multiple documents using any OpenAI-compatible API."""
import os, sys
from pathlib import Path
import httpx
API_KEY = os.environ.get("OPENAI_API_KEY", "")
MODEL = os.environ.get("LLM_MODEL", "gpt-4o-mini")
BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1")
def summarize(text, max_tokens=500):
r = httpx.post(f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": MODEL, "max_tokens": max_tokens,
"messages": [
{"role": "system", "content": "Summarize concisely. Key points as bullets."},
{"role": "user", "content": f"Summarize:\n\n{text[:8000]}"}
]}, timeout=60)
return r.json()["choices"][0]["message"]["content"]
def process_file(filepath):
path = Path(filepath)
if path.suffix == ".pdf":
try:
import fitz # pip install pymupdf
text = "\n".join(page.get_text() for page in fitz.open(str(path)))
except ImportError:
print(f" Skip {path.name} (install pymupdf for PDF)"); return None
else:
text = path.read_text(encoding="utf-8", errors="ignore")
return summarize(text)
if __name__ == "__main__":
for f in sys.argv[1:]:
print(f"\n{'='*60}\n๐ {Path(f).name}\n{'='*60}")
result = process_file(f)
if result: print(result)
# With OpenAI
python batch_summarize.py report1.pdf report2.pdf notes.txt
# With local Ollama
LLM_BASE_URL=http://localhost:11434/v1 LLM_MODEL=llama3 python batch_summarize.py *.md
Every automation needs alerts. Instead of rewriting notification code for each project, I use this one module everywhere.
#!/usr/bin/env python3
"""Send to Telegram, Slack, Discord in one call."""
import os, httpx
def send(message, title=None, channels=None):
channels = channels or ["telegram", "slack", "discord"]
full = f"*{title}*\n{message}" if title else message
results = []
if "telegram" in channels:
token = os.environ.get("TELEGRAM_BOT_TOKEN")
chat = os.environ.get("TELEGRAM_CHAT_ID")
if token and chat:
r = httpx.post(f"https://api.telegram.org/bot{token}/sendMessage",
json={"chat_id": chat, "text": full, "parse_mode": "Markdown"})
results.append(("telegram", r.status_code == 200))
if "slack" in channels:
webhook = os.environ.get("SLACK_WEBHOOK_URL")
if webhook:
r = httpx.post(webhook, json={"text": full})
results.append(("slack", r.status_code == 200))
if "discord" in channels:
webhook = os.environ.get("DISCORD_WEBHOOK_URL")
if webhook:
r = httpx.post(webhook, json={"content": full[:2000]})
results.append(("discord", r.status_code == 204))
return results
# Usage from any script:
# from notify import send
# send("Deploy complete โ
", title="CI/CD", channels=["slack"])
Dead simple. Set env vars once, import from anywhere.
These scripts are from the AI Agent Toolkit โ a collection of 50+ production-ready Python scripts plus 30+ curated AI prompts covering web scraping, DevOps, data processing, social media, and more.
Every script is standalone, documented, and MIT licensed.
Get the Toolkit โ $19