codeSearch/test/excel_to_sqlite.py

from typing import Dict, List
import pandas as pd
import sqlite3
import uuid


def create_fault_codes_table():
    """우진200 차량의 고장 코드 테이블을 생성합니다."""
    conn = sqlite3.connect('fault_codes.db')
    cursor = conn.cursor()
    
    # 우진200 테이블 생성
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS fault_codes (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        code_id TEXT UNIQUE,  -- UUID
        fault_code TEXT,      -- 고장코드
        fault_name TEXT,      -- 고장명
        fault_level TEXT,     -- 등급
        fault_type TEXT,      -- 분류
        fault_detail TEXT,    -- 고장내용
        reaction TEXT,        -- 반응
        detect_condition TEXT, -- 검지조건
        clear_condition TEXT, -- 소거조건
        action TEXT          -- 조치방법
    )
    ''')
    
    conn.commit()
    conn.close()

def create_rotem200_table():
    """로템200 차량의 고장 코드 테이블을 생성합니다."""
    conn = sqlite3.connect('fault_codes.db')
    cursor = conn.cursor()
    
    # 로템200 테이블 생성
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS rotem200 (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        code_id TEXT UNIQUE,  -- UUID
        fault_code TEXT,      -- 고장코드
        fault_name TEXT,      -- 고장명
        fault_level TEXT,     -- 등급
        fault_type TEXT,      -- 분류
        fault_detail TEXT,    -- 고장내용
        reaction TEXT,        -- 반응
        detect_condition TEXT, -- 검지조건
        clear_condition TEXT, -- 소거조건
        action TEXT          -- 조치방법
    )
    ''')
    
    conn.commit()
    conn.close()

def import_excel_to_sqlite():
    """엑셀 파일에서 고장 코드 데이터를 읽어 SQLite 데이터베이스에 저장합니다."""
    try:
        # 엑셀 파일 읽기
        df = pd.read_excel(
            '우진200백데이터.xlsx',
            sheet_name='200량 고장해설',
            header=0
        )
        
        print("엑셀 파일의 열 이름:")
        print(df.columns.tolist())
        
        # UUID 생성
        df['code_id'] = [str(uuid.uuid4()) for _ in range(len(df))]
        
        # 열 이름 매핑
        column_mapping = {
            '고장코드': 'fault_code',
            '고장명': 'fault_name',
            '등급': 'fault_level',
            '분류': 'fault_type',
            '고장내용': 'fault_detail',
            '반응': 'reaction',
            '검지조건': 'detect_condition',
            '소거조건': 'clear_condition',
            '조치방법': 'action'
        }
        
        # 열 이름 변경
        df = df.rename(columns=column_mapping)
        
        # SQLite 연결
        conn = sqlite3.connect('fault_codes.db')
        
        # 데이터프레임을 SQLite 테이블에 저장
        df.to_sql('fault_codes', conn, if_exists='replace', index=False)
        
        # 변경사항 저장 및 연결 종료
        conn.commit()
        conn.close()
        
        print("데이터가 성공적으로 SQLite 데이터베이스에 저장되었습니다.")
        
        # 저장된 데이터 확인
        conn = sqlite3.connect('fault_codes.db')
        cursor = conn.cursor()
        cursor.execute("SELECT COUNT(*) FROM fault_codes")
        count = cursor.fetchone()[0]
        print(f"총 {count}개의 레코드가 저장되었습니다.")
        
        # 샘플 데이터 출력
        cursor.execute("SELECT * FROM fault_codes LIMIT 5")
        sample_data = cursor.fetchall()
        print("\n샘플 데이터:")
        for row in sample_data:
            print(row)
            
        conn.close()
        
    except (FileNotFoundError, pd.errors.EmptyDataError) as e:
        print(f"엑셀 파일 오류: {str(e)}")
    except sqlite3.Error as e:
        print(f"데이터베이스 오류: {str(e)}")

def import_excel_to_rotem200():
    """엑셀 파일에서 고장 코드 데이터를 읽어 SQLite 데이터베이스에 저장합니다."""
    try:
        # 엑셀 파일 읽기
        df = pd.read_excel(
            '로템200백데이터.xlsx',
            sheet_name='200량 고장해설',
            header=0
        )
        
        print("엑셀 파일의 열 이름:")
        print(df.columns.tolist())
        
        # UUID 생성
        df['code_id'] = [str(uuid.uuid4()) for _ in range(len(df))]
        
        # 열 이름 매핑
        column_mapping = {
            '고장코드': 'fault_code',
            '고장명': 'fault_name',
            '등급': 'fault_level',
            '분류': 'fault_type',
            '고장내용': 'fault_detail',
            '반응': 'reaction',
            '검지조건': 'detect_condition',
            '소거조건': 'clear_condition',
            '조치방법': 'action'
        }
        
        # 열 이름 변경
        df = df.rename(columns=column_mapping)
        
        # SQLite 연결
        conn = sqlite3.connect('fault_codes.db')
        
        # 데이터프레임을 SQLite 테이블에 저장
        df.to_sql('rotem200', conn, if_exists='replace', index=False)
        
        # 변경사항 저장 및 연결 종료
        conn.commit()
        conn.close()
        
        print("데이터가 성공적으로 SQLite 데이터베이스에 저장되었습니다.")
        
        # 저장된 데이터 확인
        conn = sqlite3.connect('fault_codes.db')
        cursor = conn.cursor()
        cursor.execute("SELECT COUNT(*) FROM rotem200")
        count = cursor.fetchone()[0]
        print(f"총 {count}개의 레코드가 저장되었습니다.")
        
        # 샘플 데이터 출력
        cursor.execute("SELECT * FROM rotem200 LIMIT 5")
        sample_data = cursor.fetchall()
        print("\n샘플 데이터:")
        for row in sample_data:
            print(row)
            
        conn.close()
        
    except (FileNotFoundError, pd.errors.EmptyDataError) as e:
        print(f"엑셀 파일 오류: {str(e)}")
    except sqlite3.Error as e:
        print(f"데이터베이스 오류: {str(e)}")

if __name__ == "__main__":
    create_fault_codes_table()
    create_rotem200_table()
    import_excel_to_sqlite()
    import_excel_to_rotem200()