파이썬 파일입출력 처리와 디렉터리 감시 자동화 방법

# 파일 쓰기 예제
with open("example.txt", "w", encoding="utf-8") as f:
    f.write("안녕하세요, 파일 입출력 예제입니다.")

# 파일 읽기 예제
with open("example.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)

pip install watchdog

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            print(f"새 파일이 생성되었습니다: {event.src_path}")

if __name__ == "__main__":
    path = "./watch_folder"
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

processed_files = set()

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            filename = event.src_path
            if filename not in processed_files:
                print(f"신규 파일 처리: {filename}")
                processed_files.add(filename)

import os
import time
import queue
import threading
from dataclasses import dataclass, field
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

MAX_RETRIES = 5

@dataclass(order=True)
class RetryItem:
    ready_at: float
    path: str = field(compare=False)
    attempt: int = field(default=0, compare=False)
    reason: str = field(default="", compare=False)

retry_q = queue.PriorityQueue()  # (ready_at 기준 우선순위)
dead_letters = []
in_flight = set()               # 중복 처리 방지

def process_file(path: str):
    # 예시 처리 함수: 파일 크기가 0이면 실패로 가정
    size = os.path.getsize(path)
    if size == 0:
        raise IOError("empty file")
    # 실제 처리 로직 자리
    print(f"처리 완료: {path} ({size} bytes)")

def schedule_retry(path: str, attempt: int, reason: str):
    delay = min(2 ** attempt, 60)  # 지수 백오프, 상한 60초
    ready_at = time.time() + delay
    retry_q.put(RetryItem(ready_at=ready_at, path=path, attempt=attempt, reason=reason))
    print(f"재시도 예약: {path} {attempt}회차, {delay}s 후, 사유={reason}")

def worker():
    while True:
        try:
            item = retry_q.get(timeout=1)
        except queue.Empty:
            continue
        now = time.time()
        if item.ready_at > now:
            time.sleep(item.ready_at - now)
        try:
            process_file(item.path)
            in_flight.discard(item.path)
        except Exception as e:
            if item.attempt + 1 >= MAX_RETRIES:
                dead_letters.append((item.path, str(e)))
                in_flight.discard(item.path)
                print(f"데드 레터 이동: {item.path}, 이유={e}")
            else:
                schedule_retry(item.path, item.attempt + 1, str(e))
        finally:
            retry_q.task_done()

class Handler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        path = event.src_path
        if path in in_flight:
            return
        in_flight.add(path)
        try:
            process_file(path)
            in_flight.discard(path)
        except Exception as e:
            schedule_retry(path, 1, str(e))

if __name__ == "__main__":
    # 워커 스레드 시작
    t = threading.Thread(target=worker, daemon=True)
    t.start()

    watch_dir = "./watch_folder"
    os.makedirs(watch_dir, exist_ok=True)
    observer = Observer()
    observer.schedule(Handler(), watch_dir, recursive=False)
    observer.start()
    print("감시 시작")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

import os, time, queue, threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

processed = set()
retry_q = queue.PriorityQueue()

def process_file(path):
    if os.path.getsize(path) == 0:
        raise IOError("파일 크기 0, 실패")
    print(f"[처리 성공] {path}")

def schedule_retry(path, attempt):
    delay = min(2**attempt, 60)
    retry_q.put((time.time() + delay, path, attempt))
    print(f"[재시도 예약] {path}, {attempt}회차, {delay}초 후")

def retry_worker():
    while True:
        try:
            ready_at, path, attempt = retry_q.get(timeout=1)
        except queue.Empty:
            continue
        if ready_at > time.time():
            time.sleep(ready_at - time.time())
        try:
            process_file(path)
        except Exception as e:
            if attempt >= 5:
                print(f"[데드 레터 이동] {path}, 이유={e}")
            else:
                schedule_retry(path, attempt + 1)
        retry_q.task_done()

class Handler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory: return
        if event.src_path in processed: return
        processed.add(event.src_path)
        try:
            process_file(event.src_path)
        except Exception:
            schedule_retry(event.src_path, 1)

if __name__ == "__main__":
    threading.Thread(target=retry_worker, daemon=True).start()
    folder = "./watch_folder"
    os.makedirs(folder, exist_ok=True)
    obs = Observer()
    obs.schedule(Handler(), folder, recursive=False)
    obs.start()
    print("[감시 시작]")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        obs.stop()
    obs.join()