import os import sys import subprocess import urllib.request import json import time from pathlib import Path from typing import Dict, Any, Optional, Tuple from config import Config class BrowserManager: """ Orchestrates the isolated Chrome Dev environment via Chrome DevTools Protocol (CDP) or Playwright Persistent Context. Guarantees 100% isolation from the user's personal Chrome. """ def __init__(self, cdp_port: Optional[int] = None, profile_dir: Optional[Path] = None): self.cdp_port = cdp_port or Config.CDP_PORT self.cdp_url = f"http://localhost:{self.cdp_port}" self.profile_dir = profile_dir or Config.BOT_PROFILE_DIR self.chrome_exe = Config.CHROME_EXECUTABLE # Ensure bot profile dir exists self.profile_dir.mkdir(parents=True, exist_ok=True) def check_cdp_status(self) -> Dict[str, Any]: """ Check if Chrome Dev is running with remote debugging port active. """ version_url = f"{self.cdp_url}/json/version" try: req = urllib.request.Request(version_url, headers={"User-Agent": "DSE-Automation/1.0"}) with urllib.request.urlopen(req, timeout=2.0) as response: if response.status == 200: data = json.loads(response.read().decode("utf-8")) return { "active": True, "browser": data.get("Browser", "Chrome"), "protocol": data.get("Protocol-Version", "1.3"), "web_socket_debugger_url": data.get("webSocketDebuggerUrl"), "port": self.cdp_port, "mode": "CDP (Attached)" } except Exception: pass return { "active": False, "browser": None, "port": self.cdp_port, "mode": "Disconnected (Will launch dedicated profile)" } def open_url_in_cdp(self, url: str) -> bool: """ Opens a URL in a new tab of the running CDP Chrome instance and focuses it. """ try: import urllib.parse encoded = urllib.parse.quote(url, safe="") new_tab_url = f"{self.cdp_url}/json/new?{encoded}" req = urllib.request.Request(new_tab_url, method="PUT", headers={"User-Agent": "DSE-Automation/1.0"}) with urllib.request.urlopen(req, timeout=3.0) as response: return response.status == 200 except Exception as e: print(f"[CDP OPEN URL ERROR] {e}") return False def kill_bot_chrome(self) -> bool: """ Safely terminates only the Chrome process running on CDP port 9222 or using BotProfile_Sosmed, guaranteeing the user's personal Chrome is 100% untouched. """ try: if sys.platform == "win32": cmd = f'netstat -ano | findstr :{self.cdp_port}' output = subprocess.check_output(cmd, shell=True, text=True, stderr=subprocess.DEVNULL) for line in output.strip().splitlines(): parts = line.split() if len(parts) >= 5 and "LISTENING" in parts: pid = parts[-1] subprocess.run(f'taskkill /F /T /PID {pid}', shell=True, capture_output=True) else: # Linux / cPanel process cleanup subprocess.run(f"fuser -k -n tcp {self.cdp_port} 2>/dev/null", shell=True, capture_output=True) time.sleep(1) return True except Exception: return False def launch_isolated_chrome(self, url: Optional[str] = None) -> Dict[str, Any]: """ Launch Chrome in isolated mode with remote debugging enabled. Guaranteed to open visibly on the user's active Windows desktop, or handle headless cPanel server environments gracefully. """ target_url = url or "https://www.google.com" status = self.check_cdp_status() if status["active"]: # Already active: open the target URL in a new tab to bring Chrome into view if url: self.open_url_in_cdp(target_url) return { "success": True, "message": "Chrome Bot sudah aktif! Tab telah dibuka di browser.", "status": status } # If on server/cPanel and no standalone chrome executable found, # Playwright persistent context handles execution automatically if not self.chrome_exe or not os.path.exists(str(self.chrome_exe)): return { "success": True, "message": "Mode server/cPanel terdeteksi: Browser akan dijalankan via Playwright context secara otomatis saat kampanye dimulai.", "status": status } if sys.platform == "win32": # On Windows, 'cmd.exe /c start ""' invokes ShellExecute which guarantees # an interactive top-level desktop window is opened and focused. args = [ "cmd.exe", "/c", "start", "", str(self.chrome_exe), f"--remote-debugging-port={self.cdp_port}", f"--user-data-dir={str(self.profile_dir)}", "--profile-directory=Default", "--no-first-run", "--no-default-browser-check", "--disable-background-networking", "--disable-background-timer-throttling", "--disable-backgrounding-occluded-windows", "--disable-breakpad", "--disable-component-update", "--disable-default-apps", "--disable-domain-reliability", "--disable-sync", "--disable-renderer-backgrounding", "--mute-audio", "--start-maximized", target_url ] else: # Linux / cPanel hosting environment args = [ str(self.chrome_exe), f"--remote-debugging-port={self.cdp_port}", f"--user-data-dir={str(self.profile_dir)}", "--profile-directory=Default", "--no-first-run", "--no-default-browser-check", "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu", "--disable-background-networking", "--disable-sync", "--mute-audio", ] if Config.HEADLESS: args.append("--headless=new") args.append(target_url) try: if sys.platform == "win32": subprocess.Popen(args, shell=False) else: subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # Wait up to 6 seconds for CDP to become active for _ in range(12): time.sleep(0.5) new_status = self.check_cdp_status() if new_status["active"]: return { "success": True, "message": "Jendela Chrome Bot berhasil dibuka!", "status": new_status } return { "success": True, "message": "Jendela Chrome Bot sedang proses membuka...", "status": self.check_cdp_status() } except Exception as e: return {"success": False, "error": f"Gagal meluncurkan Chrome Bot: {e}"} def restart_chrome(self, url: Optional[str] = None) -> Dict[str, Any]: """ Force restart Chrome Bot in case the window was closed or orphaned. """ self.kill_bot_chrome() time.sleep(1) return self.launch_isolated_chrome(url=url) def acquire_session(self, playwright_instance) -> Tuple[Any, Any, str]: """ Acquires browser and browser context for tab management. Returns (browser_or_context, context, connection_mode). """ cdp_status = self.check_cdp_status() if cdp_status["active"]: try: print(f"[BROWSER] Menghubungkan ke Chrome Dev via CDP di {self.cdp_url}...") browser = playwright_instance.chromium.connect_over_cdp(self.cdp_url) contexts = browser.contexts context = contexts[0] if contexts else browser.new_context() return browser, context, "cdp" except Exception as e: print(f"[WARN] Gagal menyambung via CDP: {e}. Beralih ke persistent context...") # Fallback: Launch independent persistent context using the same isolated BotProfile is_headless = Config.HEADLESS print(f"[BROWSER] Meluncurkan Persistent Context di {self.profile_dir} (Headless: {is_headless})...") launch_args = [ "--disable-blink-features=AutomationControlled", "--no-first-run", "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu" ] if not is_headless: launch_args.append("--start-maximized") kwargs = { "user_data_dir": str(self.profile_dir), "headless": is_headless, "viewport": None if not is_headless else {"width": 1280, "height": 800}, "args": launch_args } if self.chrome_exe and os.path.exists(str(self.chrome_exe)): kwargs["executable_path"] = str(self.chrome_exe) context = playwright_instance.chromium.launch_persistent_context(**kwargs) return context, context, "persistent" def acquire_page(self, playwright_instance) -> Tuple[Any, Any, Any, str]: """ Acquires a Playwright browser, context, and a verified active web page. Filters out internal chrome:// and extension pages. Returns (browser_or_context, context, page, connection_mode). """ browser_or_ctx, context, mode = self.acquire_session(playwright_instance) # Filter out internal pages valid_pages = [ p for p in context.pages if not p.is_closed() and not p.url.startswith("chrome://") and not p.url.startswith("chrome-extension://") ] if valid_pages: page = valid_pages[0] else: page = context.new_page() try: page.bring_to_front() except Exception: pass return browser_or_ctx, context, page, mode