VOC_Monitor/test/test_voc.py

238 lines
9.1 KiB
Python

import win32com.client as win32
import os
import winreg
import re
import pandas as pd
from pathlib import Path
from datetime import datetime
# --- 상수: 부산 1호선 역사 데이터 ---
# 순서: 다대포해수욕장(0) -> 노포(39)
# 하선(노포행): 인덱스 증가 방향 / 상선(다대포행): 인덱스 감소 방향
BUSAN_L1_STATIONS = [
"다대포해수욕장", "다대포항", "낫개", "신장림", "장림", "동매", "신평", "하단", "당리", "사하",
"괴정", "대티", "서대신", "동대신", "토성", "자갈치", "남포", "중앙", "부산역", "초량",
"부산진", "좌천", "범일", "범내골", "서면", "부전", "양정", "시청", "연산", "교대",
"동래", "명륜", "온천장", "부산대", "장전", "구서", "두실", "남산", "범어사", "노포"
]
# --- 유틸리티 ---
def setup_hwp_security():
"""한글 자동화 보안 모듈 등록"""
try:
base_dir = Path(__file__).resolve().parent
dll_path = base_dir / "FilePathCheck.dll"
reg_path = r"Software\HNC\HwpAutomation\Modules"
with winreg.CreateKey(winreg.HKEY_CURRENT_USER, reg_path) as key:
winreg.SetValueEx(key, "FilePathCheckDLL", 0, winreg.REG_SZ, str(dll_path))
return True
except Exception: return False
def get_clean_number(text):
return re.sub(r'[^0-9]', '', text) if text else ""
def get_korean_weekday(date_str):
try:
dt = datetime.strptime(date_str.strip('.'), "%Y.%m.%d")
return "월화수목금토일"[dt.weekday()]
except Exception: return ""
# --- 핵심 로직: 정보 유추 및 분석 ---
def infer_info_from_context(full_text):
"""
본문 문맥을 분석하여 '호선', '방향' 정보를 유추합니다.
"""
inferred_line = ""
inferred_dir = ""
# 1. 역명 검색으로 호선 유추
for station in BUSAN_L1_STATIONS:
if station in full_text:
inferred_line = "1호선"
break
# 2. 방향성 유추
# Case A: "XX방향" 직접 언급
direction_match = re.search(r"(\w+역?)\s*방향", full_text)
if direction_match:
dir_word = direction_match.group(1)
if "노포" in dir_word: inferred_dir = "노포행(하선)"
elif "다대포" in dir_word: inferred_dir = "다대포행(상선)"
# Case B: "A역에서 타서 B역에서 내림" 이동 경로 분석
ride_match = re.search(r"(\w+역?)\s*에서\s*타서\s*(\w+역?)\s*에서\s*내렸", full_text)
if ride_match and not inferred_dir:
start_st = ride_match.group(1).replace("", "")
end_st = ride_match.group(2).replace("", "")
if start_st in BUSAN_L1_STATIONS and end_st in BUSAN_L1_STATIONS:
start_idx = BUSAN_L1_STATIONS.index(start_st)
end_idx = BUSAN_L1_STATIONS.index(end_st)
# 인덱스 증가 -> 노포행(하선), 감소 -> 다대포행(상선)
inferred_dir = "노포행(하선)" if start_idx < end_idx else "다대포행(상선)"
return inferred_line, inferred_dir
def classify_train_details(train_str, context_dir=""):
"""
열차번호를 분석합니다. 열차번호가 없으면 문맥에서 유추된 방향(context_dir)을 사용합니다.
"""
train_num = get_clean_number(train_str)
# 열차번호가 없는 경우 -> 문맥 방향 사용
if not train_num:
return {"num": "", "type": "미확인", "direction": context_dir or "미확인"}
val = int(train_num)
last_digit = val % 10
# 1. 방향 판별 (열차번호 규칙 우선)
direction = "다대포행(상선)" if last_digit % 2 == 1 else "노포행(하선)"
# 2. 열차 종별 판별
type_map = {(1, 2): "영업", (3, 4): "회송", (5, 6): "시운전", (7, 8): "단기", (9,): "임시"}
t_type = next((v for k, v in type_map.items() if last_digit in k), "미확인")
# 예외: 임시열차(9)는 보통 상선으로 간주 (규칙에 따라 조정 가능)
if last_digit == 9: direction = "다대포행(상선)"
return {"num": str(val), "type": t_type, "direction": direction}
def parse_voc_info(title, content):
"""VOC 내용 전체를 분석하여 구조화된 정보를 반환합니다."""
full_text = f"{title} {content}"
# 1. 기본 정보 정규식 추출
patterns = {
"line": r"(\d+)\s*호선",
"set": r"(\d+)\s*편성",
"car": r"(\d+)\s*호차",
"train": r"(?:제)?\s*(\d{4})\s*열차"
}
info = {k: (re.search(p, full_text).group(0) if re.search(p, full_text) else "")
for k, p in patterns.items()}
# 2. 문맥 기반 정보 유추 (호선, 방향)
inf_line, inf_dir = infer_info_from_context(full_text)
# 명시된 호선이 없으면 유추된 호선 적용
if not info["line"]: info["line"] = inf_line
# 3. 열차 상세 정보 분석 (문맥 방향 정보 전달)
train_details = classify_train_details(info["train"], inf_dir)
# 4. 파일명용 제목 정제
formatted_title = title
for pat in [r'\d+\s*호선', r'\d+\s*편성', r'\d+\s*호차', r'제?\d{4}열차', r'[,]']:
formatted_title = re.sub(pat, '', formatted_title)
pure_title = re.sub(r'\s+', ' ', formatted_title).strip()
return {
"line_str": info["line"],
"set_str": info["set"],
"car_str": info["car"],
"train_str": info["train"],
"pure_title": pure_title,
"train_details": train_details
}
def get_schedule_remarks(train_num, parquet_path="train_schedule.parquet"):
"""Parquet 시각표 데이터 조회 (모의 구현)"""
if not train_num or not os.path.exists(parquet_path):
return "정보 없음"
try:
df = pd.read_parquet(parquet_path)
target_num = int(train_num)
matched = df[df['train_no'] == target_num]
if matched.empty: return "정보 없음"
row = matched.iloc[0]
return f"{row['location']}\n{row['type']}\n입고 예정\n{row['date']}\n{row['time']}"
except Exception:
return "조회 실패"
# --- 메인 리포트 생성 ---
def generate_voc_report(raw_title, raw_content, doc_date, user_settings):
try:
hwp = win32.gencache.EnsureDispatch("HWPFrame.HwpObject")
hwp.XHwpWindows.Item(0).Visible = True
except Exception as e:
print(f"한글 실행 실패: {e}"); return
base_dir = Path(__file__).resolve().parent
sample_file = base_dir / "VOC_Sample.hwp"
if not sample_file.exists():
print("샘플 파일이 없습니다."); return
# 1. 정보 분석
parsed = parse_voc_info(raw_title, raw_content)
remarks_text = get_schedule_remarks(parsed["train_details"]["num"])
print(f"분석 결과 - 호선: {parsed['line_str']}, 방향: {parsed['train_details']['direction']}")
# 2. 필드 데이터 매핑
field_data = {
"doc_date": doc_date,
"doc_day": get_korean_weekday(doc_date),
"dept_office": user_settings.get('dept_office', ''),
"dept_team": user_settings.get('dept_team', ''),
"reporter_name": user_settings.get('reporter_name', ''),
"reporter_pos": user_settings.get('reporter_pos', ''),
"reporter_tel": user_settings.get('reporter_tel', ''),
"voc_content": raw_content,
"action_taken": user_settings.get('action_taken', ''),
"remarks": remarks_text,
"line_num": parsed['line_str'],
"train_set": parsed['set_str'],
"train_num": parsed['train_str'],
"train_type": parsed['train_details']['type'],
"train_dir": parsed['train_details']['direction']
}
# 3. 데이터 입력
hwp.Open(str(sample_file))
for field, value in field_data.items():
hwp.PutFieldText(field, value)
# (선택) 누름틀 내용 고정
# hwp.Run("AllFieldFlatten")
# 4. 저장
date_clean = doc_date.replace(".", "")
name_parts = [parsed['line_str'], parsed['set_str'], parsed['car_str'], f"{parsed['pure_title']} 관련"]
base_name = " ".join(filter(None, name_parts)).strip()
final_filename = f"{base_name}({date_clean})(VOC).hwp"
save_path = base_dir / re.sub(r'[\\/:*?"<>|]', "", final_filename)
hwp.SaveAs(str(save_path))
print(f"저장 완료: {save_path}")
if __name__ == "__main__":
setup_hwp_security()
# 공통 설정
settings = {
"dept_office": "신평차량사업소",
"dept_team": "검수1팀",
"reporter_name": "이대석",
"reporter_pos": "팀장",
"reporter_tel": "200-5144",
"action_taken": "○ 해당 편성 점검 완료"
}
print("\n--- [Case 1] 일반적인 경우 (열차번호 포함) ---")
generate_voc_report(
"1호선 34편성 3호차, 출입문 고장",
"1147열차 운행 중 출입문 재개폐 발생",
"2026.02.13.",
settings
)
print("\n--- [Case 2] 호선/열번 누락, 문맥 유추 필요 (역명 + 이동경로) ---")
generate_voc_report(
"34편성 3호차, 승객 넘어짐",
"사하역에서 타서 대티역에서 내린 승객이 넘어짐. 열차번호는 모름.",
"2026.02.13.",
settings
)