""" Module Tự Động Chụp Màn Hình & Ghép Dữ Liệu Bộ Nhớ Để Phân Tích (Screen & Memory Analyzer) Chụp ảnh khung hình in-game kết hợp trích xuất snapshot bộ nhớ thời gian thực để huấn luyện và nâng cao chất lượng bot. """ import ctypes import ctypes.wintypes import json import os import time from datetime import datetime from typing import Any, Dict, Optional import win32con import win32gui from PIL import Image, ImageDraw, ImageGrab try: from src.assistant_tool.sync_memory_snapshot import SyncMemorySnapshotManager, draw_correlated_image except ImportError: from sync_memory_snapshot import SyncMemorySnapshotManager, draw_correlated_image # type: ignore[no-redef] from src.common.win32_window import ( attach_to_default_desktop, ensure_window_focus as _ensure_window_focus, find_poe2_window as _find_poe2_window, get_client_rect, ) class ScreenCapturer: def __init__(self, output_dir: str = "captures"): self.output_dir = output_dir os.makedirs(self.output_dir, exist_ok=True) self.capture_count = 0 self.last_capture_path = "" self.last_capture_time = 0.0 self.sync_ram_manager = SyncMemorySnapshotManager() def find_poe2_window(self) -> Optional[int]: """Tìm HWND cửa sổ Path of Exile 2.""" hwnd = _find_poe2_window() return hwnd if hwnd != 0 else None def bring_to_foreground(self, hwnd: Optional[int] = None) -> bool: """Tự động chuyển tab và đưa cửa sổ Path of Exile 2 lên foreground an toàn.""" target_hwnd = hwnd or self.find_poe2_window() if not target_hwnd or not ctypes.windll.user32.IsWindow(target_hwnd): return False return _ensure_window_focus(target_hwnd, timeout_ms=800) def capture_screenshot( self, event_name: str = "MANUAL", telemetry_data: Optional[Dict[str, Any]] = None, capture_ram: bool = True, auto_focus: bool = False, background_mode: bool = True, ) -> Optional[str]: """ Chụp màn hình cửa sổ game ĐỒNG BỘ với RAM Snapshot tại cùng timestamp chính xác. Tự động xuất ảnh phân tích tương quan (_CORRELATED.png) và dump RAM (_RAM.json). Tuân thủ INV-CAPTURE-CLIENT-ONLY: Chỉ chụp Client Area (loại trừ title bar và viền), không cướp focus của người dùng khi ở background_mode=True. """ now = time.time() # Chống chụp dồn dập (chờ đủ cooldown tối thiểu 0.3s) if now - self.last_capture_time < 0.3: time.sleep(0.3 - (now - self.last_capture_time)) self.last_capture_time = time.time() timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:19] safe_event = "".join(c if c.isalnum() else "_" for c in event_name).upper() img_filename = f"{timestamp_str}_{safe_event}.png" json_filename = f"{timestamp_str}_{safe_event}.json" ram_filename = f"{timestamp_str}_{safe_event}_RAM.json" corr_filename = f"{timestamp_str}_{safe_event}_CORRELATED.png" img_path = os.path.join(self.output_dir, img_filename) json_path = os.path.join(self.output_dir, json_filename) ram_path = os.path.join(self.output_dir, ram_filename) corr_path = os.path.join(self.output_dir, corr_filename) hwnd = self.find_poe2_window() # Khi ở background_mode, tuyệt đối cấm cướp focus của người dùng if auto_focus and not background_mode and hwnd: self.bring_to_foreground(hwnd) bbox = None cl, ct, cw, ch = 0, 0, 0, 0 screenshot = None if hwnd: cl, ct, cw, ch = get_client_rect(hwnd) if cw > 0 and ch > 0: bbox = (cl, ct, cl + cw, ct + ch) # 1. Ưu tiên số 1: Chụp qua capture_window_bitblt (PrintWindow DWM DirectX 12, chống che khuất bởi IDE/trình duyệt) if hwnd: try: from src.common.screen_capture import capture_window_bitblt screenshot = capture_window_bitblt(hwnd, client_only=True) except Exception: screenshot = None # 2. Fallback chụp Client Area trực tiếp qua capture_rect_bitblt từ Desktop DC if screenshot is None and hwnd and cw > 0 and ch > 0: try: from src.common.screen_capture import capture_rect_bitblt screenshot = capture_rect_bitblt(cl, ct, cw, ch) except Exception: screenshot = None # 3. Fallback chụp qua capture_window_print với render full content và crop về client area (chuẩn Reconnect module) if screenshot is None and hwnd: try: from src.common.screen_capture import capture_window_print full = capture_window_print(hwnd, render_full_content=True) if full and cw > 0 and ch > 0: w, h = full.size bx = max(0, (w - cw) // 2) by = max(0, h - ch - bx) screenshot = full.crop((bx, by, bx + cw, by + ch)) elif full: screenshot = full except Exception: screenshot = None # 4. Fallback thử ImageGrab với bbox chính xác của Client Area # TUYỆT ĐỐI TUÂN THỦ INV-CAPTURE-CLIENT-ONLY: Cấm gọi ImageGrab.grab() không tham số để tránh rò rỉ Desktop if screenshot is None and bbox is not None: try: screenshot = ImageGrab.grab(bbox=bbox) except Exception: screenshot = None # 3. Fallback: Nếu không chụp được (game chưa mở, môi trường service/headless hoặc session locked), # tạo ảnh chuẩn đoán chuyên dụng chứa telemetry để pipeline không bao giờ đứt gãy, cấm chụp Desktop if screenshot is None: w, h = (1280, 720) if not bbox else (max(400, bbox[2] - bbox[0]), max(300, bbox[3] - bbox[1])) screenshot = Image.new("RGB", (w, h), color=(15, 23, 42)) draw = ImageDraw.Draw(screenshot) draw.rectangle([10, 10, w - 10, h - 10], outline=(245, 158, 11), width=2) draw.text((25, 25), f"AutoPOE2 Diagnostic Capture - Event: {event_name}", fill=(245, 158, 11)) draw.text((25, 50), f"Timestamp: {timestamp_str}", fill=(226, 232, 240)) y = 75 if not hwnd: draw.text((25, y), "Window Status: NOT FOUND (INV-CAPTURE-CLIENT-ONLY: Desktop capture prohibited)", fill=(239, 68, 68)) y += 22 if telemetry_data: for k, v in telemetry_data.items(): draw.text((25, y), f"{k}: {v}", fill=(148, 163, 184)) y += 22 try: screenshot.save(img_path, "PNG") self.capture_count += 1 self.last_capture_path = img_path # 3. Chụp RAM Snapshot đồng bộ tại cùng timestamp ram_snapshot = None if capture_ram: sw, sh = screenshot.size ram_snapshot = self.sync_ram_manager.capture_ram_snapshot( fallback_telemetry=telemetry_data, screen_w=sw, screen_h=sh ) with open(ram_path, "w", encoding="utf-8") as f: json.dump(ram_snapshot, f, indent=2, ensure_ascii=False) # Vẽ ảnh phân tích đối chiếu trực quan draw_correlated_image(img_path, ram_snapshot, corr_path) # Ghi file metadata JSON ghép nối bộ nhớ meta = { "capture_id": self.capture_count, "timestamp": timestamp_str, "event": event_name, "window_found": bool(hwnd), "bbox": bbox, "telemetry": telemetry_data or {}, "ram_snapshot_file": ram_filename if capture_ram else None, "correlated_image": corr_filename if capture_ram else None, "mobs_detected_in_ram": len(ram_snapshot.get("monsters", [])) if ram_snapshot else 0 } with open(json_path, "w", encoding="utf-8") as f: json.dump(meta, f, indent=2, ensure_ascii=False) return img_path except Exception as e: print(f"[ScreenCapture Error] Lỗi lưu file: {e}", flush=True) return None def get_stats(self) -> Dict[str, Any]: return { "total_captures": self.capture_count, "output_dir": os.path.abspath(self.output_dir), "last_capture": self.last_capture_path } @staticmethod def is_diagnostic_capture_file(file_path: Optional[str]) -> bool: """Kiểm tra xem file ảnh có phải là ảnh chuẩn đoán chắp vá (Diagnostic Fallback) không.""" if not file_path or not os.path.exists(file_path): return True try: with Image.open(file_path) as img: # Ảnh chẩn đoán có màu nền đặc trưng (15, 23, 42) tại góc px = img.getpixel((20, 20)) if isinstance(px, tuple) and len(px) >= 3: if abs(px[0] - 15) <= 2 and abs(px[1] - 23) <= 2 and abs(px[2] - 42) <= 2: return True except Exception: pass return False