SameOS ~/tools/automation-lessons.md

한국어 English

🌙☀️

Python 자동화 코드 강의: 중복 제거, 주기 실행, SQLite 기록

운영 자동화 코드는 그대로 공개하기보다 핵심 패턴만 떼어 설명할 때 더 오래 쓸 수 있습니다. 이 문서는 파일 정리, 반복 작업, 처리 이력 기록에서 자주 쓰는 구조를 작은 예제 세 개로 나눕니다. 모두 파이썬 코드이며, 파일로 저장한 뒤 터미널(명령어를 입력하는 검은 창)에서 python3 스크립트명.py 로 실행합니다.

from pathlib import Path
import hashlib

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

def file_hash(path: Path) -> str:
    # 파일을 1MB씩 나눠 읽어 SHA-256 해시(내용 지문)를 계산
    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  # 썸네일 폴더는 건너뜀
        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  # (기준 파일, 중복 후보)
        else:
            seen[digest] = path  # 처음 본 해시는 기준 파일로 기억

import asyncio

running = False

async def scan_once():
    # 느린 네트워크·디스크 작업은 여기에 둡니다.
    return await asyncio.to_thread(do_blocking_scan)

async def scheduler(interval_seconds: int):
    global running
    while True:
        if not running:  # 이전 작업이 아직 실행 중이면 이번 회차는 건너뜀
            running = True
            try:
                results = await scan_once()
                await send_results(results)
            finally:
                running = False  # 예외가 나도 플래그를 반드시 내림
        await asyncio.sleep(interval_seconds)

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:
    # 이미 처리한 항목인지 확인
    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):
    # 처리 완료를 기록 - 같은 ID는 중복 저장되지 않음
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT OR IGNORE INTO history (item_id) VALUES (?)",
            (item_id,),
        )