from __future__ import annotations import re from typing import Optional from .models import ChatHeader, ChatMessage, MessageType, UserMeta # --------------------------------------------------------------------------- # 컴파일된 정규식 # --------------------------------------------------------------------------- # 날짜 구분선: "--------------- 2025년 12월 3일 수요일 ---------------" _RE_DATE_SEP = re.compile( r"^-{15} (\d{4})년 (\d{1,2})월 (\d{1,2})일 \S+요일 -{15}$" ) # 메시지 헤더: "[닉네임] [오전/오후 HH:MM] 내용" _RE_MSG = re.compile( r"^\[(.+?)\] \[(오전|오후) (\d{1,2}):(\d{2})\] (.+)$" ) # 입퇴장: "이름님이 들어왔습니다." / "이름님이 나갔습니다." (뒤에 텍스트 허용) _RE_JOIN_LEAVE = re.compile(r"^(.+)님이 (들어왔습니다|나갔습니다)\.") # 강퇴: "이름님을 내보냈습니다." _RE_KICK = re.compile(r"^(.+)님을 내보냈습니다\.") # 카카오 시스템 노이즈 (단독 줄로 나오는 경우) _SYSTEM_NOISE_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"^불법촬영물 식별 및 게재제한 안내"), ) # --------------------------------------------------------------------------- # 내부 헬퍼 # --------------------------------------------------------------------------- def _to_24h(ampm: str, hour: int, minute: int) -> str: """오전/오후 + 시/분 → "HH:MM" (24시간제).""" if ampm == "오전": h = 0 if hour == 12 else hour else: # 오후 h = hour if hour == 12 else hour + 12 return f"{h:02d}:{minute:02d}" def _parse_user_meta(raw: str) -> UserMeta: """닉네임 raw 문자열 → UserMeta.""" parts = [p.strip() for p in raw.split("/")] if len(parts) >= 3: return UserMeta(name=parts[0], company=parts[1], region=parts[2]) if len(parts) == 2: return UserMeta(name=parts[0], region=parts[1]) return UserMeta(name=parts[0]) def _media_type(content: str) -> Optional[MessageType]: """content가 미디어 키워드(단독 또는 수량 접미사)이면 해당 타입 반환, 아니면 None. 예) "사진", "사진 3장", "동영상" → 해당 타입 """ stripped = content.strip() # 정확한 단어 일치 exact: dict[str, MessageType] = { "사진": MessageType.PHOTO, "이모티콘": MessageType.EMOTICON, "동영상": MessageType.VIDEO, } if stripped in exact: return exact[stripped] # "사진 N장" 형태 처리 for keyword, msg_type in exact.items(): if stripped.startswith(keyword + " "): return msg_type return None def _is_system_noise(line: str) -> bool: return any(p.search(line) for p in _SYSTEM_NOISE_PATTERNS) # --------------------------------------------------------------------------- # 메인 파서 # --------------------------------------------------------------------------- def parse_kakao_chat(filepath: str) -> dict: # type: ignore[type-arg] """ 카카오톡 오픈채팅 txt 파일을 파싱한다. Returns ------- { "header": {"chat_name": str, "save_date": str}, "messages": [ChatMessage, ...], # bot 메시지 제외 "stats": { "total_lines": int, "total_messages": int, "filtered_bot_messages": int, "join_leave_count": int, "media_count": int, "unique_users": int, "date_range": {"start": str, "end": str}, }, } """ messages: list[ChatMessage] = [] bot_messages: list[ChatMessage] = [] current_date = "" pending: Optional[ChatMessage] = None # 멀티라인 조립 중인 메시지 total_lines = 0 def _flush_pending() -> None: """pending 메시지를 적절한 버킷으로 이동.""" nonlocal pending if pending is None: return if pending.type == MessageType.BOT: bot_messages.append(pending) else: messages.append(pending) pending = None with open(filepath, encoding="utf-8") as fh: # ── 헤더 2줄 ────────────────────────────────────────────────────── line1 = fh.readline().rstrip("\n") line2 = fh.readline().rstrip("\n") total_lines += 2 # 줄 1: "... 님과 카카오톡 대화" → chat_name chat_name = line1.strip() # 줄 2: "저장한 날짜 : YYYY-MM-DD HH:MM:SS" → save_date save_date_raw = line2.strip() save_date = save_date_raw.replace("저장한 날짜 : ", "").strip() header = ChatHeader(chat_name=chat_name, save_date=save_date) # ── 본문 줄별 처리 ─────────────────────────────────────────────── for raw_line in fh: total_lines += 1 line = raw_line.rstrip("\n") stripped = line.strip() # 빈 줄 if not stripped: continue # 시스템 노이즈 (단독 줄) if _is_system_noise(stripped): continue # 날짜 구분선 m_date = _RE_DATE_SEP.match(stripped) if m_date: _flush_pending() y, mo, d = m_date.group(1), m_date.group(2), m_date.group(3) current_date = f"{y}-{int(mo):02d}-{int(d):02d}" continue # 입퇴장 / 강퇴 m_jl = _RE_JOIN_LEAVE.match(stripped) m_kick = _RE_KICK.match(stripped) if m_jl or m_kick: _flush_pending() if m_kick: actor = m_kick.group(1) msg_type = MessageType.KICK elif m_jl: actor = m_jl.group(1) action = m_jl.group(2) msg_type = ( MessageType.JOIN if action == "들어왔습니다" else MessageType.LEAVE ) else: # 타입 체커를 위한 방어 코드 (실제로 도달하지 않음) actor = stripped msg_type = MessageType.SYSTEM pending = ChatMessage( date=current_date, time="", user=actor, user_meta=_parse_user_meta(actor), content=stripped, type=msg_type, ) # 입퇴장 메시지는 멀티라인 없이 바로 flush _flush_pending() continue # 일반 메시지 헤더 m_msg = _RE_MSG.match(line) if m_msg: _flush_pending() raw_user = m_msg.group(1) ampm = m_msg.group(2) hour = int(m_msg.group(3)) minute = int(m_msg.group(4)) content = m_msg.group(5) time_str = _to_24h(ampm, hour, minute) # 미디어 타입 결정 media = _media_type(content) if media is not None: msg_type = media elif raw_user == "오픈채팅봇": msg_type = MessageType.BOT else: msg_type = MessageType.MESSAGE pending = ChatMessage( date=current_date, time=time_str, user=raw_user, user_meta=_parse_user_meta(raw_user), content=content, type=msg_type, ) continue # 멀티라인 연속 줄 — 위 패턴 어디에도 해당 안 됨 if pending is not None: pending.content = pending.content + "\n" + line # pending이 None이면 파일 초반 잔여 텍스트 등 → 무시 # 마지막 pending 처리 _flush_pending() # ── 통계 계산 ──────────────────────────────────────────────────────── join_leave_count = sum( 1 for msg in messages if msg.type in (MessageType.JOIN, MessageType.LEAVE, MessageType.KICK) ) media_count = sum( 1 for msg in messages if msg.type in (MessageType.PHOTO, MessageType.EMOTICON, MessageType.VIDEO) ) unique_users: set[str] = { msg.user for msg in messages if msg.type not in (MessageType.JOIN, MessageType.LEAVE, MessageType.KICK, MessageType.SYSTEM) } dates = [msg.date for msg in messages if msg.date] date_range = { "start": min(dates) if dates else "", "end": max(dates) if dates else "", } return { "header": header.model_dump(), "messages": messages, "stats": { "total_lines": total_lines, "total_messages": len(messages), "filtered_bot_messages": len(bot_messages), "join_leave_count": join_leave_count, "media_count": media_count, "unique_users": len(unique_users), "date_range": date_range, }, }