xhs-note-crawling/workplace/playwright.py

"""
Playwright 公共入口：项目根路径、NODE_OPTIONS、同步 API 与 browser_core 的 CDP 封装。
其它脚本请先保证项目根在 sys.path，再 `from workplace.playwright import ...`。
"""

from __future__ import annotations

import sys
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Iterator


def ensure_repo_on_path() -> Path:
    """将项目根目录加入 sys.path。"""
    root = Path(__file__).resolve().parent.parent
    if str(root) not in sys.path:
        sys.path.insert(0, str(root))
    return root


ensure_repo_on_path()

from workplace.playwright_node_environment import (  # noqa: E402
    ensure_playwright_driver_node_options,
)

ensure_playwright_driver_node_options()

from playwright.sync_api import sync_playwright  # noqa: E402

from workplace.cdp_chrome_browser_core import (  # noqa: E402
    _is_blank_url,
    connect_cdp,
    pick_target_page,
    repo_root,
)

DEFAULT_CDP_URL = "http://127.0.0.1:9222"


def init_singleton() -> None:
    """
    应用入口显式初始化：保证仓库在 path、Node/Playwright 驱动环境就绪。
    不启动浏览器进程；可多次调用。
    """
    ensure_repo_on_path()
    ensure_playwright_driver_node_options()


@contextmanager
def sync_playwright_singleton() -> Iterator[Any]:
    """
    单次自动化流程内共用一个 Playwright 同步实例：进入时 ``start()``，退出时 ``stop()``。
    与 ``with sync_playwright() as p`` 等价，供 ``main.start`` 集中管理生命周期。
    """
    init_singleton()
    pw = sync_playwright().start()
    try:
        yield pw
    finally:
        pw.stop()


def default_profile_dir() -> Path:
    return repo_root() / ".chrome-xhs-debug-profile"


def connect_browser_via_cdp(
    playwright_inst,
    *,
    cdp: str | None = None,
    auto_chrome: bool = True,
    profile_dir: Path | None = None,
):
    """附着 CDP；缺省 CDP URL 与调试版 Chrome 用户目录同本仓库惯例。"""
    return connect_cdp(
        playwright_inst,
        cdp or DEFAULT_CDP_URL,
        auto_chrome=auto_chrome,
        profile_dir=profile_dir or default_profile_dir(),
    )