codeSearch/database/db_manager.py

import sqlite3
from typing import List, Dict, Optional

class DatabaseManager:
    """데이터베이스 관리 클래스 - 조회 전용"""
    
    def __init__(self, db_path: str = "fault_codes.db"):
        """
        데이터베이스 매니저 초기화
        Args:
            db_path (str): 데이터베이스 파일 경로
        """
        self.db_path = db_path
        self.conn = None
    
    def _get_connection(self):
        """데이터베이스 연결을 반환합니다."""
        return sqlite3.connect(self.db_path)

    def get_manufacturers(self) -> List[str]:
        """제작사 목록을 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT name FROM manufacturers ORDER BY name")
        manufacturers = [row[0] for row in cursor.fetchall()]
        conn.close()
        return manufacturers

    def get_fault_types(self) -> List[str]:
        """고장 타입 목록을 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT DISTINCT fault_type FROM fault_code_list WHERE fault_type IS NOT NULL ORDER BY fault_type")
        types = [row[0] for row in cursor.fetchall()]
        conn.close()
        return types

    def get_fault_codes(self, manufacturer: Optional[str] = None, 
                       fault_type: Optional[str] = None) -> List[Dict]:
        """고장 코드 목록을 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        
        query = """
            SELECT f.fault_code, f.fault_name, f.fault_type, m.name as manufacturer
            FROM fault_code_list f
            LEFT JOIN manufacturers m ON f.manufacturer_id = m.id
            WHERE f.manufacturer_id IS NOT NULL
        """
        params = []
        
        if manufacturer:
            query += " AND m.name = ?"
            params.append(manufacturer)
        if fault_type:
            query += " AND f.fault_type = ?"
            params.append(fault_type)
            
        query += " ORDER BY f.fault_code"
        
        cursor.execute(query, params)
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'code': row[0],
                'name': row[1],
                'type': row[2],
                'manufacturer': row[3]
            }
            for row in results
        ]

    def search_fault_codes(self, search_term: str) -> List[Dict]:
        """고장 코드를 검색합니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        
        query = """
            SELECT f.fault_code, f.fault_name, f.fault_type, m.name as manufacturer
            FROM fault_code_list f
            LEFT JOIN manufacturers m ON f.manufacturer_id = m.id
            WHERE f.manufacturer_id IS NOT NULL
              AND (f.fault_code LIKE ? OR f.fault_name LIKE ?)
            ORDER BY f.fault_code
        """
        
        cursor.execute(query, (f'%{search_term}%', f'%{search_term}%'))
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'code': row[0],
                'name': row[1],
                'type': row[2],
                'manufacturer': row[3]
            }
            for row in results
        ]

    def get_abbreviations(self) -> List[Dict]:
        """약어 목록을 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT abbreviation, description FROM abbreviations ORDER BY abbreviation")
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'abbreviation': row[0],
                'description': row[1]
            }
            for row in results
        ]

    def get_drawings(self) -> List[Dict]:
        """도면 목록을 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT drawing_code, drawing_name, related_device FROM drawings ORDER BY drawing_code")
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'code': row[0],
                'name': row[1],
                'device': row[2]
            }
            for row in results
        ]

    def get_signals(self) -> List[Dict]:
        """신호 목록을 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT signal_code, signal_name, signal_type FROM signals ORDER BY signal_code")
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'code': row[0],
                'name': row[1],
                'type': row[2]
            }
            for row in results
        ]

    def get_fault_detail(self, fault_code: str) -> Optional[Dict]:
        """특정 고장 코드의 상세 정보를 가져옵니다."""
        conn = self._get_connection()
        cursor = conn.cursor()
        
        query = """
            SELECT f.fault_code, f.fault_name, f.fault_level, f.fault_type, 
                  f.fault_detail, f.reaction, f.detect_condition, f.clear_condition, 
                  f.action, f.drawing, m.name as manufacturer
            FROM fault_code_list f
            LEFT JOIN manufacturers m ON f.manufacturer_id = m.id
            WHERE f.fault_code = ? AND f.manufacturer_id IS NOT NULL
        """
        
        cursor.execute(query, (fault_code,))
        result = cursor.fetchone()
        conn.close()
        
        if not result:
            return None
            
        return {
            'code': result[0],
            'name': result[1],
            'level': result[2],
            'type': result[3],
            'detail': result[4],
            'reaction': result[5],
            'detect_condition': result[6],
            'clear_condition': result[7],
            'action': result[8],
            'drawing': result[9],
            'manufacturer': result[10]
        }

    def __del__(self):
        """데이터베이스 연결을 종료합니다."""
        if hasattr(self, 'conn') and self.conn:
            self.conn.close()