Python - CSV파일 감지 및 데이터베이스 자동 업데이트

PYTHON

watchdog, sqlalchemy 이용

 

Watchdog - 파일 및 디렉터리 변경 사항 실시간 감지

Sqlalchemy - 데이터베이스 ORM 및 쿼리 실행

 

경로 설정

# 감시할 CSV 파일 경로
CSV_FILE_PATH = "csv_files/quiz_data.csv"
observer = None  # 감시 객체 전역 변수

 

데이터베이스가 비어 있는지 확인하는 함수

CSV 파일을 감시하는 역할

# 데이터베이스가 비어 있는지 확인하는 함수
def is_db_empty():
    """데이터베이스에 데이터가 있는지 확인"""
    with SessionLocal() as session:
        try:
            result = session.execute(text("SELECT COUNT(*) FROM quizzes")).fetchone()
            return result[0] == 0 if result else True
        except Exception as e:
            print(f"🚨 DB 확인 중 오류 발생: {str(e)}")
            return True  # 오류 발생 시 데이터를 저장하도록 처리

 

 

CSV 파일을 읽고 데이터베이스에 저장하는 함수

# CSV 파일을 읽어 데이터베이스에 저장하는 함수
def store_csv_to_db(CSV_FILE_PATH):
    if not os.path.exists(CSV_FILE_PATH):
        print(f"CSV 파일을 찾을 수 없음: {CSV_FILE_PATH}")
        return

    try:
        with SessionLocal() as session:
            with open(CSV_FILE_PATH, 'r', encoding='utf-8') as file:
                csv_reader = csv.reader(file)
                headers = next(csv_reader, None)

                if not headers or len(headers) < 5:
                    print(f"CSV 형식이 잘못됨: {CSV_FILE_PATH}")
                    return

                for row_number, row in enumerate(csv_reader, start=2):
                    if not any(row):
                        print(f"⚠ [행 {row_number}] 빈 행 건너뜀")
                        continue

                    if len(row) < 5:
                        print(f"⚠ [행 {row_number}] 잘못된 데이터 행 건너뜀: {row}")
                        continue

                    try:
                        quiz_id = int(row[0])
                        question = row[1]
                        explanation = row[2]
                        answer = str(row[3])
                        category = row[4]  # 카테고리 추가

                        # SQLite와 PostgreSQL에서 각각 다르게 처리
                        if "sqlite" in config.DATABASE_URL:
                            session.execute(text('''
                                INSERT OR REPLACE INTO quizzes (id, question, explanation, answer, category)
                                VALUES (:id, :question, :explanation, :answer, :category)
                            '''), {
                                "id": quiz_id, "question": question,
                                "explanation": explanation, "answer": answer, "category": category
                            })
                        else:  # PostgreSQL
                            session.execute(text('''
                                INSERT INTO quizzes (id, question, explanation, answer, category)
                                VALUES (:id, :question, :explanation, :answer, :category)
                                ON CONFLICT (id) DO UPDATE 
                                SET question=EXCLUDED.question, 
                                    explanation=EXCLUDED.explanation, 
                                    answer=EXCLUDED.answer, 
                                    category=EXCLUDED.category;
                            '''), {
                                "id": quiz_id, "question": question,
                                "explanation": explanation, "answer": answer, "category": category
                            })

                    except Exception as e:
                        print(f"❌ [행 {row_number}] 데이터 변환 오류: {row}, 오류: {str(e)}")
                        continue

            session.commit()
        print(f"✅ CSV 데이터 저장 완료!")
    except Exception as e:
        print(f"🚨 CSV 처리 중 오류 발생: {str(e)}")

SessionLocal - 데이터베이스와 연결하는 sqlalchemy 세션 객체

config.DATABASE_URL - 데이터베이스 URL을 포함한 설정

 

데이터베이스 URL이 SQLite인지 PostgreSQL인지 확인하고

각 데이터베이스에 맞는 쿼리를 실행

CSV 감지 함수

# 리스너 클래스 정의
class CsvFileListener(FileSystemEventHandler):
    def __init__(self, CSV_FILE_PATH):
        self.CSV_FILE_PATH = os.path.abspath(CSV_FILE_PATH)

    def on_modified(self, event):
        if event.is_directory:
            return
        if event.src_path == self.CSV_FILE_PATH:
            print(f"📂 CSV 파일 {event.src_path} 가 수정되었습니다.")
            print("🔄 DB에 저장을 시작합니다.")
            store_csv_to_db(event.src_path)

FileSystemEventHandler 상속
CsvFileListener 클래스는 파일 변경 이벤트를 감지하기 위해
Watchdog 라이브러리의 FileSystemEventHandler를 상속

Watchdog 라이브러리에서 제공하는 이벤트 핸들러 메서드(예: on_modified)를
오버라이딩하여 원하는 동작을 정의