Files
MoviePilot/app/adapters/network/browser.py
Aqr-K 6610a8c008 fix: retain plugin lifecycle ownership and recover installation state (#7146)
* fix: retain plugin lifecycle ownership and recover installation state

* test: cover plugin rollback and uninstall lifecycle boundaries

---------

Co-authored-by: jxxghp <jxxghp@gmail.com>
2026-10-23 07:44:07 +08:00

1489 lines
55 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
浏览器启动、可复用会话和站点页面访问适配,负责关闭所持有的页面与上下文。
"""
import ipaddress
import threading
import time
import uuid
from concurrent.futures import Future, ThreadPoolExecutor
from contextvars import Context, copy_context
from dataclasses import dataclass, field
from typing import Any, Callable, Optional, Protocol, cast
from urllib.parse import urlparse
from app.adapters.network.http import RequestUtils, cookie_parse
from app.runtime.log import logger
from app.runtime.resources import (
acquire_managed_resource,
acquire_managed_resource_async,
)
from app.runtime.settings import get_runtime_setting
class BrowserElement(Protocol):
"""
页面元素的最小接口,避免为了类型标注直接导入 Playwright。
"""
def is_visible(self) -> bool:
"""
检查当前元素是否可见。
:return: 元素满足浏览器可见性规则时返回 True,否则返回 False
"""
...
def fill(self, value: str) -> None:
"""
替换元素中的文本内容。
:param value: 要填写的完整文本;空字符串用于清空内容
"""
...
def inner_text(self) -> str:
"""
读取元素渲染后的可见文本。
:return: 当前元素的可见文本,允许为空字符串
"""
...
class BrowserContext(Protocol):
"""
CloakBrowser 返回的上下文只需要满足这些能力即可。
"""
def new_page(self) -> "BrowserPage":
"""
在当前上下文中创建页面。
:return: 新页面,由调用方或上下文关闭流程释放
"""
...
def cookies(self) -> list[dict[str, Any]]:
"""
读取上下文当前保存的 Cookie。
:return: 浏览器 Cookie 对象列表,没有 Cookie 时为空列表
"""
...
def close(self) -> None:
"""
关闭上下文及其页面,释放该上下文持有的浏览器资源。
"""
...
class BrowserPage(Protocol):
"""
同步浏览器页面接口,覆盖页面会话和站点登录所需能力。
实现方沿用浏览器的超时和异常行为;页面只可在所属上下文和线程仍有效时使用。
:ivar context: 创建此页面的浏览器上下文,负责页面共享会话状态
:ivar url: 当前页面地址,导航后随页面状态更新
"""
context: BrowserContext
url: str
def set_extra_http_headers(self, headers: dict[str, str]) -> None:
"""
设置此页面后续请求使用的额外 HTTP 请求头。
:param headers: 请求头名称与文本值的映射
"""
...
def set_default_timeout(self, timeout: int) -> None:
"""
设置页面操作的默认等待上限。
:param timeout: 超时时间,单位为毫秒;0 沿用浏览器禁用超时的语义
"""
...
def goto(self, url: str, *args: Any, **kwargs: Any) -> Any:
"""
导航到目标地址并按指定条件等待加载。
:param url: 目标页面地址
:param args: 导航接口保留的位置参数
:param kwargs: 导航等待条件、超时(毫秒)或来源页等选项
:return: 最终导航响应对象;没有发生可返回响应的导航时可能为 None
"""
...
def wait_for_load_state(self, state: str, *args: Any, **kwargs: Any) -> Any:
"""
等待页面达到指定加载状态。
:param state: 浏览器加载状态,例如 load、domcontentloaded 或 networkidle
:param args: 加载等待接口保留的位置参数
:param kwargs: 等待超时(毫秒)等加载选项
:return: 等待完成后不返回业务数据;超时沿用浏览器异常
"""
...
def wait_for_selector(self, selector: str, *args: Any, **kwargs: Any) -> Any:
"""
等待选择器对应元素达到指定状态。
:param selector: 定位目标元素的选择器
:param args: 元素等待接口保留的位置参数
:param kwargs: 目标状态、严格匹配或超时(毫秒)等选项
:return: 匹配元素;等待隐藏或移除状态时可能为 None
"""
...
def fill(self, selector: str, value: str, *args: Any, **kwargs: Any) -> Any:
"""
定位输入元素并替换文本。
:param selector: 定位输入元素的选择器
:param value: 要填入的完整文本,空字符串用于清空
:param args: 文本填写接口保留的位置参数
:param kwargs: 强制执行、严格匹配或超时(毫秒)等选项
:return: 操作完成后不返回业务数据;定位或填写失败沿用浏览器异常
"""
...
def click(self, selector: str, *args: Any, **kwargs: Any) -> Any:
"""
定位元素并执行点击。
:param selector: 定位可点击元素的选择器
:param args: 点击接口保留的位置参数
:param kwargs: 鼠标按键、点击次数、位置或超时(毫秒)等选项
:return: 操作完成后不返回业务数据;定位或点击失败沿用浏览器异常
"""
...
def select_option(self, selector: str, *args: Any, **kwargs: Any) -> Any:
"""
设置下拉框选项。
:param selector: 定位下拉框的选择器
:param args: 传给下拉框接口的选项值
:param kwargs: 选项值、标签、索引或超时(毫秒)等选项
:return: 选中选项的值列表,沿用浏览器接口的返回格式
"""
...
def query_selector(self, selector: str) -> Optional[BrowserElement]:
"""
查找首个匹配选择器的元素。
:param selector: 浏览器支持的元素选择器
:return: 首个匹配元素;没有匹配项时返回 None
"""
...
def title(self) -> str:
"""
读取当前页面标题。
:return: 页面标题文本,允许为空字符串
"""
...
def inner_text(self, selector: str) -> str:
"""
读取选择器对应元素的可见文本。
:param selector: 定位目标元素的选择器
:return: 目标元素的可见文本;定位失败沿用浏览器异常
"""
...
def content(self) -> str:
"""
读取当前页面完整 HTML 源码。
:return: 包含文档类型声明的页面源码;导航期间读取可能抛出浏览器异常
"""
...
def evaluate(self, expression: str, *args: Any, **kwargs: Any) -> Any:
"""
在当前页面中执行 JavaScript 表达式。
:param expression: 待执行的 JavaScript 表达式或函数文本
:param args: 传入 JavaScript 函数的参数
:param kwargs: 浏览器表达式执行接口接受的关键字参数
:return: 表达式结果,按浏览器接口的序列化规则转换为 Python 对象
"""
...
def screenshot(self, *args: Any, **kwargs: Any) -> bytes:
"""
截取当前页面图像。
:param args: 截图接口保留的位置参数
:param kwargs: 图像格式、质量、全页范围或超时(毫秒)等选项
:return: 截图的二进制内容,格式由请求选项决定
"""
...
def close(self) -> None:
"""
关闭当前页面,所属上下文及其他页面保持有效。
"""
...
def launch_browser_context(headless: bool = True, **kwargs: Any) -> BrowserContext:
"""
启动同步浏览器上下文;有界面模式先显式获取宿主显示资源。
:param headless: 是否使用无头模式
:param kwargs: 浏览器实现接受的其余启动参数
:return: 浏览器上下文
"""
if not headless:
acquire_managed_resource(
"host.display",
reason="headed_browser_launch",
retry=True,
)
from cloakbrowser import launch_context
return cast(BrowserContext, launch_context(headless=headless, **kwargs))
async def launch_browser_context_async(
headless: bool = True,
**kwargs: Any,
) -> Any:
"""
启动异步浏览器上下文;有界面模式等待宿主显示资源就绪。
:param headless: 是否使用无头模式
:param kwargs: 浏览器实现接受的其余启动参数
:return: 浏览器上下文
"""
if not headless:
await acquire_managed_resource_async(
"host.display",
reason="headed_browser_launch",
retry=True,
)
from cloakbrowser import launch_context_async
return await launch_context_async(headless=headless, **kwargs)
@dataclass
class _BrowserSessionState:
"""
保存可复用浏览器上下文、页面游标与会话资源归属。
:param session_key: 会话唯一键,用于索引复用会话和专属工作线程
:param context: 会话持有的浏览器上下文,销毁会话时关闭
:param pages: 会话打开的页面列表
:param active_index: 当前活动页在 pages 中的索引,默认 0
:param user_agent: 创建会话时使用的浏览器标识;None 表示使用实现默认值
:param cookies: 用于会话的 Cookie 请求头文本;None 表示未额外指定
:param owner: 持有会话资源的运行时登记对象;None 表示未登记
:param created_at: 创建时间,使用单调时钟的秒值
:param last_used_at: 最近使用时间,使用单调时钟的秒值,供空闲过期判断
:param lock: 保护会话状态和页面操作的可重入锁,默认每个会话独立创建
"""
session_key: str
context: BrowserContext
pages: list[BrowserPage]
active_index: int = 0
user_agent: Optional[str] = None
cookies: Optional[str] = None
owner: Optional[Any] = field(default=None, repr=False)
created_at: float = field(default_factory=time.monotonic)
last_used_at: float = field(default_factory=time.monotonic)
lock: threading.RLock = field(default_factory=threading.RLock)
@property
def active_page(self) -> BrowserPage:
"""
取得活动索引对应的页面。
:return: 当前活动页面;调用方需保证页面列表非空且索引有效
"""
return self.pages[self.active_index]
class BrowserSessionHelper:
"""
Agent 浏览器会话辅助类,负责复用 CloakBrowser 上下文并生成可操作页面快照。
"""
SESSION_TTL_SECONDS = 15 * 60
MAX_SESSIONS = 8
DEFAULT_VIEWPORT = {"width": 1280, "height": 720}
PRIVATE_HOST_SUFFIXES = (".localhost", ".local", ".lan", ".home", ".internal")
PRIVATE_HOSTNAMES = {"localhost", "ip6-localhost", "ip6-loopback"}
REF_ATTRIBUTE = "data-moviepilot-agent-ref"
SESSION_WORKER_NAME_PREFIX = "browser-session"
_sessions: dict[str, _BrowserSessionState] = {}
_session_executors: dict[str, ThreadPoolExecutor] = {}
_session_executor_owners: dict[str, Any] = {}
_session_thread_ids: dict[str, int] = {}
_sessions_lock = threading.RLock()
def __init__(self, headless: bool = True, viewport: Optional[dict[str, int]] = None):
"""
初始化浏览器会话辅助类。
:param headless: 是否使用无头浏览器
:param viewport: 默认视口大小
"""
self.headless = headless
self.viewport = viewport or self.DEFAULT_VIEWPORT
@classmethod
def validate_url(cls, url: str, allow_private_network: bool = False) -> str:
"""
校验浏览器可访问的 URL,默认拒绝本机、私网和非 HTTP 协议。
:param url: 待访问的 URL
:param allow_private_network: 是否允许访问本机或私网地址
:return: 原始 URL
"""
parsed = urlparse(url or "")
if parsed.scheme not in {"http", "https"}:
raise ValueError("仅支持 http/https URL")
if not parsed.hostname:
raise ValueError("URL 缺少主机名")
hostname = parsed.hostname.lower().rstrip(".")
if allow_private_network:
return url
if hostname in cls.PRIVATE_HOSTNAMES or hostname.endswith(
cls.PRIVATE_HOST_SUFFIXES
):
raise ValueError("默认不允许访问本机或私网地址")
try:
ip_address = ipaddress.ip_address(hostname)
except ValueError:
return url
if not ip_address.is_global:
raise ValueError("默认不允许访问本机或私网地址")
return url
@classmethod
def ref_to_selector(cls, ref: str) -> str:
"""
将页面快照中的元素引用转换为稳定选择器。
:param ref: 快照返回的元素引用
:return: 可传给浏览器的属性选择器
"""
clean_ref = (ref or "").strip()
if not clean_ref:
raise ValueError("元素 ref 不能为空")
escaped_ref = clean_ref.replace("\\", "\\\\").replace('"', '\\"')
return f'[{cls.REF_ATTRIBUTE}="{escaped_ref}"]'
@classmethod
def close_all_sessions(cls) -> None:
"""
关闭所有 Agent 浏览器会话。
"""
with cls._sessions_lock:
session_keys = set(cls._sessions) | set(cls._session_executors)
for session_key in session_keys:
with cls._sessions_lock:
session = cls._sessions.get(session_key)
owner = session.owner if session else cls._session_executor_owners.get(session_key)
cls.close_session(session_key, owner=owner)
@classmethod
def close_session(cls, session_key: str, owner: Optional[Any] = None) -> bool:
"""
关闭指定 Agent 浏览器会话。
:param session_key: 会话标识
:param owner: 可选宿主作用域;提供时只允许关闭该作用域拥有的会话
:return: 找到并关闭会话时返回 True
"""
if cls._is_current_session_thread(session_key):
closed = cls._close_session_in_thread(session_key, owner=owner)
cls._shutdown_session_executor(session_key, wait=False)
return closed
executor = cls._get_existing_session_executor(session_key, owner=owner)
if executor:
context = copy_context()
# 会话线程保持空底层上下文,每次操作只使用当前调用快照。
def close_in_context() -> bool:
"""
在调用方上下文中关闭会话,保持会话线程底层上下文为空。
"""
return bool(context.run(
cls._run_session_task, session_key, cls._close_session_in_thread,
session_key, owner=owner,
))
future: Future[bool] = Context().run(executor.submit, close_in_context)
try:
return future.result()
finally:
cls._shutdown_session_executor(session_key)
closed = cls._close_session_in_thread(session_key, owner=owner)
cls._shutdown_session_executor(session_key)
return closed
@classmethod
def close_owner(cls, owner: Any) -> bool:
"""
关闭宿主作用域创建的全部浏览器会话和其专属执行线程。
"""
with cls._sessions_lock:
session_keys = {
key for key, session in cls._sessions.items() if session.owner is owner
}
session_keys.update(
key for key, session_owner in cls._session_executor_owners.items()
if session_owner is owner
)
for session_key in session_keys:
cls.close_session(session_key, owner=owner)
with cls._sessions_lock:
return not any(
session.owner is owner
for session in cls._sessions.values()
) and not any(
session_owner is owner
for session_owner in cls._session_executor_owners.values()
)
def with_session(
self,
session_key: str,
callback: Callable[[_BrowserSessionState], Any],
user_agent: Optional[str] = None,
cookies: Optional[str] = None,
timeout: Optional[int] = 30,
owner: Optional[Any] = None,
) -> Any:
"""
获取或创建浏览器会话,并在持有会话锁时执行回调。
:param session_key: 会话标识
:param callback: 使用浏览器会话执行操作的回调函数
:param user_agent: 新建会话时使用的 User-Agent
:param cookies: 本次操作要注入的 Cookie 请求头
:param timeout: 默认操作超时时间,单位秒
:param owner: 可选宿主作用域;提供时会隔离并校验会话归属
:return: 回调函数返回值
"""
# 会话线程只能执行自己的浏览器任务;不能在其中同步等待其他会话关闭。
with self._sessions_lock:
in_session_thread = threading.get_ident() in self._session_thread_ids.values()
if not in_session_thread:
self._prune_sessions()
try:
return self._run_in_session_thread(
session_key,
self._with_session_in_thread,
session_key,
callback,
user_agent=user_agent,
cookies=cookies,
timeout=timeout,
owner=owner,
)
finally:
if not in_session_thread:
self._enforce_session_limit(protect_session_key=session_key)
def _with_session_in_thread(
self,
session_key: str,
callback: Callable[[_BrowserSessionState], Any],
user_agent: Optional[str] = None,
cookies: Optional[str] = None,
timeout: Optional[int] = 30,
owner: Optional[Any] = None,
) -> Any:
"""
在会话专属线程内获取浏览器会话并执行回调。
:param session_key: 会话标识
:param callback: 使用浏览器会话执行操作的回调函数
:param user_agent: 新建会话时使用的 User-Agent
:param cookies: 本次操作要注入的 Cookie 请求头
:param timeout: 默认操作超时时间,单位秒
:return: 回调函数返回值
"""
session = self._get_or_create_session(
session_key=session_key,
user_agent=user_agent,
cookies=cookies,
owner=owner,
)
with session.lock:
session.last_used_at = time.monotonic()
if timeout and hasattr(session.active_page, "set_default_timeout"):
session.active_page.set_default_timeout(int(timeout) * 1000)
if cookies:
session.cookies = cookies
session.active_page.set_extra_http_headers({"cookie": cookies})
return callback(session)
@classmethod
def _run_in_session_thread(
cls,
session_key: str,
callback: Callable[..., Any],
*args: Any,
owner: Optional[Any] = None,
**kwargs: Any,
) -> Any:
"""
将浏览器同步 API 调用投递到当前会话固定的单线程执行器。
:param session_key: 会话标识
:param callback: 需要在会话线程中运行的回调
:param args: 回调位置参数
:param kwargs: 回调关键字参数
:return: 回调返回值
"""
if cls._is_current_session_thread(session_key):
return callback(*args, owner=owner, **kwargs)
for _ in range(2):
executor = cls._get_session_executor(session_key, owner=owner)
try:
context = copy_context()
# 会话线程保持空底层上下文,每次操作只使用当前调用快照。
def invoke_in_context() -> Any:
"""
在当前调用方快照中执行会话操作,避免上下文在会话线程间泄漏。
"""
return context.run(
cls._run_session_task, session_key, callback,
*args, owner=owner, **kwargs,
)
future: Future[Any] = Context().run(executor.submit, invoke_in_context)
except RuntimeError:
cls._discard_session_executor(session_key, executor)
continue
try:
return future.result()
except Exception:
with cls._sessions_lock:
has_session = session_key in cls._sessions
if not has_session:
cls._shutdown_session_executor(session_key)
raise
raise RuntimeError("浏览器会话线程已关闭")
@classmethod
def _run_session_task(
cls,
session_key: str,
callback: Callable[..., Any],
*args: Any,
**kwargs: Any,
) -> Any:
"""
记录当前会话工作线程后执行实际任务。
:param session_key: 会话标识
:param callback: 需要执行的回调
:param args: 回调位置参数
:param kwargs: 回调关键字参数
:return: 回调返回值
"""
with cls._sessions_lock:
cls._session_thread_ids[session_key] = threading.get_ident()
return callback(*args, **kwargs)
@classmethod
def _is_current_session_thread(cls, session_key: str) -> bool:
"""
判断当前代码是否已运行在指定会话的固定线程内。
:param session_key: 会话标识
:return: 当前线程是会话工作线程时返回 True
"""
with cls._sessions_lock:
thread_id = cls._session_thread_ids.get(session_key)
return thread_id == threading.get_ident()
@classmethod
def _get_session_executor(cls, session_key: str, owner: Optional[Any] = None) -> ThreadPoolExecutor:
"""
获取或创建指定会话的单线程执行器。
:param session_key: 会话标识
:return: 会话专属执行器
"""
with cls._sessions_lock:
executor = cls._session_executors.get(session_key)
if executor:
existing_owner = cls._session_executor_owners.get(session_key)
if existing_owner is not owner:
raise PermissionError("浏览器会话不属于当前任务")
return executor
executor = ThreadPoolExecutor(
max_workers=1,
thread_name_prefix=cls.SESSION_WORKER_NAME_PREFIX,
)
cls._session_executors[session_key] = executor
cls._session_executor_owners[session_key] = owner
return executor
@classmethod
def _get_existing_session_executor(
cls, session_key: str, owner: Optional[Any] = None
) -> Optional[ThreadPoolExecutor]:
"""
获取指定会话已存在的执行器。
:param session_key: 会话标识
:return: 已存在的执行器,不存在时返回 None
"""
with cls._sessions_lock:
executor = cls._session_executors.get(session_key)
if executor and cls._session_executor_owners.get(session_key) is not owner:
raise PermissionError("浏览器会话不属于当前任务")
return executor
@classmethod
def _discard_session_executor(
cls,
session_key: str,
executor: ThreadPoolExecutor,
) -> None:
"""
丢弃已经关闭的会话执行器。
:param session_key: 会话标识
:param executor: 需要从缓存中移除的执行器
"""
with cls._sessions_lock:
if cls._session_executors.get(session_key) is executor:
cls._session_executors.pop(session_key, None)
cls._session_executor_owners.pop(session_key, None)
@classmethod
def _shutdown_session_executor(
cls,
session_key: str,
wait: bool = True,
) -> None:
"""
关闭并移除指定会话的执行器。
:param session_key: 会话标识
:param wait: 是否等待工作线程退出
"""
with cls._sessions_lock:
executor = cls._session_executors.pop(session_key, None)
cls._session_executor_owners.pop(session_key, None)
cls._session_thread_ids.pop(session_key, None)
if executor:
executor.shutdown(wait=wait, cancel_futures=True)
@classmethod
def _close_session_in_thread(cls, session_key: str, owner: Optional[Any] = None) -> bool:
"""
在会话固定线程内关闭并移除浏览器会话。
:param session_key: 会话标识
:param owner: 可选宿主作用域;提供时校验会话归属
:return: 找到并关闭会话时返回 True
"""
with cls._sessions_lock:
session = cls._sessions.get(session_key)
if session and session.owner is not owner:
raise PermissionError("浏览器会话不属于当前任务")
session = cls._sessions.pop(session_key, None)
if not session:
return False
cls._close_session_state(session)
return True
def open_tab(
self,
session: _BrowserSessionState,
url: Optional[str] = None,
timeout: Optional[int] = 30,
allow_private_network: bool = False,
) -> BrowserPage:
"""
在当前会话中新建标签页,并可选导航到指定 URL。
:param session: 当前浏览器会话
:param url: 可选的目标 URL
:param timeout: 导航超时时间,单位秒
:param allow_private_network: 是否允许访问本机或私网地址
:return: 新建的页面对象
"""
page = session.context.new_page()
if timeout and hasattr(page, "set_default_timeout"):
page.set_default_timeout(int(timeout) * 1000)
if session.cookies:
page.set_extra_http_headers({"cookie": session.cookies})
session.pages.append(page)
session.active_index = len(session.pages) - 1
if url:
self.goto(
page,
url,
timeout=timeout,
allow_private_network=allow_private_network,
)
return page
@staticmethod
def list_tabs(session: _BrowserSessionState) -> list[dict[str, Any]]:
"""
列出当前浏览器会话中的标签页。
:param session: 当前浏览器会话
:return: 标签页摘要列表
"""
tabs = []
for index, page in enumerate(session.pages):
tabs.append(
{
"index": index,
"active": index == session.active_index,
"url": getattr(page, "url", ""),
"title": BrowserSessionHelper._safe_page_title(page),
}
)
return tabs
@staticmethod
def focus_tab(session: _BrowserSessionState, tab_index: int) -> BrowserPage:
"""
切换当前会话的活动标签页。
:param session: 当前浏览器会话
:param tab_index: 标签页索引
:return: 切换后的页面对象
"""
if tab_index < 0 or tab_index >= len(session.pages):
raise ValueError(f"标签页索引不存在: {tab_index}")
session.active_index = tab_index
return session.active_page
@staticmethod
def close_tab(session: _BrowserSessionState, tab_index: int) -> list[dict[str, Any]]:
"""
关闭当前会话中的指定标签页。
:param session: 当前浏览器会话
:param tab_index: 标签页索引
:return: 关闭后的标签页列表
"""
if tab_index < 0 or tab_index >= len(session.pages):
raise ValueError(f"标签页索引不存在: {tab_index}")
active_index = session.active_index
page = session.pages.pop(tab_index)
close_error: Exception | None = None
try:
page.close()
except Exception as err:
close_error = err
logger.warning(f"关闭浏览器标签页失败: {str(err)}")
if not session.pages:
session.pages.append(session.context.new_page())
elif tab_index < active_index:
active_index -= 1
session.active_index = min(active_index, len(session.pages) - 1)
if close_error is not None:
raise RuntimeError("关闭浏览器标签页失败") from close_error
return BrowserSessionHelper.list_tabs(session)
@staticmethod
def get_cookies(
session: _BrowserSessionState,
page: Optional[BrowserPage] = None,
) -> dict[str, Any]:
"""
读取活动页面所属域名的 Cookie、User-Agent 和页面地址。
"""
active_page = page or session.active_page
current_url = getattr(active_page, "url", "") or ""
current_host = (urlparse(current_url).hostname or "").lower().rstrip(".")
values: dict[str, str] = {}
if session.cookies:
injected = cookie_parse(session.cookies)
if isinstance(injected, dict):
values.update({str(key): str(value) for key, value in injected.items()})
for cookie in session.context.cookies() or []:
if not isinstance(cookie, dict):
continue
name = cookie.get("name")
value = cookie.get("value")
if name is None or value is None:
continue
domain = str(cookie.get("domain") or "").lower().lstrip(".").rstrip(".")
if current_host and domain and current_host != domain and not current_host.endswith(f".{domain}"):
continue
values[str(name)] = str(value)
user_agent = session.user_agent
if not user_agent:
try:
user_agent = str(
active_page.evaluate("() => window.navigator.userAgent") or ""
)
except Exception:
user_agent = ""
cookie_header = "; ".join(f"{name}={value}" for name, value in values.items())
return {
"url": current_url,
"user_agent": user_agent or "",
"cookie": cookie_header,
"cookies": [
{"name": name, "value": value}
for name, value in values.items()
],
}
def goto(
self,
page: BrowserPage,
url: str,
timeout: Optional[int] = 30,
allow_private_network: bool = False,
) -> Any:
"""
校验并导航页面到指定 URL。
:param page: 页面对象
:param url: 目标 URL
:param timeout: 导航超时时间,单位秒
:param allow_private_network: 是否允许访问本机或私网地址
:return: 浏览器导航响应对象
"""
self.validate_url(url, allow_private_network=allow_private_network)
response = page.goto(
url,
wait_until="domcontentloaded",
timeout=int(timeout or 30) * 1000,
)
try:
page.wait_for_load_state(
"networkidle",
timeout=min(int(timeout or 30), 15) * 1000,
)
except Exception:
pass
self.validate_current_url(page, allow_private_network=allow_private_network)
return response
@classmethod
def validate_current_url(
cls, page: BrowserPage, allow_private_network: bool = False
) -> None:
"""
校验当前页面地址,捕获跳转后的不安全目标。
:param page: 页面对象
:param allow_private_network: 是否允许访问本机或私网地址
"""
current_url = getattr(page, "url", "")
if current_url and current_url.startswith(("http://", "https://")):
cls.validate_url(current_url, allow_private_network=allow_private_network)
@classmethod
def build_snapshot(
cls,
page: BrowserPage,
status: Optional[Any] = None,
max_text_chars: int = 8000,
max_elements: int = 40,
) -> dict[str, Any]:
"""
构建包含可读文本和可交互元素 ref 的页面快照。
:param page: 页面对象
:param status: 可选的导航状态码
:param max_text_chars: 页面文本最大返回长度
:param max_elements: 最大可交互元素数量
:return: 页面快照字典
"""
text_content = cls._safe_inner_text(page, "body")
elements = cls._extract_interactive_elements(page, max_elements=max_elements)
result: dict[str, Any] = {
"url": getattr(page, "url", ""),
"title": cls._safe_page_title(page),
"text_content": cls._truncate_text(text_content, max_text_chars),
"interactive_elements": elements,
}
if status is not None:
result["status"] = status
links = [
{
"ref": element.get("ref"),
"text": element.get("text"),
"href": element.get("href"),
}
for element in elements
if element.get("tag") == "a" and element.get("href")
][:30]
forms = [
element
for element in elements
if element.get("tag") in {"input", "textarea", "select", "button"}
][:30]
if links:
result["links"] = links
if forms:
result["form_elements"] = forms
return result
@staticmethod
def _launch_context(
headless: bool,
user_agent: Optional[str] = None,
viewport: Optional[dict[str, int]] = None,
) -> BrowserContext:
"""
按宿主反检测配置创建 CloakBrowser 上下文。
"""
context_kwargs = {
"humanize": get_runtime_setting('CLOAKBROWSER_HUMANIZE'),
"human_preset": get_runtime_setting('CLOAKBROWSER_HUMAN_PRESET'),
}
if user_agent:
context_kwargs["user_agent"] = user_agent
if viewport:
context_kwargs["viewport"] = viewport
return launch_browser_context(headless=headless, **context_kwargs)
def _get_or_create_session(
self,
session_key: str,
user_agent: Optional[str] = None,
cookies: Optional[str] = None,
owner: Optional[Any] = None,
) -> _BrowserSessionState:
"""
复用匹配会话,或为新的会话键创建上下文。
"""
self._validate_owner(owner)
with self._sessions_lock:
session = self._sessions.get(session_key)
if session and session.owner is not owner:
raise PermissionError("浏览器会话不属于当前任务")
if session and user_agent and session.user_agent != user_agent:
self._sessions.pop(session_key, None)
self._close_session_state(session)
session = None
if session:
return session
context = self._launch_context(
headless=self.headless,
user_agent=user_agent,
viewport=self.viewport,
)
page = context.new_page()
if cookies:
page.set_extra_http_headers({"cookie": cookies})
session = _BrowserSessionState(
session_key=session_key,
context=context,
pages=[page],
user_agent=user_agent,
cookies=cookies,
owner=owner,
)
with self._sessions_lock:
existing_session = self._sessions.get(session_key)
if existing_session:
if existing_session.owner is not owner:
self._close_session_state(session)
raise PermissionError("浏览器会话不属于当前任务")
self._close_session_state(session)
return existing_session
self._sessions[session_key] = session
return session
@staticmethod
def _validate_owner(owner: Optional[Any]) -> None:
"""
拒绝已封口作用域创建或复用浏览器会话。
"""
if owner is not None and bool(getattr(owner, "closed", False)):
raise PermissionError("浏览器会话所属任务已停止")
@classmethod
def _prune_sessions(cls) -> None:
"""
关闭超过空闲时限的浏览器会话。
"""
now = time.monotonic()
with cls._sessions_lock:
expired_keys = [
session_key
for session_key, session in cls._sessions.items()
if now - session.last_used_at > cls.SESSION_TTL_SECONDS
]
for session_key in expired_keys:
with cls._sessions_lock:
session = cls._sessions.get(session_key)
owner = session.owner if session else None
cls.close_session(session_key, owner=owner)
@classmethod
def _enforce_session_limit(cls, protect_session_key: Optional[str] = None) -> None:
"""
清理超过数量上限的旧会话。
:param protect_session_key: 本次刚创建、需要优先保留的会话标识
"""
while True:
with cls._sessions_lock:
if len(cls._sessions) <= cls.MAX_SESSIONS:
return
candidate_keys = [
session_key
for session_key in cls._sessions
if session_key != protect_session_key
]
if not candidate_keys:
return
oldest_key = min(
candidate_keys,
key=lambda key: cls._sessions[key].last_used_at,
)
oldest_owner = cls._sessions[oldest_key].owner
if not cls.close_session(oldest_key, owner=oldest_owner):
return
@staticmethod
def _close_session_state(session: _BrowserSessionState) -> None:
"""
尽力关闭会话中的全部页面和浏览器上下文。
"""
with session.lock:
for page in list(session.pages):
try:
page.close()
except Exception as err:
logger.warning(f"关闭浏览器页面失败: {str(err)}")
try:
session.context.close()
except Exception as err:
logger.warning(f"关闭浏览器上下文失败: {str(err)}")
@staticmethod
def _safe_page_title(page: BrowserPage) -> str:
"""
读取页面标题,页面失效时返回空字符串。
"""
try:
return page.title()
except Exception:
return ""
@staticmethod
def _safe_inner_text(page: BrowserPage, selector: str) -> str:
"""
读取节点文本,选择器或页面失效时返回空字符串。
"""
try:
return page.inner_text(selector)
except Exception:
return ""
@staticmethod
def _truncate_text(text: Optional[str], max_chars: int) -> str:
"""
按 Agent 上下文上限截断页面文本。
"""
if not text:
return ""
if len(text) <= max_chars:
return text
return text[:max_chars] + "\n\n...(内容已截断)"
@classmethod
def _extract_interactive_elements(
cls, page: BrowserPage, max_elements: int
) -> list[dict[str, Any]]:
"""
提取页面中可见且可交互的元素快照。
"""
script = f"""
() => {{
const limit = {int(max_elements)};
document.querySelectorAll('[{cls.REF_ATTRIBUTE}]').forEach((el) =>
el.removeAttribute('{cls.REF_ATTRIBUTE}'));
const selector = [
'a[href]',
'button',
'input',
'textarea',
'select',
'[role="button"]',
'[role="link"]',
'[onclick]',
'summary'
].join(',');
const isVisible = (el) => {{
const style = window.getComputedStyle(el);
const rect = el.getBoundingClientRect();
return style && style.visibility !== 'hidden'
&& style.display !== 'none'
&& rect.width > 0
&& rect.height > 0;
}};
return Array.from(document.querySelectorAll(selector))
.filter(isVisible)
.slice(0, limit)
.map((el, index) => {{
const ref = `e${{index + 1}}`;
el.setAttribute('{cls.REF_ATTRIBUTE}', ref);
const tag = el.tagName.toLowerCase();
const text = (
el.innerText
|| el.value
|| el.getAttribute('aria-label')
|| el.getAttribute('title')
|| el.getAttribute('placeholder')
|| ''
).trim();
return {{
ref,
tag,
type: el.type || '',
text: text.substring(0, 120),
name: el.name || '',
id: el.id || '',
role: el.getAttribute('role') || '',
placeholder: el.getAttribute('placeholder') || '',
href: el.href || '',
value: tag === 'select' ? '' : (el.value || '').substring(0, 80),
selector: `[${cls.REF_ATTRIBUTE}="${{ref}}"]`
}};
}});
}}
"""
try:
elements = page.evaluate(script)
except Exception as err:
logger.debug(f"提取页面可交互元素失败: {str(err)}")
return []
if not isinstance(elements, list):
return []
return elements
class PlaywrightHelper:
"""
为旧插件浏览器调用提供统一的 CloakBrowser 适配。
"""
def __init__(self, browser_type: Optional[str] = None, *args: Any, **kwargs: Any) -> None:
"""
兼容旧的 PlaywrightHelper(browser_type=...) 构造方式。
"""
self.browser_type = browser_type or get_runtime_setting(
"PLAYWRIGHT_BROWSER_TYPE"
)
@staticmethod
def __browser_emulation() -> str:
"""
当前浏览器仿真类型。
"""
return (
get_runtime_setting('BROWSER_EMULATION') or "cloakbrowser"
).lower()
@staticmethod
def __launch_cloakbrowser_context(headless: bool,
user_agent: Optional[str] = None,
proxies: Optional[dict[str, Any]] = None) -> BrowserContext:
"""
启动 CloakBrowser 上下文。
"""
return launch_browser_context(headless=headless,
proxy=proxies,
user_agent=user_agent,
humanize=get_runtime_setting('CLOAKBROWSER_HUMANIZE'),
human_preset=get_runtime_setting('CLOAKBROWSER_HUMAN_PRESET'))
@staticmethod
def __fs_cookie_str(cookies: list[dict[str, Any]]) -> str:
"""
将 FlareSolverr Cookie 列表转换为请求头字符串。
"""
if not cookies:
return ""
return "; ".join([f"{c.get('name')}={c.get('value')}" for c in cookies if c and c.get('name') is not None])
@staticmethod
def __flaresolverr_request(url: str,
cookies: Optional[str] = None,
proxy_config: Optional[dict[str, Any]] = None,
timeout: Optional[int] = 60) -> Optional[dict[str, Any]]:
"""
调用 FlareSolverr 解决 Cloudflare 并返回 solution 结果
参考: https://github.com/FlareSolverr/FlareSolverr
"""
flaresolverr_url = get_runtime_setting('FLARESOLVERR_URL')
if not flaresolverr_url:
logger.warn("未配置 FLARESOLVERR_URL,无法使用 FlareSolverr")
return None
fs_api = flaresolverr_url.rstrip("/") + "/v1"
session_id = None
try:
# 检查是否需要代理认证
need_proxy_auth = (proxy_config and proxy_config.get("server") and
(proxy_config.get("username") or proxy_config.get("password")))
if need_proxy_auth:
# 使用 session 模式支持代理认证
logger.debug("检测到flaresolverr代理需要认证,使用 session 模式")
session_id = str(uuid.uuid4())
create_payload: dict[str, Any] = {
"cmd": "sessions.create",
"session": session_id
}
# 添加代理配置到会话创建请求
if proxy_config and proxy_config.get("server"):
proxy_payload: dict[str, Any] = {"url": proxy_config["server"]}
if proxy_config.get("username"):
proxy_payload["username"] = proxy_config["username"]
if proxy_config.get("password"):
proxy_payload["password"] = proxy_config["password"]
create_payload["proxy"] = proxy_payload
create_result = RequestUtils(content_type="application/json",
timeout=timeout or 60).post_json(url=fs_api, json=create_payload)
if not create_result or create_result.get("status") != "ok":
logger.error(
f"创建 FlareSolverr 会话失败: {create_result.get('message') if create_result else '无响应'}")
return None
request_payload = {
"cmd": "request.get",
"url": url,
"session": session_id,
"maxTimeout": int(timeout or 60) * 1000,
}
else:
# 使用普通模式(无代理认证)
request_payload = {
"cmd": "request.get",
"url": url,
"maxTimeout": int(timeout or 60) * 1000,
}
# 添加代理配置(仅 URL,无认证)
if proxy_config and proxy_config.get("server"):
request_payload["proxy"] = {"url": proxy_config["server"]}
# 将 cookies 以数组形式传递给 FlareSolverr
if cookies:
try:
request_payload["cookies"] = cookie_parse(cookies, array=True)
except Exception as e:
logger.debug(f"解析 cookies 失败,忽略: {str(e)}")
data = RequestUtils(content_type="application/json",
timeout=timeout or 60).post_json(url=fs_api, json=request_payload)
if not data:
logger.error("FlareSolverr 返回空响应")
return None
if data.get("status") != "ok":
logger.error(f"FlareSolverr 调用失败: {data.get('message')}")
return None
return data.get("solution")
except Exception as e:
logger.error(f"调用 FlareSolverr 失败: {str(e)}")
return None
finally:
if session_id:
try:
destroy_payload = {
"cmd": "sessions.destroy",
"session": session_id
}
RequestUtils(content_type="application/json",
timeout=10).post_json(url=fs_api, json=destroy_payload)
logger.debug(f"已清理 FlareSolverr 会话: {session_id}")
except Exception as e:
logger.warning(f"清理 FlareSolverr 会话失败: {str(e)}")
def action(self, url: str,
callback: Callable[[BrowserPage], Any],
cookies: Optional[str] = None,
ua: Optional[str] = None,
proxies: Optional[dict[str, Any]] = None,
headless: Optional[bool] = False,
timeout: Optional[int] = 60) -> Any:
"""
访问页面并在受控浏览器会话内执行回调,结束后关闭页面与上下文。
配置 FlareSolverr 时优先使用其返回的 Cookie 和浏览器标识,回调不可把页面留到会话外使用。
异常会记录日志;关闭异常不会覆盖已经获得的回调结果。
:param url: 要访问的页面地址
:param callback: 接收页面对象并返回业务结果的同步回调
:param cookies: 初始 Cookie 请求头文本;可被 FlareSolverr 结果替代
:param ua: 浏览器标识;可被 FlareSolverr 结果替代
:param proxies: 浏览器代理配置;None 表示未指定
:param headless: 是否使用无头模式,默认使用有界面模式
:param timeout: 导航超时,单位为秒;None 或 0 回退到 60 秒
:return: 回调结果;初始化、导航或回调失败而未取得结果时返回 None
"""
timeout = timeout or 60
result = None
try:
context = None
page = None
try:
# 如果配置使用 FlareSolverr,先通过其获取清除后的 cookies 与 UA
fs_cookie_header = None
fs_ua = None
if self.__browser_emulation() == "flaresolverr":
solution = self.__flaresolverr_request(url=url, cookies=cookies,
proxy_config=proxies, timeout=timeout)
if solution:
fs_cookie_header = self.__fs_cookie_str(solution.get("cookies", []))
fs_ua = solution.get("userAgent")
context = self.__launch_cloakbrowser_context(headless=bool(headless),
user_agent=fs_ua or ua,
proxies=proxies)
page = context.new_page()
# 优先使用 FlareSolverr 返回,其次使用入参
merged_cookie = fs_cookie_header or cookies
if merged_cookie:
page.set_extra_http_headers({"cookie": merged_cookie})
page.goto(url, wait_until="domcontentloaded", timeout=timeout * 1000)
try:
# 登录页的统计与长连接请求可能持续存在,不应阻断已就绪表单的处理。
page.wait_for_load_state(
"networkidle",
timeout=min(timeout, 15) * 1000,
)
except Exception:
pass
result = callback(page)
except Exception as e:
logger.error(f"网页操作失败: {str(e)}")
finally:
if page:
page.close()
if context:
context.close()
except Exception as e:
logger.error(f"CloakBrowser初始化失败: {str(e)}")
return result
def get_page_source(self, url: str,
cookies: Optional[str] = None,
ua: Optional[str] = None,
proxies: Optional[dict[str, Any]] = None,
headless: Optional[bool] = False,
timeout: Optional[int] = 60) -> Optional[str]:
"""
获取网页源码,优先尝试配置的 FlareSolverr,再回退到浏览器读取。
浏览器路径会等待挑战页标题消失或达到期限,读取结束后释放页面和上下文。
:param url: 要读取的页面地址
:param cookies: 发送给页面的 Cookie 请求头文本
:param ua: 指定的浏览器标识,未指定时使用浏览器默认值
:param proxies: 浏览器代理配置;None 表示未指定
:param headless: 是否使用无头模式,默认使用有界面模式
:param timeout: 页面加载与挑战等待上限,单位为秒;None 或 0 回退到 60 秒
:return: 获取到的 HTML 文本;未取得源码时返回 None
"""
timeout = timeout or 60
source = None
# 如果配置为 FlareSolverr,则直接调用获取页面源码
if self.__browser_emulation() == "flaresolverr":
try:
solution = self.__flaresolverr_request(url=url, cookies=cookies,
proxy_config=proxies, timeout=timeout)
if solution:
return solution.get("response")
except Exception as e:
logger.error(f"FlareSolverr 获取源码失败: {str(e)}")
try:
context = None
page = None
try:
context = self.__launch_cloakbrowser_context(headless=bool(headless),
user_agent=ua,
proxies=proxies)
page = context.new_page()
if cookies:
page.set_extra_http_headers({"cookie": cookies})
page.goto(url, wait_until="load", timeout=timeout * 1000)
# 部分挑战页会持续轮询,无法进入 networkidle;加载后按标题判断是否仍需等待。
challenge_titles = ("just a moment", "请稍候", "loading")
deadline = time.time() + timeout
while time.time() < deadline:
try:
current_title = (page.title() or "").strip().lower()
except Exception:
current_title = ""
if current_title and not any(
t in current_title for t in challenge_titles):
break
time.sleep(2)
# 页面跳转中 content() 可能失败, 重试几次
source = None
for _attempt in range(5):
try:
source = page.content()
if source:
break
except Exception:
source = None
time.sleep(2)
except Exception as e:
logger.error(f"获取网页源码失败: {str(e)}")
source = None
finally:
# 确保资源被正确清理
if page:
page.close()
if context:
context.close()
except Exception as e:
logger.error(f"CloakBrowser初始化失败: {str(e)}")
return source