341 lines
10 KiB
Python
341 lines
10 KiB
Python
# -*- 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
|
|
|
|
|