SameOS ~/tools/automation-lessons.md

Python Automation Code Lessons: Deduplication, Scheduled Jobs, SQLite History

Automation code is usually more useful when the core pattern is explained instead of publishing the full private worker. This guide breaks file cleanup, repeat jobs, and processing history into three small examples. All of them are plain Python: save the code as a file and run it from a terminal (the command-line window) with python3 script_name.py.

1. Remove Duplicate Files With Hashes

Filenames change easily, but a SHA-256 hash (a short fingerprint that summarizes a file's contents) matches only when the file contents match, so comparing hashes instead of names finds duplicates reliably. For large image folders, limit the scope first by skipping thumbnail folders and scanning only original files.

  • Use extension filters to reduce work
  • Read in 1 MB chunks to keep memory use low
  • Treat a repeated hash as a duplicate candidate
from pathlib import Path
import hashlib

IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif"}

def file_hash(path: Path) -> str:
    # Read the file in 1 MB chunks and compute a SHA-256 hash (content fingerprint)
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()

def image_files(root: Path):
    for path in root.rglob("*"):
        if "thumbs" in path.parts:
            continue  # skip thumbnail folders
        if path.is_file() and path.suffix.lower() in IMAGE_EXTS:
            yield path

def duplicate_groups(root: str):
    seen = {}
    for path in image_files(Path(root)):
        digest = file_hash(path)
        if digest in seen:
            yield seen[digest], path  # (original, duplicate candidate)
        else:
            seen[digest] = path  # remember the first file for each new hash

2. Run Scheduled Jobs Without Overlap

Automation jobs can take longer than expected. If the next scan starts before the previous one ends, the same file may be processed twice or files may conflict. A "currently running" flag (a simple on/off variable), lowered in a finally block that runs even when an error occurs, keeps the job predictable.

  • Skip the next cycle while a job is running
  • Move heavy work to a worker thread to avoid blocking the event loop
  • Restore state even when an exception occurs
import asyncio

running = False

async def scan_once():
    # Put slow network or disk work here.
    return await asyncio.to_thread(do_blocking_scan)

async def scheduler(interval_seconds: int):
    global running
    while True:
        if not running:  # skip this cycle while the previous job is still running
            running = True
            try:
                results = await scan_once()
                await send_results(results)
            finally:
                running = False  # always lower the flag, even on exceptions
        await asyncio.sleep(interval_seconds)

3. Keep Processing History in SQLite

One table in SQLite (a lightweight database that lives in a single file - no server to install, and built into Python) is enough to avoid processing the same item twice. PRIMARY KEY (a unique key that cannot be stored twice) plus INSERT OR IGNORE naturally prevents duplicate history rows.

  • Store processed IDs as primary keys
  • Keep history after restarts
  • Store settings in a key-value table too
import sqlite3

DB_PATH = "worker.sqlite3"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS history (
                item_id TEXT PRIMARY KEY,
                saved_at INTEGER DEFAULT (unixepoch())
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS settings (
                key TEXT PRIMARY KEY,
                value TEXT
            )
        """)

def seen(item_id: str) -> bool:
    # Check whether this item was already processed
    with sqlite3.connect(DB_PATH) as conn:
        row = conn.execute(
            "SELECT 1 FROM history WHERE item_id = ?",
            (item_id,),
        ).fetchone()
    return row is not None

def remember(item_id: str):
    # Record completion - the same ID is never stored twice
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT OR IGNORE INTO history (item_id) VALUES (?)",
            (item_id,),
        )
What to Remove Before Publishing

Real operation code should not expose tokens, cookies, account names, internal URLs, or private folder names. Keeping only the problem-solving pattern and small examples reads naturally as site content and is safer too.

Get the SQLite history example (GitHub)