# retoor <retoor@molodetz.nl>
import asyncio
import logging
import random
import time
from typing import Optional
from playwright.async_api import Page, Playwright, async_playwright
from devplacepy.services.bot.config import (
MONITOR_JPEG_QUALITY,
MONITOR_MIN_INTERVAL_SECONDS,
MONITOR_SCALE,
)
from devplacepy.services.bot.monitor import monitor
logger = logging.getLogger(__name__)
DEFAULT_USER_AGENT = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
)
DEFAULT_VIEWPORT = {"width": 1280, "height": 900}
class BotBrowser:
def __init__(
self,
headless: bool = True,
user_agent: str = "",
viewport: Optional[dict] = None,
):
self._headless = headless
self._user_agent = user_agent or DEFAULT_USER_AGENT
self._viewport = dict(viewport) if viewport else dict(DEFAULT_VIEWPORT)
self._playwright: Optional[Playwright] = None
self._browser = None
self._context = None
self._page: Optional[Page] = None
self._monitor_slot: Optional[int] = None
self._last_capture: float = 0.0
async def launch(self) -> None:
self._playwright = await async_playwright().start()
self._browser = await self._playwright.chromium.launch(
headless=self._headless,
args=[
"--disable-blink-features=AutomationControlled",
"--no-first-run",
"--disable-dev-shm-usage",
f"--window-size={self._viewport['width']},{self._viewport['height']}",
],
)
self._context = await self._browser.new_context(
user_agent=self._user_agent,
viewport=self._viewport,
locale="en-US",
)
self._page = await self._context.new_page()
async def close(self) -> None:
context, self._context = self._context, None
browser, self._browser = self._browser, None
playwright, self._playwright = self._playwright, None
self._page = None
for label, resource, shutdown in (
("context", context, lambda r: r.close()),
("browser", browser, lambda r: r.close()),
("playwright", playwright, lambda r: r.stop()),
):
if resource is None:
continue
try:
await shutdown(resource)
except Exception as e:
logger.debug("Browser %s close error: %s", label, e)
@property
def page(self) -> Page:
return self._page
def bind_monitor(
self, slot: int, *, username: str = "", persona: str = ""
) -> None:
self._monitor_slot = slot
monitor.update_meta(slot, username=username, persona=persona)
def note(self, *, action: str = "", status: str = "") -> None:
if self._monitor_slot is None:
return
monitor.update_meta(self._monitor_slot, action=action, status=status)
async def capture(self, reason: str = "", *, force: bool = False) -> None:
if self._monitor_slot is None or self._page is None:
return
now = time.time()
if not force and now - self._last_capture < MONITOR_MIN_INTERVAL_SECONDS:
return
self._last_capture = now
try:
current_url = ""
try:
current_url = self._page.url
except Exception:
current_url = ""
monitor.update_meta(
self._monitor_slot, status=reason, url=current_url
)
image = await self._page.screenshot(
type="jpeg",
quality=MONITOR_JPEG_QUALITY,
scale=MONITOR_SCALE,
full_page=False,
timeout=4000,
)
monitor.store_image(self._monitor_slot, image)
except Exception as e:
logger.debug("monitor capture (%s) failed: %s", reason, e)
async def goto(self, url: str) -> None:
try:
await self._page.goto(url, wait_until="domcontentloaded", timeout=15000)
except Exception as e:
logger.debug("goto (domcontentloaded) failed for %s: %s", url[:80], e)
try:
await self._page.goto(url, timeout=20000)
except Exception as e2:
logger.warning("goto failed for %s: %s", url[:80], e2)
await self._idle(0.5, 1.5)
await self.capture("page load", force=True)
async def _idle(self, lo: float = 0.5, hi: float = 2.0) -> None:
await asyncio.sleep(random.uniform(lo, hi))
async def scroll(self, amount: Optional[int] = None) -> None:
if amount is None:
amount = random.randint(300, 800)
try:
await self._page.evaluate(f"window.scrollBy(0, {amount})")
except Exception as e:
logger.debug("scroll failed: %s", e)
await self._idle(0.2, 0.6)
await self.capture("scroll")
async def scroll_to_bottom(self) -> None:
try:
await self._page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
except Exception as e:
logger.debug("scroll_to_bottom failed: %s", e)
async def fill(self, sel: str, val: str) -> bool:
try:
el = self._page.locator(sel).first
if await el.count() == 0:
logger.debug("fill: selector '%s' not found", sel)
return False
await el.click(timeout=3000)
await self._idle(0.05, 0.15)
await el.fill("")
for ch in val:
await self._page.keyboard.type(ch, delay=random.randint(20, 60))
if ch == " ":
await asyncio.sleep(random.uniform(0.02, 0.08))
await self.capture("field input")
return True
except Exception as e:
logger.debug("fill failed for '%s': %s", sel, e)
return False
async def click(self, sel: str) -> bool:
try:
el = self._page.locator(sel).first
if await el.count() == 0:
logger.debug("click: selector '%s' not found", sel)
return False
box = await el.bounding_box()
if box:
tx = box["x"] + box["width"] / 2 + random.uniform(-3, 3)
ty = box["y"] + box["height"] / 2 + random.uniform(-3, 3)
await self._page.mouse.move(tx, ty, steps=random.randint(8, 18))
await self._idle(0.02, 0.06)
await el.click(timeout=5000)
return True
except Exception as e:
logger.debug("click failed for '%s': %s", sel, e)
return False
async def text(self, sel: str) -> str:
try:
return (await self._page.locator(sel).first.inner_text()) or ""
except Exception as e:
logger.debug("text failed for '%s': %s", sel, e)
return ""
async def attr(self, sel: str, attr: str) -> str:
try:
val = await self._page.locator(sel).first.get_attribute(attr)
return val or ""
except Exception as e:
logger.debug("attr '%s' on '%s' failed: %s", attr, sel, e)
return ""
async def html(self) -> str:
try:
return await self._page.content()
except Exception as e:
logger.debug("html() failed: %s", e)
return ""
async def url(self) -> str:
try:
return self._page.url
except Exception as e:
logger.debug("url() failed: %s", e)
return ""
async def title(self) -> str:
try:
return await self._page.title()
except Exception as e:
logger.debug("title() failed: %s", e)
return ""
async def reload(self) -> None:
try:
await self._page.reload(timeout=15000)
await self._idle()
await self.capture("reload", force=True)
except Exception as e:
logger.debug("reload failed: %s", e)
Reference in New Issue View Git Blame Copy Permalink