""" Bộ Giám Sát Nhật Ký Giao Dịch & Sự Kiện Trực Tiếp từ Client.txt (POE2 Realtime Log Watcher) """ import os import re import threading import time from dataclasses import dataclass from typing import Callable, List, Optional @dataclass class LogEvent: event_type: str # "WHISPER_TRADE", "TRADE_CHAT", "ZONE_CHANGE", "AREA_GENERATED", ... sender: str message: str timestamp: str raw_line: str area_name: str = "" area_code: str = "" class LogWatcher: # Các UI scene overlay mở bằng phím U hoặc giá trị không hợp lệ IGNORED_SCENES = frozenset({ "act 1", "act 2", "act 3", "act 4", "atlas", "(null)", "(unknown)", "world map", "[dnt] world map", }) # Cửa sổ đọc đuôi file khi seed zone hiện tại (Client.txt có thể >100MB) SEED_TAIL_BYTES = 2 * 1024 * 1024 def __init__(self, log_path: str): self.log_path = log_path self._running = False self._thread: Optional[threading.Thread] = None self._listeners: List[Callable[[LogEvent], None]] = [] # Các mẫu Regex phân tích cú pháp nhật ký POE2 # Mẫu 1: Thì thầm giao dịch (@From / @To) self._re_whisper = re.compile( r"^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?\[INFO Client \d+\] @(From|To) (?:<[^>]+> )?([a-zA-Z0-9_]+): (.+)$" ) # Mẫu 2: Tin nhắn kênh buôn bán Trade Chat ($) self._re_trade_chat = re.compile( r"^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?\[INFO Client \d+\] \$([a-zA-Z0-9_]+): (.+)$" ) # Mẫu 3: Chuyển đổi bản đồ / khu vực self._re_scene = re.compile( r"^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?\[SCENE\] Set Source \[(.+?)\]" ) # Mẫu 3b: Tạo instance (area code nội bộ + seed) self._re_area_gen = re.compile( r'^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?Generating level (\d+) area "([^"]+)" with seed (\d+)' ) # Mẫu 4: Trạng thái cửa sổ giao dịch self._re_trade_status = re.compile( r"^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?\[INFO Client \d+\] (Trade accepted\.|Trade cancelled\.)" ) # Mẫu 5: Người chơi tử trận (Death detection) self._re_player_death = re.compile( r"^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}).+?\[INFO Client \d+\] : (.+?) (has been slain|has committed suicide)\." ) @classmethod def is_ignored_scene(cls, zone_name: str) -> bool: clean = (zone_name or "").strip() if not clean: return True lower = clean.lower() if lower.startswith("[dnt"): return True return lower in cls.IGNORED_SCENES def register_listener(self, callback: Callable[[LogEvent], None]) -> None: self._listeners.append(callback) def _dispatch(self, event: LogEvent) -> None: for cb in self._listeners: try: cb(event) except Exception: pass def start(self) -> bool: if not os.path.exists(self.log_path): return False self._running = True self._thread = threading.Thread(target=self._tail_loop, daemon=True) self._thread.start() return True def stop(self) -> None: self._running = False if self._thread and self._thread.is_alive(): self._thread.join(timeout=1.0) def seed_latest_playable_zone(self) -> Optional[LogEvent]: """ Đọc đuôi Client.txt và phát ZONE_CHANGE cho SCENE chơi thật gần nhất. Cần gọi SAU khi listener (lifecycle) đã đăng ký: tool mở giữa map không được kẹt 'chờ xác thực' chỉ vì tail bắt đầu ở EOF. Overlay Act 1–4 / Atlas / (null) bị bỏ qua. """ if not os.path.isfile(self.log_path): return None try: size = os.path.getsize(self.log_path) start = max(0, size - self.SEED_TAIL_BYTES) last_scene: Optional[LogEvent] = None with open(self.log_path, "r", encoding="utf-8", errors="replace") as f: if start > 0: f.seek(start) f.readline() for raw in f: event = self.parse_line(raw.strip()) if event and event.event_type == "ZONE_CHANGE": last_scene = event if last_scene: return last_scene return None except Exception as e: print(f"[LogWatcher] Lỗi seed zone từ {self.log_path}: {e}") return None def _tail_loop(self) -> None: try: with open(self.log_path, "r", encoding="utf-8", errors="replace") as f: # Tua thẳng tới cuối file để chỉ đón nhận sự kiện mới nhất f.seek(0, os.SEEK_END) while self._running: line = f.readline() if not line: time.sleep(0.2) continue event = self.parse_line(line.strip()) if event: self._dispatch(event) except Exception as e: print(f"[LogWatcher] Lỗi đọc file {self.log_path}: {e}") def parse_line(self, line: str) -> Optional[LogEvent]: # 1. Kiểm tra Whisper m_whisp = self._re_whisper.search(line) if m_whisp: ts, direction, sender, msg = m_whisp.groups() return LogEvent( event_type="WHISPER_TRADE" if ("buy" in msg.lower() or "wts" in msg.lower()) else "WHISPER", sender=f"{direction} {sender}", message=msg, timestamp=ts, raw_line=line ) # 2. Kiểm tra Trade Chat m_tc = self._re_trade_chat.search(line) if m_tc: ts, sender, msg = m_tc.groups() return LogEvent( event_type="TRADE_CHAT", sender=sender, message=msg, timestamp=ts, raw_line=line ) # 3a. Tạo instance (area code). Không dùng code nội bộ làm tên hiển thị. m_gen = self._re_area_gen.search(line) if m_gen: ts, _level, area_code, _seed = m_gen.groups() return LogEvent( event_type="AREA_GENERATED", sender="System", message=area_code, timestamp=ts, raw_line=line, area_code=area_code, ) # 3b. Kiểm tra Chuyển Khu Vực m_sc = self._re_scene.search(line) if m_sc: ts, zone_name = m_sc.groups() clean_zone = (zone_name or "").strip() if self.is_ignored_scene(clean_zone): return None return LogEvent( event_type="ZONE_CHANGE", sender="System", message=f"Đã chuyển đến khu vực: {clean_zone}", timestamp=ts, raw_line=line, area_name=clean_zone, ) # 4. Kiểm tra Trạng thái Giao dịch m_stat = self._re_trade_status.search(line) if m_stat: ts, status = m_stat.groups() return LogEvent( event_type="TRADE_STATUS", sender="System", message=status, timestamp=ts, raw_line=line ) # 5. Kiểm tra Người chơi tử trận (Death Event) m_death = self._re_player_death.search(line) if m_death: ts, player_name, reason = m_death.groups() return LogEvent( event_type="PLAYER_DEATH", sender=player_name, message=f"{player_name} {reason}.", timestamp=ts, raw_line=line ) return None def simulate_event(self, event_type: str, sender: str, message: str) -> None: """Hỗ trợ bắn sự kiện giả lập phục vụ kiểm thử và demo nhanh.""" event = LogEvent( event_type=event_type, sender=sender, message=message, timestamp=time.strftime("%Y/%m/%d %H:%M:%S"), raw_line=f"[SIMULATED] {sender}: {message}" ) self._dispatch(event)