94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
"""Минимальный драйвер Chrome/Edge через DevTools Protocol (без Selenium/Playwright)."""
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.request
|
|
|
|
import websockets
|
|
|
|
EDGE = r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe"
|
|
CHROME = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
|
|
|
|
|
|
class Page:
|
|
"""Страница, к которой подключаемся по websocket (Edge headless или Chrome на эмуляторе)."""
|
|
|
|
def __init__(self, ws_url):
|
|
self.ws_url, self.n, self.ws = ws_url, 0, None
|
|
|
|
async def open(self):
|
|
self.ws = await websockets.connect(self.ws_url, max_size=2**27)
|
|
await self.call("Page.enable")
|
|
await self.call("Runtime.enable")
|
|
|
|
async def close(self):
|
|
if self.ws:
|
|
await self.ws.close()
|
|
|
|
async def call(self, method, params=None):
|
|
self.n += 1
|
|
await self.ws.send(json.dumps({"id": self.n, "method": method, "params": params or {}}))
|
|
while True:
|
|
m = json.loads(await self.ws.recv())
|
|
if m.get("id") == self.n:
|
|
if "error" in m:
|
|
raise RuntimeError(f"{method}: {m['error']}")
|
|
return m.get("result", {})
|
|
|
|
async def eval(self, expr):
|
|
r = await self.call("Runtime.evaluate", {"expression": expr, "awaitPromise": True, "returnByValue": True})
|
|
if "exceptionDetails" in r:
|
|
d = r["exceptionDetails"]
|
|
raise RuntimeError("JS: " + str(d.get("exception", {}).get("description", d.get("text"))))
|
|
return r["result"].get("value")
|
|
|
|
async def goto(self, url, wait=1.2):
|
|
await self.call("Page.navigate", {"url": url})
|
|
await asyncio.sleep(wait)
|
|
|
|
async def emulate(self, width, height, dpr=1, mobile=False, ua=None):
|
|
await self.call("Emulation.setDeviceMetricsOverride", {"width": width, "height": height, "deviceScaleFactor": dpr, "mobile": mobile})
|
|
await self.call("Emulation.setTouchEmulationEnabled", {"enabled": mobile})
|
|
if ua:
|
|
await self.call("Emulation.setUserAgentOverride", {"userAgent": ua})
|
|
|
|
async def screenshot(self, path):
|
|
r = await self.call("Page.captureScreenshot", {"format": "png", "fromSurface": True})
|
|
with open(path, "wb") as f:
|
|
f.write(base64.b64decode(r["data"]))
|
|
|
|
|
|
class Browser:
|
|
def __init__(self, port=9333):
|
|
self.port, self.proc, self.tmp = port, None, None
|
|
|
|
def start(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="guide_edge_")
|
|
exe = EDGE if os.path.exists(EDGE) else CHROME
|
|
self.proc = subprocess.Popen([exe, "--headless=new", f"--remote-debugging-port={self.port}", f"--user-data-dir={self.tmp}",
|
|
"--no-first-run", "--no-default-browser-check", "--hide-scrollbars", "--lang=ru-RU",
|
|
"--force-color-profile=srgb", "about:blank"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
for _ in range(60):
|
|
try:
|
|
urllib.request.urlopen(f"http://localhost:{self.port}/json/version", timeout=1)
|
|
return
|
|
except Exception:
|
|
time.sleep(0.5)
|
|
raise RuntimeError("браузер не запустился")
|
|
|
|
def page_ws(self):
|
|
tabs = [t for t in json.load(urllib.request.urlopen(f"http://localhost:{self.port}/json")) if t["type"] == "page"]
|
|
return tabs[0]["webSocketDebuggerUrl"]
|
|
|
|
def stop(self):
|
|
if self.proc:
|
|
self.proc.terminate()
|
|
try:
|
|
self.proc.wait(10)
|
|
except Exception:
|
|
self.proc.kill()
|