233 lines
9.2 KiB
Python
233 lines
9.2 KiB
Python
import re
|
|
import time
|
|
import os
|
|
from typing import Dict, List, Tuple, Any, Optional
|
|
from datetime import datetime
|
|
|
|
class ConversationAnalyzer:
|
|
def __init__(self):
|
|
# 대화 구분을 위한 패턴
|
|
self.controller_patterns = [
|
|
r"전철\s*(?:신평|보안)",
|
|
r"관제\s*(?:\d+)?",
|
|
]
|
|
|
|
self.driver_patterns = [
|
|
r"(\d{4})\s*열차\s*(?:(\d+)\s*편성)?",
|
|
r"다대\s*회차선\s*(\d{4})\s*열차\s*(?:(\d+)\s*편성)?",
|
|
]
|
|
|
|
# 현재 및 최근 대화 저장
|
|
self.current_conversation = {
|
|
"id": "",
|
|
"start_time": None,
|
|
"last_update": None,
|
|
"train_number": None,
|
|
"train_formation": None,
|
|
"speakers": [],
|
|
"messages": [],
|
|
"faults": []
|
|
}
|
|
|
|
self.conversations = []
|
|
self.max_conversation_history = 50
|
|
|
|
# 로그 디렉토리 생성
|
|
self.log_dir = "conversation_logs"
|
|
try:
|
|
if not os.path.exists(self.log_dir):
|
|
os.makedirs(self.log_dir)
|
|
except Exception as e:
|
|
print(f"로그 디렉토리 생성 오류: {e}")
|
|
|
|
def identify_speakers(self, text: str) -> Dict[str, str]:
|
|
"""텍스트에서 화자를 식별합니다."""
|
|
result = {"role": "unknown", "name": "알 수 없음"}
|
|
|
|
# 관제사 패턴 확인
|
|
for pattern in self.controller_patterns:
|
|
match = re.search(pattern, text)
|
|
if match:
|
|
result["role"] = "controller"
|
|
result["name"] = "관제사"
|
|
break
|
|
|
|
# 기관사 패턴 확인
|
|
for pattern in self.driver_patterns:
|
|
match = re.search(pattern, text)
|
|
if match:
|
|
result["role"] = "driver"
|
|
result["name"] = "기관사"
|
|
|
|
# 열차 번호와 편성 추출
|
|
train_number = match.group(1) if match.groups() and len(match.groups()) > 0 else None
|
|
train_formation = match.group(2) if match.groups() and len(match.groups()) > 1 else None
|
|
|
|
if train_number:
|
|
result["train_number"] = train_number
|
|
if train_formation:
|
|
result["train_formation"] = train_formation
|
|
break
|
|
|
|
return result
|
|
|
|
def identify_faults(self, text: str) -> List[str]:
|
|
"""텍스트에서 고장 관련 키워드를 식별합니다."""
|
|
fault_patterns = [
|
|
r"HVAC",
|
|
r"배기팬",
|
|
r"고장",
|
|
r"장애",
|
|
r"오류",
|
|
r"문제",
|
|
r"에러",
|
|
r"점검",
|
|
r"이상",
|
|
]
|
|
|
|
found_faults = []
|
|
for pattern in fault_patterns:
|
|
if re.search(pattern, text, re.IGNORECASE):
|
|
found_faults.append(pattern)
|
|
|
|
return found_faults
|
|
|
|
def analyze_conversation(self, text: str, speaker_info: Dict[str, str]) -> Dict[str, Any]:
|
|
"""대화 내용을 분석하고 현재 대화 컨텍스트를 업데이트합니다."""
|
|
try:
|
|
current_time = time.time()
|
|
|
|
# 새 대화 시작 여부 확인
|
|
if (self.current_conversation["last_update"] is None or
|
|
current_time - self.current_conversation["last_update"] > 60): # 1분 이상 침묵
|
|
# 이전 대화가 있으면 저장
|
|
if self.current_conversation["start_time"] is not None:
|
|
# 대화 로그 저장
|
|
self._save_conversation_log(self.current_conversation)
|
|
|
|
self.conversations.insert(0, self.current_conversation.copy())
|
|
# 최대 저장 개수 제한
|
|
if len(self.conversations) > self.max_conversation_history:
|
|
self.conversations = self.conversations[:self.max_conversation_history]
|
|
|
|
# 새 대화 초기화
|
|
conversation_id = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
self.current_conversation = {
|
|
"id": conversation_id,
|
|
"start_time": current_time,
|
|
"last_update": current_time,
|
|
"train_number": None,
|
|
"train_formation": None,
|
|
"speakers": [],
|
|
"messages": [],
|
|
"faults": []
|
|
}
|
|
else:
|
|
# 기존 대화 업데이트
|
|
self.current_conversation["last_update"] = current_time
|
|
|
|
# 화자 정보 업데이트
|
|
speaker_role = speaker_info.get("role", "unknown")
|
|
speaker_name = speaker_info.get("name", "알 수 없음")
|
|
|
|
speaker_exists = False
|
|
for s in self.current_conversation["speakers"]:
|
|
if s["role"] == speaker_role:
|
|
speaker_exists = True
|
|
break
|
|
|
|
if not speaker_exists:
|
|
self.current_conversation["speakers"].append({
|
|
"role": speaker_role,
|
|
"name": speaker_name
|
|
})
|
|
|
|
# 열차 번호와 편성 업데이트
|
|
if "train_number" in speaker_info and self.current_conversation["train_number"] is None:
|
|
self.current_conversation["train_number"] = speaker_info["train_number"]
|
|
|
|
if "train_formation" in speaker_info and self.current_conversation["train_formation"] is None:
|
|
self.current_conversation["train_formation"] = speaker_info["train_formation"]
|
|
|
|
# 고장 정보 식별 및 업데이트
|
|
faults = self.identify_faults(text)
|
|
for fault in faults:
|
|
if fault not in self.current_conversation["faults"]:
|
|
self.current_conversation["faults"].append(fault)
|
|
|
|
# 메시지 추가
|
|
timestamp = datetime.fromtimestamp(current_time).strftime("%H:%M:%S")
|
|
self.current_conversation["messages"].append({
|
|
"time": timestamp,
|
|
"speaker": speaker_name,
|
|
"text": text
|
|
})
|
|
|
|
return self.current_conversation
|
|
|
|
except Exception as e:
|
|
print(f"대화 분석 오류: {e}")
|
|
# 오류 발생 시 기본 대화 객체 반환
|
|
return {
|
|
"id": datetime.now().strftime("%Y%m%d%H%M%S"),
|
|
"start_time": time.time(),
|
|
"last_update": time.time(),
|
|
"train_number": None,
|
|
"train_formation": None,
|
|
"speakers": [{
|
|
"role": "unknown",
|
|
"name": "알 수 없음"
|
|
}],
|
|
"messages": [{
|
|
"time": datetime.now().strftime("%H:%M:%S"),
|
|
"speaker": "알 수 없음",
|
|
"text": text
|
|
}],
|
|
"faults": []
|
|
}
|
|
|
|
def _save_conversation_log(self, conversation: Dict[str, Any]):
|
|
"""대화 내용을 로그 파일로 저장합니다."""
|
|
try:
|
|
if not conversation or not conversation.get("messages"):
|
|
return
|
|
|
|
log_filename = f"{self.log_dir}/conversation_{conversation['id']}.txt"
|
|
|
|
with open(log_filename, "w", encoding="utf-8") as f:
|
|
# 헤더 정보
|
|
f.write(f"대화 ID: {conversation['id']}\n")
|
|
f.write(f"시작 시간: {datetime.fromtimestamp(conversation['start_time']).strftime('%Y-%m-%d %H:%M:%S')}\n")
|
|
|
|
if conversation.get("train_number"):
|
|
f.write(f"열차 번호: {conversation['train_number']}\n")
|
|
|
|
if conversation.get("train_formation"):
|
|
f.write(f"편성 번호: {conversation['train_formation']}\n")
|
|
|
|
if conversation.get("faults"):
|
|
f.write(f"감지된 고장: {', '.join(conversation['faults'])}\n")
|
|
|
|
f.write("\n--- 대화 내용 ---\n\n")
|
|
|
|
# 대화 내용
|
|
for msg in conversation["messages"]:
|
|
f.write(f"[{msg['time']}] {msg['speaker']}: {msg['text']}\n")
|
|
|
|
except Exception as e:
|
|
print(f"대화 로그 저장 오류: {e}")
|
|
|
|
def get_conversation_history(self) -> List[Dict[str, Any]]:
|
|
"""대화 기록을 반환합니다."""
|
|
return self.conversations
|
|
|
|
def get_conversation_by_id(self, conversation_id: str) -> Optional[Dict[str, Any]]:
|
|
"""ID로 대화를 찾아 반환합니다."""
|
|
if self.current_conversation["id"] == conversation_id:
|
|
return self.current_conversation
|
|
|
|
for conv in self.conversations:
|
|
if conv["id"] == conversation_id:
|
|
return conv
|
|
|
|
return None |