893 lines
32 KiB
Python
893 lines
32 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
설정 관리자 모듈 (Settings Manager)
|
|
사용자 설정을 별도의 SQLite 데이터베이스로 관리합니다.
|
|
|
|
이 모듈은 다음 기능을 제공합니다:
|
|
- 사용자 설정 저장/로드 (SQLite DB)
|
|
- 팀별 필드 설정 관리
|
|
- 마스터 데이터 관리 (편성, 역명, 제조사 등)
|
|
- Supabase 동기화 준비
|
|
|
|
향후 Supabase 통합 시:
|
|
- 로컬 DB는 오프라인 캐시로 사용
|
|
- 온라인 시 Supabase와 동기화
|
|
- 충돌 해결 전략 적용
|
|
"""
|
|
|
|
import sqlite3
|
|
import json
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Optional, List, Dict, Any, Tuple
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass, asdict
|
|
from datetime import datetime
|
|
|
|
from .constants import DATA_DIR, TEAMS
|
|
from .logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
# ============================================================================
|
|
# 데이터 클래스
|
|
# ============================================================================
|
|
|
|
@dataclass
|
|
class FieldSetting:
|
|
"""필드 설정 데이터 클래스"""
|
|
name: str
|
|
visible: bool = True
|
|
width: int = 100
|
|
display_format: Optional[str] = None # "full", "short", "month_day" 등
|
|
|
|
|
|
@dataclass
|
|
class TrainInfo:
|
|
"""편성 정보 데이터 클래스 (마스터 데이터)"""
|
|
train_number: str # 편성번호 (예: "101A")
|
|
train_type: str = "B" # A: 구형, B: 신형
|
|
manufacturer: str = "" # 제조사
|
|
manufacture_year: int = 0 # 제조년도
|
|
depot: str = "" # 소속 차량사업소
|
|
is_active: bool = True
|
|
|
|
|
|
@dataclass
|
|
class StationInfo:
|
|
"""역 정보 데이터 클래스 (마스터 데이터)"""
|
|
station_code: str # 역 코드
|
|
station_name: str # 역명
|
|
line_number: int = 1 # 호선
|
|
order: int = 0 # 순서
|
|
is_active: bool = True
|
|
|
|
|
|
@dataclass
|
|
class ManufacturerInfo:
|
|
"""제조사 정보 데이터 클래스"""
|
|
id: int
|
|
name: str # 제조사명
|
|
|
|
|
|
@dataclass
|
|
class FaultCodeInfo:
|
|
"""고장 코드 정보 데이터 클래스"""
|
|
f_code: str # 고장 코드
|
|
f_code_num: Optional[int] = None # 고장 코드 번호
|
|
f_name: Optional[str] = None # 고장명
|
|
car_type: Optional[str] = None # 차량 타입
|
|
f_class: Optional[str] = None # 고장 분류
|
|
fault_name: Optional[str] = None # 고장 상세명
|
|
grade: Optional[str] = None # 등급
|
|
device: Optional[str] = None # 장치
|
|
fault_detail: Optional[str] = None # 고장 상세
|
|
fault_action: Optional[str] = None # 조치 방법
|
|
manufacturer: Optional[str] = None # 제조사
|
|
|
|
|
|
# ============================================================================
|
|
# SQL 스키마
|
|
# ============================================================================
|
|
|
|
SETTINGS_DB_SCHEMA = """
|
|
-- 사용자 설정 테이블 (키-값 형태)
|
|
CREATE TABLE IF NOT EXISTS user_settings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
team TEXT NOT NULL, -- 운용당무 (1팀, 2팀, 3팀, 4팀)
|
|
category TEXT NOT NULL, -- 설정 카테고리 (field_settings, ui_settings, etc.)
|
|
key TEXT NOT NULL, -- 설정 키
|
|
value TEXT, -- JSON 형태의 값
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(team, category, key)
|
|
);
|
|
|
|
-- 필드 설정 테이블 (섹션별 필드 설정)
|
|
CREATE TABLE IF NOT EXISTS field_settings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
team TEXT NOT NULL, -- 운용당무
|
|
section_name TEXT NOT NULL, -- 섹션 이름 (지시, 고장, 작업, 기타)
|
|
field_name TEXT NOT NULL, -- 필드 이름
|
|
visible INTEGER DEFAULT 1, -- 표시 여부
|
|
width INTEGER DEFAULT 100, -- 너비
|
|
display_format TEXT, -- 표시 형식
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(team, section_name, field_name)
|
|
);
|
|
|
|
-- 편성 정보 테이블 (마스터 데이터)
|
|
CREATE TABLE IF NOT EXISTS train_info (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
train_number TEXT UNIQUE NOT NULL, -- 편성번호
|
|
train_type TEXT DEFAULT 'B', -- A: 구형, B: 신형
|
|
manufacturer TEXT, -- 제조사
|
|
manufacture_year INTEGER, -- 제조년도
|
|
depot TEXT, -- 소속 차량사업소
|
|
is_active INTEGER DEFAULT 1,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
synced_at DATETIME -- Supabase 동기화 시간
|
|
);
|
|
|
|
-- 역 정보 테이블 (마스터 데이터)
|
|
CREATE TABLE IF NOT EXISTS station_info (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
station_code TEXT UNIQUE NOT NULL, -- 역 코드
|
|
station_name TEXT NOT NULL, -- 역명
|
|
line_number INTEGER DEFAULT 1, -- 호선
|
|
"order" INTEGER DEFAULT 0, -- 순서
|
|
is_active INTEGER DEFAULT 1,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
synced_at DATETIME -- Supabase 동기화 시간
|
|
);
|
|
|
|
-- 제조사 정보 테이블 (마스터 데이터)
|
|
CREATE TABLE IF NOT EXISTS manufacturer_info (
|
|
id INTEGER PRIMARY KEY,
|
|
name TEXT UNIQUE NOT NULL, -- 제조사명
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
synced_at DATETIME -- Supabase 동기화 시간
|
|
);
|
|
|
|
-- 고장 코드 테이블 (마스터 데이터)
|
|
CREATE TABLE IF NOT EXISTS fault_code_info (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
f_code TEXT NOT NULL, -- 고장 코드
|
|
f_code_num INTEGER, -- 고장 코드 번호
|
|
f_name TEXT, -- 고장명
|
|
car_type TEXT, -- 차량 타입
|
|
f_class TEXT, -- 고장 분류
|
|
fault_name TEXT, -- 고장 상세명
|
|
grade TEXT, -- 등급
|
|
device TEXT, -- 장치
|
|
fault_detail TEXT, -- 고장 상세
|
|
fault_action TEXT, -- 조치 방법
|
|
manufacturer TEXT, -- 제조사
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
synced_at DATETIME, -- Supabase 동기화 시간
|
|
UNIQUE(f_code, car_type, manufacturer)
|
|
);
|
|
|
|
-- 동기화 상태 테이블
|
|
CREATE TABLE IF NOT EXISTS sync_status (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
table_name TEXT UNIQUE NOT NULL, -- 테이블 이름
|
|
last_synced_at DATETIME, -- 마지막 동기화 시간
|
|
last_sync_hash TEXT, -- 마지막 동기화 해시 (변경 감지용)
|
|
status TEXT DEFAULT 'pending' -- pending, syncing, synced, error
|
|
);
|
|
|
|
-- 인덱스 생성
|
|
CREATE INDEX IF NOT EXISTS idx_user_settings_team ON user_settings(team);
|
|
CREATE INDEX IF NOT EXISTS idx_field_settings_team_section ON field_settings(team, section_name);
|
|
CREATE INDEX IF NOT EXISTS idx_train_info_number ON train_info(train_number);
|
|
CREATE INDEX IF NOT EXISTS idx_station_info_name ON station_info(station_name);
|
|
CREATE INDEX IF NOT EXISTS idx_manufacturer_name ON manufacturer_info(name);
|
|
CREATE INDEX IF NOT EXISTS idx_fault_code ON fault_code_info(f_code);
|
|
CREATE INDEX IF NOT EXISTS idx_fault_code_car_type ON fault_code_info(car_type);
|
|
"""
|
|
|
|
|
|
# ============================================================================
|
|
# 설정 관리자 클래스
|
|
# ============================================================================
|
|
|
|
class SettingsManager:
|
|
"""
|
|
설정 관리자 클래스
|
|
|
|
사용자 설정을 별도의 SQLite 데이터베이스로 관리합니다.
|
|
Supabase 통합을 고려한 구조로 설계되었습니다.
|
|
|
|
Examples:
|
|
>>> settings = SettingsManager()
|
|
>>> settings.save_field_settings("1팀", "고장", [FieldSetting("train_number", True, 80)])
|
|
>>> fields = settings.load_field_settings("1팀", "고장")
|
|
"""
|
|
|
|
_instance: Optional['SettingsManager'] = None
|
|
_lock = threading.Lock()
|
|
|
|
def __new__(cls):
|
|
"""싱글톤 패턴"""
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
cls._instance._initialized = False
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
"""초기화"""
|
|
if self._initialized:
|
|
return
|
|
|
|
self.db_path = DATA_DIR / "settings.db"
|
|
self._local = threading.local()
|
|
|
|
# 데이터 디렉토리 생성
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 데이터베이스 초기화
|
|
self._init_database()
|
|
|
|
self._initialized = True
|
|
logger.info(f"설정 관리자 초기화 완료: {self.db_path}")
|
|
|
|
def _init_database(self):
|
|
"""데이터베이스 초기화"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
conn.executescript(SETTINGS_DB_SCHEMA)
|
|
conn.commit()
|
|
logger.info("설정 데이터베이스 스키마 생성 완료")
|
|
except Exception as e:
|
|
logger.error(f"설정 데이터베이스 초기화 실패: {e}")
|
|
raise
|
|
|
|
@contextmanager
|
|
def get_connection(self):
|
|
"""스레드별 연결 관리"""
|
|
if not hasattr(self._local, 'conn') or self._local.conn is None:
|
|
self._local.conn = sqlite3.connect(
|
|
self.db_path,
|
|
check_same_thread=False,
|
|
detect_types=sqlite3.PARSE_DECLTYPES
|
|
)
|
|
self._local.conn.row_factory = sqlite3.Row
|
|
|
|
try:
|
|
yield self._local.conn
|
|
except Exception as e:
|
|
self._local.conn.rollback()
|
|
raise e
|
|
|
|
def close(self):
|
|
"""연결 종료"""
|
|
if hasattr(self._local, 'conn') and self._local.conn:
|
|
self._local.conn.close()
|
|
self._local.conn = None
|
|
|
|
# ========================================================================
|
|
# 필드 설정 관리
|
|
# ========================================================================
|
|
|
|
def save_field_settings(
|
|
self,
|
|
team: str,
|
|
section_name: str,
|
|
fields: List[FieldSetting]
|
|
):
|
|
"""
|
|
필드 설정 저장
|
|
|
|
Args:
|
|
team: 팀 이름 (예: "1팀")
|
|
section_name: 섹션 이름 (예: "고장")
|
|
fields: 필드 설정 리스트
|
|
"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
for field in fields:
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO field_settings
|
|
(team, section_name, field_name, visible, width, display_format, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
team,
|
|
section_name,
|
|
field.name,
|
|
1 if field.visible else 0,
|
|
field.width,
|
|
field.display_format,
|
|
datetime.now().isoformat()
|
|
))
|
|
|
|
conn.commit()
|
|
logger.info(f"필드 설정 저장 완료: {team} - {section_name} ({len(fields)}개 필드)")
|
|
|
|
except Exception as e:
|
|
logger.error(f"필드 설정 저장 실패: {e}")
|
|
raise
|
|
|
|
def load_field_settings(
|
|
self,
|
|
team: str,
|
|
section_name: str
|
|
) -> Optional[List[FieldSetting]]:
|
|
"""
|
|
필드 설정 로드
|
|
|
|
Args:
|
|
team: 팀 이름
|
|
section_name: 섹션 이름
|
|
|
|
Returns:
|
|
필드 설정 리스트 (없으면 None)
|
|
"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT field_name, visible, width, display_format
|
|
FROM field_settings
|
|
WHERE team = ? AND section_name = ?
|
|
""", (team, section_name))
|
|
|
|
rows = cursor.fetchall()
|
|
if not rows:
|
|
return None
|
|
|
|
fields = []
|
|
for row in rows:
|
|
field = FieldSetting(
|
|
name=row['field_name'],
|
|
visible=bool(row['visible']),
|
|
width=row['width'] or 100,
|
|
display_format=row['display_format']
|
|
)
|
|
fields.append(field)
|
|
|
|
return fields
|
|
|
|
except Exception as e:
|
|
logger.error(f"필드 설정 로드 실패: {e}")
|
|
return None
|
|
|
|
def apply_field_settings_to_fields(
|
|
self,
|
|
team: str,
|
|
section_name: str,
|
|
fields: List[Any] # FieldConfig 리스트
|
|
):
|
|
"""
|
|
저장된 설정을 필드에 적용
|
|
|
|
Args:
|
|
team: 팀 이름
|
|
section_name: 섹션 이름
|
|
fields: FieldConfig 리스트
|
|
"""
|
|
saved_fields = self.load_field_settings(team, section_name)
|
|
if not saved_fields:
|
|
return
|
|
|
|
# 필드 이름으로 매핑
|
|
saved_dict = {f.name: f for f in saved_fields}
|
|
|
|
# 각 필드에 설정 적용
|
|
for field in fields:
|
|
if field.name in saved_dict:
|
|
saved = saved_dict[field.name]
|
|
field.visible = saved.visible
|
|
field.width = saved.width
|
|
if saved.display_format is not None:
|
|
field.display_format = saved.display_format
|
|
|
|
logger.debug(f"필드 설정 적용 완료: {team} - {section_name}")
|
|
|
|
def reset_team_field_settings(self, team: str):
|
|
"""팀별 필드 설정 초기화"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM field_settings WHERE team = ?", (team,))
|
|
conn.commit()
|
|
logger.info(f"팀 필드 설정 초기화: {team}")
|
|
except Exception as e:
|
|
logger.error(f"팀 필드 설정 초기화 실패: {e}")
|
|
|
|
def reset_all_field_settings(self):
|
|
"""모든 필드 설정 초기화"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM field_settings")
|
|
conn.commit()
|
|
logger.info("모든 필드 설정 초기화 완료")
|
|
except Exception as e:
|
|
logger.error(f"모든 필드 설정 초기화 실패: {e}")
|
|
|
|
# ========================================================================
|
|
# 사용자 설정 관리 (일반 키-값)
|
|
# ========================================================================
|
|
|
|
def save_setting(
|
|
self,
|
|
team: str,
|
|
category: str,
|
|
key: str,
|
|
value: Any
|
|
):
|
|
"""
|
|
일반 설정 저장
|
|
|
|
Args:
|
|
team: 팀 이름
|
|
category: 카테고리 (예: "ui_settings", "preferences")
|
|
key: 설정 키
|
|
value: 설정 값 (JSON 직렬화 가능해야 함)
|
|
"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 값을 JSON 문자열로 변환
|
|
json_value = json.dumps(value, ensure_ascii=False)
|
|
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO user_settings
|
|
(team, category, key, value, updated_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (team, category, key, json_value, datetime.now().isoformat()))
|
|
|
|
conn.commit()
|
|
logger.debug(f"설정 저장: {team}/{category}/{key}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"설정 저장 실패: {e}")
|
|
raise
|
|
|
|
def load_setting(
|
|
self,
|
|
team: str,
|
|
category: str,
|
|
key: str,
|
|
default: Any = None
|
|
) -> Any:
|
|
"""
|
|
일반 설정 로드
|
|
|
|
Args:
|
|
team: 팀 이름
|
|
category: 카테고리
|
|
key: 설정 키
|
|
default: 기본값
|
|
|
|
Returns:
|
|
설정 값
|
|
"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT value FROM user_settings
|
|
WHERE team = ? AND category = ? AND key = ?
|
|
""", (team, category, key))
|
|
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return json.loads(row['value'])
|
|
return default
|
|
|
|
except Exception as e:
|
|
logger.error(f"설정 로드 실패: {e}")
|
|
return default
|
|
|
|
def load_settings_by_category(
|
|
self,
|
|
team: str,
|
|
category: str
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
카테고리별 모든 설정 로드
|
|
|
|
Args:
|
|
team: 팀 이름
|
|
category: 카테고리
|
|
|
|
Returns:
|
|
설정 딕셔너리 {key: value}
|
|
"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT key, value FROM user_settings
|
|
WHERE team = ? AND category = ?
|
|
""", (team, category))
|
|
|
|
result = {}
|
|
for row in cursor.fetchall():
|
|
try:
|
|
result[row['key']] = json.loads(row['value'])
|
|
except json.JSONDecodeError:
|
|
result[row['key']] = row['value']
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"카테고리 설정 로드 실패: {e}")
|
|
return {}
|
|
|
|
# ========================================================================
|
|
# 마스터 데이터 관리
|
|
# ========================================================================
|
|
|
|
def save_train_info(self, train: TrainInfo):
|
|
"""편성 정보 저장"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO train_info
|
|
(train_number, train_type, manufacturer, manufacture_year, depot, is_active, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
train.train_number,
|
|
train.train_type,
|
|
train.manufacturer,
|
|
train.manufacture_year,
|
|
train.depot,
|
|
1 if train.is_active else 0,
|
|
datetime.now().isoformat()
|
|
))
|
|
conn.commit()
|
|
except Exception as e:
|
|
logger.error(f"편성 정보 저장 실패: {e}")
|
|
raise
|
|
|
|
def load_train_info(self, train_number: str) -> Optional[TrainInfo]:
|
|
"""편성 정보 로드"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM train_info WHERE train_number = ?
|
|
""", (train_number,))
|
|
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return TrainInfo(
|
|
train_number=row['train_number'],
|
|
train_type=row['train_type'],
|
|
manufacturer=row['manufacturer'] or "",
|
|
manufacture_year=row['manufacture_year'] or 0,
|
|
depot=row['depot'] or "",
|
|
is_active=bool(row['is_active'])
|
|
)
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"편성 정보 로드 실패: {e}")
|
|
return None
|
|
|
|
def get_all_trains(self, active_only: bool = True) -> List[TrainInfo]:
|
|
"""모든 편성 정보 조회"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
if active_only:
|
|
cursor.execute("SELECT * FROM train_info WHERE is_active = 1 ORDER BY train_number")
|
|
else:
|
|
cursor.execute("SELECT * FROM train_info ORDER BY train_number")
|
|
|
|
trains = []
|
|
for row in cursor.fetchall():
|
|
train = TrainInfo(
|
|
train_number=row['train_number'],
|
|
train_type=row['train_type'],
|
|
manufacturer=row['manufacturer'] or "",
|
|
manufacture_year=row['manufacture_year'] or 0,
|
|
depot=row['depot'] or "",
|
|
is_active=bool(row['is_active'])
|
|
)
|
|
trains.append(train)
|
|
|
|
return trains
|
|
except Exception as e:
|
|
logger.error(f"편성 정보 조회 실패: {e}")
|
|
return []
|
|
|
|
def save_station_info(self, station: StationInfo):
|
|
"""역 정보 저장"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO station_info
|
|
(station_code, station_name, line_number, "order", is_active, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
station.station_code,
|
|
station.station_name,
|
|
station.line_number,
|
|
station.order,
|
|
1 if station.is_active else 0,
|
|
datetime.now().isoformat()
|
|
))
|
|
conn.commit()
|
|
except Exception as e:
|
|
logger.error(f"역 정보 저장 실패: {e}")
|
|
raise
|
|
|
|
def get_all_stations(self, line_number: int = None, active_only: bool = True) -> List[StationInfo]:
|
|
"""모든 역 정보 조회"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
query = "SELECT * FROM station_info"
|
|
params = []
|
|
|
|
conditions = []
|
|
if active_only:
|
|
conditions.append("is_active = 1")
|
|
if line_number:
|
|
conditions.append("line_number = ?")
|
|
params.append(line_number)
|
|
|
|
if conditions:
|
|
query += " WHERE " + " AND ".join(conditions)
|
|
|
|
query += ' ORDER BY "order"'
|
|
|
|
cursor.execute(query, params)
|
|
|
|
stations = []
|
|
for row in cursor.fetchall():
|
|
station = StationInfo(
|
|
station_code=row['station_code'],
|
|
station_name=row['station_name'],
|
|
line_number=row['line_number'],
|
|
order=row['order'],
|
|
is_active=bool(row['is_active'])
|
|
)
|
|
stations.append(station)
|
|
|
|
return stations
|
|
except Exception as e:
|
|
logger.error(f"역 정보 조회 실패: {e}")
|
|
return []
|
|
|
|
# ========================================================================
|
|
# 제조사 정보 관리
|
|
# ========================================================================
|
|
|
|
def save_manufacturer_info(self, manufacturer: 'ManufacturerInfo'):
|
|
"""제조사 정보 저장"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO manufacturer_info
|
|
(id, name, updated_at, synced_at)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (
|
|
manufacturer.id,
|
|
manufacturer.name,
|
|
datetime.now().isoformat(),
|
|
datetime.now().isoformat()
|
|
))
|
|
conn.commit()
|
|
except Exception as e:
|
|
logger.error(f"제조사 정보 저장 실패: {e}")
|
|
raise
|
|
|
|
def get_all_manufacturers(self) -> List['ManufacturerInfo']:
|
|
"""모든 제조사 정보 조회"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM manufacturer_info ORDER BY name")
|
|
|
|
manufacturers = []
|
|
for row in cursor.fetchall():
|
|
mfr = ManufacturerInfo(
|
|
id=row['id'],
|
|
name=row['name']
|
|
)
|
|
manufacturers.append(mfr)
|
|
|
|
return manufacturers
|
|
except Exception as e:
|
|
logger.error(f"제조사 정보 조회 실패: {e}")
|
|
return []
|
|
|
|
# ========================================================================
|
|
# 고장 코드 정보 관리
|
|
# ========================================================================
|
|
|
|
def save_fault_code_info(self, fault_code: 'FaultCodeInfo'):
|
|
"""고장 코드 정보 저장"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO fault_code_info
|
|
(f_code, f_code_num, f_name, car_type, f_class, fault_name,
|
|
grade, device, fault_detail, fault_action, manufacturer,
|
|
updated_at, synced_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
fault_code.f_code,
|
|
fault_code.f_code_num,
|
|
fault_code.f_name,
|
|
fault_code.car_type,
|
|
fault_code.f_class,
|
|
fault_code.fault_name,
|
|
fault_code.grade,
|
|
fault_code.device,
|
|
fault_code.fault_detail,
|
|
fault_code.fault_action,
|
|
fault_code.manufacturer,
|
|
datetime.now().isoformat(),
|
|
datetime.now().isoformat()
|
|
))
|
|
conn.commit()
|
|
except Exception as e:
|
|
logger.error(f"고장 코드 정보 저장 실패: {e}")
|
|
raise
|
|
|
|
def get_fault_codes(
|
|
self,
|
|
car_type: str = None,
|
|
manufacturer: str = None
|
|
) -> List['FaultCodeInfo']:
|
|
"""고장 코드 정보 조회"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
query = "SELECT * FROM fault_code_info"
|
|
params = []
|
|
conditions = []
|
|
|
|
if car_type:
|
|
conditions.append("car_type = ?")
|
|
params.append(car_type)
|
|
if manufacturer:
|
|
conditions.append("manufacturer = ?")
|
|
params.append(manufacturer)
|
|
|
|
if conditions:
|
|
query += " WHERE " + " AND ".join(conditions)
|
|
|
|
query += " ORDER BY f_code_num"
|
|
|
|
cursor.execute(query, params)
|
|
|
|
codes = []
|
|
for row in cursor.fetchall():
|
|
code = FaultCodeInfo(
|
|
f_code=row['f_code'],
|
|
f_code_num=row['f_code_num'],
|
|
f_name=row['f_name'],
|
|
car_type=row['car_type'],
|
|
f_class=row['f_class'],
|
|
fault_name=row['fault_name'],
|
|
grade=row['grade'],
|
|
device=row['device'],
|
|
fault_detail=row['fault_detail'],
|
|
fault_action=row['fault_action'],
|
|
manufacturer=row['manufacturer']
|
|
)
|
|
codes.append(code)
|
|
|
|
return codes
|
|
except Exception as e:
|
|
logger.error(f"고장 코드 정보 조회 실패: {e}")
|
|
return []
|
|
|
|
def search_fault_codes(self, keyword: str) -> List['FaultCodeInfo']:
|
|
"""고장 코드 검색"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
search_term = f"%{keyword}%"
|
|
cursor.execute("""
|
|
SELECT * FROM fault_code_info
|
|
WHERE f_code LIKE ? OR f_name LIKE ? OR fault_name LIKE ? OR device LIKE ?
|
|
ORDER BY f_code_num
|
|
""", (search_term, search_term, search_term, search_term))
|
|
|
|
codes = []
|
|
for row in cursor.fetchall():
|
|
code = FaultCodeInfo(
|
|
f_code=row['f_code'],
|
|
f_code_num=row['f_code_num'],
|
|
f_name=row['f_name'],
|
|
car_type=row['car_type'],
|
|
f_class=row['f_class'],
|
|
fault_name=row['fault_name'],
|
|
grade=row['grade'],
|
|
device=row['device'],
|
|
fault_detail=row['fault_detail'],
|
|
fault_action=row['fault_action'],
|
|
manufacturer=row['manufacturer']
|
|
)
|
|
codes.append(code)
|
|
|
|
return codes
|
|
except Exception as e:
|
|
logger.error(f"고장 코드 검색 실패: {e}")
|
|
return []
|
|
|
|
# ========================================================================
|
|
# Supabase 동기화 준비
|
|
# ========================================================================
|
|
|
|
def get_sync_status(self, table_name: str) -> Dict[str, Any]:
|
|
"""동기화 상태 조회"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM sync_status WHERE table_name = ?
|
|
""", (table_name,))
|
|
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return {
|
|
'table_name': row['table_name'],
|
|
'last_synced_at': row['last_synced_at'],
|
|
'last_sync_hash': row['last_sync_hash'],
|
|
'status': row['status']
|
|
}
|
|
return {'table_name': table_name, 'status': 'never_synced'}
|
|
except Exception as e:
|
|
logger.error(f"동기화 상태 조회 실패: {e}")
|
|
return {'table_name': table_name, 'status': 'error'}
|
|
|
|
def update_sync_status(
|
|
self,
|
|
table_name: str,
|
|
status: str = 'synced',
|
|
sync_hash: str = None
|
|
):
|
|
"""동기화 상태 업데이트"""
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO sync_status
|
|
(table_name, last_synced_at, last_sync_hash, status)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (
|
|
table_name,
|
|
datetime.now().isoformat(),
|
|
sync_hash,
|
|
status
|
|
))
|
|
conn.commit()
|
|
except Exception as e:
|
|
logger.error(f"동기화 상태 업데이트 실패: {e}")
|
|
|
|
|
|
# ============================================================================
|
|
# 모듈 레벨 편의 함수
|
|
# ============================================================================
|
|
|
|
def get_settings_manager() -> SettingsManager:
|
|
"""설정 관리자 인스턴스 반환"""
|
|
return SettingsManager()
|
|
|