feat(rpa_helpers): add browser.launch_browser_with_profile and close_browser
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
74
scripts/service/rpa_helpers/browser.py
Normal file
74
scripts/service/rpa_helpers/browser.py
Normal file
@@ -0,0 +1,74 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
"""Unified Chromium launch with stealth + persistent profile for RPA helpers."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from playwright.sync_api import sync_playwright
|
||||||
|
|
||||||
|
from service.browser_service import resolve_chromium_channel
|
||||||
|
from util.playwright_stealth import (
|
||||||
|
STEALTH_INIT_SCRIPT,
|
||||||
|
persistent_context_launch_parts,
|
||||||
|
stealth_enabled,
|
||||||
|
)
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from playwright.sync_api import BrowserContext, Page
|
||||||
|
|
||||||
|
|
||||||
|
def launch_browser_with_profile(
|
||||||
|
profile_dir: str,
|
||||||
|
*,
|
||||||
|
channel: str | None = None,
|
||||||
|
headless: bool = False,
|
||||||
|
extra_args: list[str] | None = None,
|
||||||
|
locale: str = "zh-CN",
|
||||||
|
) -> tuple["BrowserContext", "Page"]:
|
||||||
|
"""启动浏览器(Chrome 优先 / Edge fallback),返回 (context, page)。
|
||||||
|
|
||||||
|
调用方应在 finally 中调用 close_browser(context)。
|
||||||
|
"""
|
||||||
|
os.makedirs(profile_dir, exist_ok=True)
|
||||||
|
|
||||||
|
resolved_channel = channel if channel is not None else resolve_chromium_channel()
|
||||||
|
if not resolved_channel:
|
||||||
|
raise RuntimeError("no chrome/edge found on system")
|
||||||
|
|
||||||
|
_pw = sync_playwright().start()
|
||||||
|
try:
|
||||||
|
args, ignore_automation = persistent_context_launch_parts(extra_args=extra_args)
|
||||||
|
launch_kwargs: dict = dict(
|
||||||
|
user_data_dir=profile_dir,
|
||||||
|
headless=headless,
|
||||||
|
channel=resolved_channel,
|
||||||
|
no_viewport=True,
|
||||||
|
locale=locale,
|
||||||
|
args=args,
|
||||||
|
)
|
||||||
|
if ignore_automation is not None:
|
||||||
|
launch_kwargs["ignore_default_args"] = ignore_automation
|
||||||
|
ctx = _pw.chromium.launch_persistent_context(**launch_kwargs)
|
||||||
|
if stealth_enabled():
|
||||||
|
ctx.add_init_script(STEALTH_INIT_SCRIPT)
|
||||||
|
page = ctx.pages[0] if ctx.pages else ctx.new_page()
|
||||||
|
setattr(ctx, "_openclaw_pw", _pw)
|
||||||
|
return ctx, page
|
||||||
|
except Exception:
|
||||||
|
_pw.stop()
|
||||||
|
raise
|
||||||
|
|
||||||
|
|
||||||
|
def close_browser(ctx) -> None:
|
||||||
|
"""关闭 context 并清理 Playwright manager。"""
|
||||||
|
pw = getattr(ctx, "_openclaw_pw", None)
|
||||||
|
try:
|
||||||
|
ctx.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
if pw is not None:
|
||||||
|
try:
|
||||||
|
pw.stop()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
Reference in New Issue
Block a user