xhs-note-crawling/workplace/cdp_chrome_browser_core.py

#!/usr/bin/env python3
"""Chrome 远程调试附着与启动。"""

from __future__ import annotations

import socket
import subprocess
import sys
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

_SINGLETON_MODULE_SOURCE_FILE_PATH = Path(__file__).resolve()
_REPOSITORY_ROOT_DIRECTORY = _SINGLETON_MODULE_SOURCE_FILE_PATH.parent.parent

from workplace import pyautogui as _human_timing  # noqa: E402


def repo_root() -> Path:
    return _REPOSITORY_ROOT_DIRECTORY


def _is_blank_url(page_url: str) -> bool:
    normalized_url = (page_url or "").strip().lower()
    if normalized_url in ("", "about:blank"):
        return True
    if normalized_url in ("chrome://newtab/", "chrome://new-tab-page/"):
        return True
    if normalized_url.startswith("chrome://new-tab-page"):
        return True
    return False


def cdp_endpoint_open(cdp_browser_http_url: str) -> bool:
    raw = (cdp_browser_http_url or "").strip()
    if not raw:
        return False
    parsed = urlparse(raw)
    hostname = parsed.hostname or "127.0.0.1"
    port = parsed.port
    if port is None:
        scheme = (parsed.scheme or "http").lower()
        port = 443 if scheme == "https" else 80
    socket_handle = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    socket_handle.settimeout(2.0)
    connect_result = socket_handle.connect_ex((hostname, port))
    socket_handle.close()
    return connect_result == 0


def _remote_debugging_tcp_port_from_http_url(cdp_browser_http_url: str) -> int:
    parsed = urlparse((cdp_browser_http_url or "").strip())
    if parsed.port is not None:
        return int(parsed.port)
    scheme = (parsed.scheme or "http").lower()
    return 443 if scheme == "https" else 80


def _resolve_chrome_executable_path() -> Path:
    import os

    chrome_path_environment = os.environ.get("CHROME_PATH", "").strip()
    if chrome_path_environment:
        candidate_path = Path(chrome_path_environment)
        if candidate_path.is_file():
            return candidate_path
    program_files = Path(os.environ.get("ProgramFiles", r"C:\Program Files"))
    program_files_x86 = Path(os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"))
    local_app_data = Path(os.environ.get("LOCALAPPDATA", ""))
    search_list = [
        program_files / "Google" / "Chrome" / "Application" / "chrome.exe",
        program_files_x86 / "Google" / "Chrome" / "Application" / "chrome.exe",
        local_app_data / "Google" / "Chrome" / "Application" / "chrome.exe",
    ]
    for chrome_executable_candidate in search_list:
        if chrome_executable_candidate.is_file():
            return chrome_executable_candidate
    return program_files / "Google" / "Chrome" / "Application" / "chrome.exe"


def _start_chrome_process_with_remote_debugging_port(
    *,
    chrome_user_data_directory_path: Path,
    remote_debugging_tcp_port: int,
) -> None:
    chrome_executable_path = _resolve_chrome_executable_path()
    command_argument_list = [
        str(chrome_executable_path),
        f"--remote-debugging-port={remote_debugging_tcp_port}",
        f"--user-data-dir={chrome_user_data_directory_path}",
        "--no-first-run",
        "--no-default-browser-check",
    ]
    creation_flags = 0
    if sys.platform == "win32":
        creation_flags = subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP
    subprocess.Popen(
        command_argument_list,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        creationflags=creation_flags,
    )


def connect_cdp(
    playwright_inst: Any,
    cdp_browser_http_url: str,
    *,
    auto_chrome: bool,
    profile_dir: Path,
) -> Any:
    normalized_cdp_url = (cdp_browser_http_url or "").strip() or "http://127.0.0.1:9222"
    debugging_endpoint_ready = cdp_endpoint_open(normalized_cdp_url)
    if not debugging_endpoint_ready and auto_chrome:
        remote_debugging_tcp_port = _remote_debugging_tcp_port_from_http_url(
            normalized_cdp_url,
        )
        _start_chrome_process_with_remote_debugging_port(
            chrome_user_data_directory_path=profile_dir,
            remote_debugging_tcp_port=remote_debugging_tcp_port,
        )
        _human_timing.sleep_human_chrome_debug_process_ready()
    browser = playwright_inst.chromium.connect_over_cdp(normalized_cdp_url)
    return browser


def pick_target_page(browser: Any) -> Any:
    """优先选用 **空白 / 新标签页**，避免多标签时误附着到同站非首页而跳过 ``goto`` 主页。"""
    candidates: list[Any] = []
    for browser_context in browser.contexts:
        for page in browser_context.pages:
            if page is not None and not page.is_closed():
                candidates.append(page)
    if not candidates:
        if browser.contexts:
            return browser.contexts[0].new_page()
        return None
    for page in candidates:
        if _is_blank_url(page.url or ""):
            return page
    return candidates[0]


__all__ = [
    "_is_blank_url",
    "cdp_endpoint_open",
    "connect_cdp",
    "pick_target_page",
    "repo_root",
]