# -*- coding: utf-8 -*- """ 열차 정보 파싱 모듈 열번과 역명에서 발생 시간을 유추하고, 입력된 정보를 파싱하는 기능을 제공합니다. 기능: - 열번에서 열차 종별, 방향 추출 - 열번과 역명으로 발생 시간 추정 - 고장 정보 자동 파싱 """ import re from datetime import date, time, datetime from typing import Optional, Dict, Tuple, List, Any from dataclasses import dataclass from core.logger import get_logger logger = get_logger(__name__) @dataclass class ColumnNumberInfo: """열번 정보""" raw: str # 원본 열번 train_type: str # 열차 종별 (정기, 회송, 시운전, 구간, 임시) direction: str # 방향 (상행, 하행) sequence: int # 순번 is_valid: bool # 유효 여부 @dataclass class ParsedFaultInfo: """파싱된 고장 정보""" occurrence_date: Optional[date] = None occurrence_time: Optional[time] = None column_number: str = "" train_number: str = "" car_number: str = "" occurrence_station: str = "" device_category: str = "" fault_code: str = "" fault_content: str = "" action_content: str = "" fault_source: str = "" raw_text: str = "" class TrainParser: """ 열차 정보 파싱 클래스 열번에서 열차 종별과 방향을 추출하고, 열번과 역명으로 발생 시간을 추정합니다. """ # 1호선 열차 종별 (1000자리) TRAIN_TYPES = { 1: ("정기", "up"), # 상행 정기 2: ("정기", "down"), # 하행 정기 3: ("회송", "up"), # 상행 회송 4: ("회송", "down"), # 하행 회송 5: ("시운전", "up"), # 상행 시운전 6: ("시운전", "down"), # 하행 시운전 7: ("구간", "up"), # 상행 구간 8: ("구간", "down"), # 하행 구간 9: ("임시", "both"), # 임시 } # 방향 판단 (1자리: 홀수=상행, 짝수=하행) @staticmethod def get_direction_from_last_digit(digit: int) -> str: """마지막 자리로 방향 판단""" return "up" if digit % 2 == 1 else "down" def __init__(self): """초기화""" self._crud = None @property def crud(self): """CRUD 매니저 (지연 로딩)""" if self._crud is None: from database.crud import get_crud self._crud = get_crud() return self._crud def parse_column_number(self, column_number: str) -> ColumnNumberInfo: """ 열번 파싱 Args: column_number: 열번 (예: "1001", "2034", "3511") Returns: ColumnNumberInfo: 열번 정보 Examples: >>> parser = TrainParser() >>> info = parser.parse_column_number("1001") >>> print(info.train_type, info.direction) 정기 상행 """ column_number = column_number.strip() # 4자리 숫자 검증 if not column_number or not column_number.isdigit(): return ColumnNumberInfo( raw=column_number, train_type="", direction="", sequence=0, is_valid=False ) if len(column_number) != 4: return ColumnNumberInfo( raw=column_number, train_type="", direction="", sequence=0, is_valid=False ) # 각 자리 추출 d1000 = int(column_number[0]) # 1000자리: 열차 종별 d100 = int(column_number[1]) # 100자리 d10 = int(column_number[2]) # 10자리 d1 = int(column_number[3]) # 1자리: 방향 판단 # 열차 종별 train_type_info = self.TRAIN_TYPES.get(d1000, ("알수없음", "unknown")) train_type = train_type_info[0] # 방향 (1자리 기준) direction = self.get_direction_from_last_digit(d1) direction_text = "상행" if direction == "up" else "하행" # 순번 (100자리 + 10자리) sequence = d100 * 10 + d10 return ColumnNumberInfo( raw=column_number, train_type=train_type, direction=direction_text, sequence=sequence, is_valid=True ) def estimate_time( self, column_number: str, station: str, occurrence_date: Optional[date] = None ) -> Optional[time]: """ 열번과 역명으로 발생 시간 추정 Args: column_number: 열번 station: 역명 occurrence_date: 발생일 (평일/주말 판단용) Returns: 추정 시간 또는 None """ if not column_number or not station: return None try: return self.crud.estimate_time_by_column_station( column_number, station, occurrence_date ) except Exception as e: logger.warning(f"시간 추정 실패: {column_number}, {station} - {e}") return None def parse_train_number(self, train_number: str) -> Tuple[int, str]: """ 편성번호 파싱 Args: train_number: 편성번호 (예: "132B", "101A") Returns: (숫자 부분, 접미사) 튜플 Examples: >>> parser = TrainParser() >>> num, suffix = parser.parse_train_number("132B") >>> print(num, suffix) 132 B """ train_number = train_number.strip().upper() if not train_number: return 0, "" # 숫자와 접미사 분리 match = re.match(r'^(\d+)([AB])?$', train_number) if match: num = int(match.group(1)) suffix = match.group(2) or "" return num, suffix return 0, "" def format_train_number(self, num: int, suffix: str = "") -> str: """ 편성번호 포맷팅 Args: num: 숫자 부분 (예: 132) suffix: 접미사 (예: "A", "B") Returns: 포맷된 편성번호 (예: "132B") """ if num <= 0: return "" if suffix: return f"{num}{suffix}" return str(num) def get_train_display(self, train_number: str) -> str: """ 편성번호 표시용 텍스트 반환 테이블에서 중간 2자리만 표시할 때 사용 Args: train_number: 편성번호 (예: "132B") Returns: 표시용 텍스트 (예: "32") """ if not train_number: return "" # 숫자 부분에서 마지막 2자리 추출 match = re.match(r'^\d*(\d{2})[AB]?$', train_number.strip()) if match: return match.group(1) return train_number def parse_fault_text(self, text: str) -> ParsedFaultInfo: """ 고장 텍스트 자동 파싱 복사/붙여넣기 된 고장 정보를 파싱하여 각 필드로 분리합니다. Args: text: 고장 정보 텍스트 Returns: ParsedFaultInfo: 파싱된 고장 정보 """ result = ParsedFaultInfo(raw_text=text) if not text: return result lines = text.strip().split('\n') # 날짜 패턴 찾기 date_pattern = r'(\d{4}[-./]\d{2}[-./]\d{2}|\d{2}[-./]\d{2}[-./]\d{2})' time_pattern = r'(\d{2}:\d{2}(?::\d{2})?)' for line in lines: # 날짜 추출 date_match = re.search(date_pattern, line) if date_match and not result.occurrence_date: date_str = date_match.group(1).replace('/', '-').replace('.', '-') try: if len(date_str) == 8: # YY-MM-DD result.occurrence_date = datetime.strptime(date_str, "%y-%m-%d").date() else: # YYYY-MM-DD result.occurrence_date = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: pass # 시간 추출 time_match = re.search(time_pattern, line) if time_match and not result.occurrence_time: time_str = time_match.group(1) try: if len(time_str) == 5: # HH:MM result.occurrence_time = datetime.strptime(time_str, "%H:%M").time() else: # HH:MM:SS result.occurrence_time = datetime.strptime(time_str, "%H:%M:%S").time() except ValueError: pass # 편성번호 추출 (예: 132B, 101A) train_match = re.search(r'\b(1\d{2}[AB])\b', line) if train_match and not result.train_number: result.train_number = train_match.group(1) # 열번 추출 (4자리 숫자) column_match = re.search(r'\b([1-9]\d{3})\b', line) if column_match and not result.column_number: result.column_number = column_match.group(1) # 호차 추출 (1~8) car_match = re.search(r'\b([1-8])호차?\b', line) if car_match and not result.car_number: result.car_number = car_match.group(1) # 역명 추출 (xx역) station_match = re.search(r'([가-힣]+역)', line) if station_match and not result.occurrence_station: result.occurrence_station = station_match.group(1) # 나머지 텍스트는 고장내용으로 if not result.fault_content and lines: result.fault_content = '\n'.join(lines) return result # 싱글톤 인스턴스 _parser_instance: Optional[TrainParser] = None def get_train_parser() -> TrainParser: """ TrainParser 싱글톤 인스턴스 반환 Returns: TrainParser 인스턴스 """ global _parser_instance if _parser_instance is None: _parser_instance = TrainParser() return _parser_instance