500 lines
22 KiB
Python
500 lines
22 KiB
Python
"""Cloudflare 会话管理器：自动获取并复用通过验证的浏览器会话
|
|
|
|
骏河屋网站受 Cloudflare 保护，首次访问需通过浏览器完成 JS 挑战。
|
|
本模块负责：
|
|
- 检测并等待 Cloudflare 挑战完成
|
|
- 保存通过验证后的浏览器存储状态（含 cf_clearance Cookie）
|
|
- 后续请求复用已验证会话，避免重复触发挑战
|
|
- 会话失效时支持主动失效和重建
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import logging
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
from uuid import uuid4
|
|
from pathlib import Path
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import CloudflareChallengeError, UpstreamBlockedError
|
|
from app.services.browser_pool import BrowserPool
|
|
from app.services.session_store import SessionState, SessionStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class VerifiedSession:
|
|
"""已通过 Cloudflare 验证的会话信息"""
|
|
session_id: str
|
|
storage_state: dict
|
|
cf_clearance: str | None # cf_clearance Cookie 值
|
|
html: str | None = None # 新创建会话时携带的页面 HTML，复用场景为 None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class WarmContext:
|
|
"""保活的浏览器上下文：通过 Cloudflare 挑战后长期驻留，供后续请求复用
|
|
|
|
cf_clearance 是上下文级 cookie：只要请求指纹一致（见 stealth 钩子集成），
|
|
复用通过挑战的上下文、在其内开新标签页访问目标页即可被 CF 接受。
|
|
同一上下文可并发开多个标签页；带引用计数以保证并发使用时安全退役。
|
|
"""
|
|
session_id: str
|
|
context: Any # Playwright BrowserContext
|
|
cf_clearance: str | None
|
|
created_at: float
|
|
last_used_at: float
|
|
in_use: int = 0 # 当前正在该上下文内使用页面的请求数（用于安全退役）
|
|
retired: bool = False # 已从注册表摘除、等待空闲后关闭
|
|
close_done: bool = False # 关闭只执行一次的标记
|
|
|
|
|
|
class CloudflareSessionManager:
|
|
"""Cloudflare 会话管理器，负责会话的获取、验证和失效"""
|
|
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
browser_pool: BrowserPool,
|
|
session_store: SessionStore,
|
|
):
|
|
self._settings = settings
|
|
self._browser_pool = browser_pool
|
|
self._session_store = session_store
|
|
|
|
# 保活上下文注册表：session_name -> WarmContext；构建锁防止并发重建风暴
|
|
self._warm_contexts: dict[str, WarmContext] = {}
|
|
self._build_locks: dict[str, asyncio.Lock] = {}
|
|
|
|
async def get_verified_session(self, target_url: str, force_refresh: bool = False) -> VerifiedSession:
|
|
"""获取一个已通过 Cloudflare 验证的会话
|
|
|
|
优先复用已有会话；当 force_refresh=True 或无可用会话时，
|
|
启动浏览器访问目标页面，等待 Cloudflare 挑战通过后保存会话。
|
|
|
|
Args:
|
|
target_url: 目标页面 URL，用于确定会话名称
|
|
force_refresh: 是否强制创建新会话（忽略已有会话）
|
|
|
|
Returns:
|
|
VerifiedSession: 已验证的会话信息
|
|
|
|
Raises:
|
|
UpstreamBlockedError: 目标站点返回阻断响应（如 1020/403）
|
|
CloudflareChallengeError: 等待 Cloudflare 挑战超时
|
|
"""
|
|
session_name = self._session_name_for_url(target_url)
|
|
if not force_refresh:
|
|
existing = await self._session_store.get_session(session_name)
|
|
if existing is not None:
|
|
logger.debug("复用已验证会话：session_name=%s session_id=%s", session_name, existing.session_id)
|
|
return self._to_verified_session(existing)
|
|
|
|
logger.debug("创建新会话：session_name=%s force_refresh=%s", session_name, force_refresh)
|
|
session = await self._build_with_browser_retry(
|
|
f"创建验证会话({session_name})",
|
|
lambda: self._create_verified_session(session_name=session_name, target_url=target_url),
|
|
)
|
|
return session
|
|
|
|
async def fetch_html(self, target_url: str) -> tuple[str, VerifiedSession]:
|
|
"""获取目标页面 HTML，基于「保活上下文」复用通过验证的浏览器会话
|
|
|
|
cf_clearance 无法跨上下文重放，因此保留通过挑战的上下文本身：
|
|
1. 已有保活上下文且未过期：在其内开新页面访问目标页（同上下文 clearance 有效）
|
|
2. 复用失败或无保活上下文：在构建锁内创建新的保活上下文（顺带完成本次访问）
|
|
|
|
Args:
|
|
target_url: 目标页面 URL
|
|
|
|
Returns:
|
|
(html, session) 元组，session 携带当前生效会话的 session_id 等信息
|
|
|
|
Raises:
|
|
UpstreamBlockedError: 重建后仍无法获取有效页面
|
|
CloudflareChallengeError: 等待 Cloudflare 挑战超时
|
|
ResourceBusyError: 等待页面槽位超时
|
|
"""
|
|
session_name = self._session_name_for_url(target_url)
|
|
|
|
# 1. 优先复用已有保活上下文
|
|
reused = await self._reuse_warm(session_name, target_url)
|
|
if reused is not None:
|
|
html, warm = reused
|
|
return html, self._warm_to_session(warm)
|
|
|
|
# 2. 在构建锁内创建/复用保活上下文，避免并发重建风暴
|
|
lock = self._build_locks.setdefault(session_name, asyncio.Lock())
|
|
async with lock:
|
|
# 双重检查：可能已被其他协程在等待锁期间建好
|
|
reused = await self._reuse_warm(session_name, target_url)
|
|
if reused is not None:
|
|
html, warm = reused
|
|
return html, self._warm_to_session(warm)
|
|
|
|
logger.info("创建保活上下文：session_name=%s", session_name)
|
|
warm, html = await self._build_with_browser_retry(
|
|
f"创建保活上下文({session_name})",
|
|
lambda: self._build_warm_context(session_name, target_url),
|
|
)
|
|
return html, self._warm_to_session(warm)
|
|
|
|
async def _build_with_browser_retry(self, what: str, factory: Any) -> Any:
|
|
"""执行 factory()；若因浏览器断链失败，重建浏览器后整段重试一次。
|
|
|
|
factory 每次都从零创建上下文与页面（_build_warm_context / _create_verified_session
|
|
均如此），故断链重建后重试是安全的。此层覆盖 new_page、goto、content 等
|
|
「上下文创建之后」才发生的断链——这些阶段浏览器池内逐操作的重试无法触及。
|
|
"""
|
|
try:
|
|
return await factory()
|
|
except Exception as exc:
|
|
if not BrowserPool.is_disconnect_error(exc):
|
|
raise
|
|
logger.warning("%s 因浏览器断链失败，重建后重试一次：err=%s", what, exc)
|
|
await self._browser_pool.ensure_ready()
|
|
return await factory()
|
|
|
|
async def _reuse_warm(self, session_name: str, target_url: str) -> tuple[str, WarmContext] | None:
|
|
"""尝试复用已有保活上下文。
|
|
|
|
过期或复用失败时退役（关闭）该上下文并返回 None；
|
|
成功时返回 (html, warm)。
|
|
"""
|
|
warm = self._warm_contexts.get(session_name)
|
|
if warm is None:
|
|
return None
|
|
if self._warm_expired(warm):
|
|
await self._retire_warm(session_name, warm)
|
|
return None
|
|
html = await self._fetch_via_warm(warm, target_url)
|
|
if html is None:
|
|
await self._retire_warm(session_name, warm)
|
|
return None
|
|
return html, warm
|
|
|
|
def _warm_expired(self, warm: WarmContext) -> bool:
|
|
"""保活上下文是否已退役或超过最大存活时长（沿用 session_ttl）。"""
|
|
if warm.retired:
|
|
return True
|
|
age = asyncio.get_running_loop().time() - warm.created_at
|
|
return age >= self._settings.session_ttl_seconds
|
|
|
|
async def _fetch_via_warm(self, warm: WarmContext, target_url: str) -> str | None:
|
|
"""在保活上下文内开新标签页访问目标页（同上下文 cookie 共享，clearance 复用）。
|
|
|
|
携带同站 Referer，使复用请求更像站内跳转。同一上下文可并发开多个标签页。
|
|
|
|
Returns:
|
|
通过验证的页面 HTML；若判定已失效（非 200 / 挑战未过 / 上下文异常），返回 None
|
|
"""
|
|
warm.in_use += 1
|
|
try:
|
|
async with self._browser_pool.page_in_context(warm.context) as page:
|
|
response = await page.goto(
|
|
target_url,
|
|
wait_until="domcontentloaded",
|
|
timeout=int(self._settings.request_timeout_seconds * 1000),
|
|
referer=self._settings.base_url,
|
|
)
|
|
status = response.status if response is not None else None
|
|
if status != 200:
|
|
logger.info("保活上下文复用导航非 200，判定失效：url=%s status=%s", target_url, status)
|
|
return None
|
|
|
|
# 首响应 200：极少数情况下仍是软挑战，用短超时给一次自动通过的机会
|
|
try:
|
|
await self._wait_for_clearance(
|
|
page, timeout_seconds=self._settings.cloudflare_reuse_wait_timeout_seconds
|
|
)
|
|
except (CloudflareChallengeError, UpstreamBlockedError):
|
|
logger.info("保活上下文复用软挑战未通过，判定失效：url=%s", target_url)
|
|
return None
|
|
|
|
warm.last_used_at = asyncio.get_running_loop().time()
|
|
return await page.content()
|
|
except Exception as exc:
|
|
# 上下文/浏览器已被关闭（如浏览器回收）或导航异常：视为失效，触发重建
|
|
logger.info("保活上下文复用异常，判定失效：url=%s err=%s", target_url, exc)
|
|
return None
|
|
finally:
|
|
warm.in_use -= 1
|
|
if warm.retired and warm.in_use <= 0:
|
|
await self._close_warm_context(warm)
|
|
|
|
async def _build_warm_context(self, session_name: str, target_url: str) -> tuple[WarmContext, str]:
|
|
"""创建一个通过 Cloudflare 挑战的保活上下文，并返回本次访问的页面 HTML。"""
|
|
# 退役并关闭可能残留的旧上下文（如已过期未复用的），避免覆盖注册表导致泄漏
|
|
stale = self._warm_contexts.get(session_name)
|
|
if stale is not None:
|
|
await self._retire_warm(session_name, stale)
|
|
|
|
context = await self._browser_pool.create_persistent_context()
|
|
try:
|
|
async with self._browser_pool.page_in_context(context) as page:
|
|
response = await page.goto(
|
|
target_url,
|
|
wait_until="domcontentloaded",
|
|
timeout=int(self._settings.request_timeout_seconds * 1000),
|
|
)
|
|
status = response.status if response is not None else None
|
|
if status != 200:
|
|
screenshot_path = await self._capture_screenshot(page)
|
|
raise UpstreamBlockedError(
|
|
f"Upstream status={status}, 页面返回非200状态码, 截图地址：{screenshot_path}"
|
|
)
|
|
await self._wait_for_clearance(page)
|
|
html_content = await page.content()
|
|
cookies = await context.cookies()
|
|
cf_clearance = self._get_cookie_value(cookies, "cf_clearance")
|
|
except Exception:
|
|
await self._browser_pool.close_context(context)
|
|
raise
|
|
|
|
now = asyncio.get_running_loop().time()
|
|
warm = WarmContext(
|
|
session_id=str(uuid4()),
|
|
context=context,
|
|
cf_clearance=cf_clearance,
|
|
created_at=now,
|
|
last_used_at=now,
|
|
)
|
|
self._warm_contexts[session_name] = warm
|
|
logger.info("保活上下文创建完成：session_name=%s cf_clearance=%s", session_name, bool(cf_clearance))
|
|
return warm, html_content
|
|
|
|
async def _retire_warm(self, session_name: str, warm: WarmContext) -> None:
|
|
"""将保活上下文摘除注册表并退役（无在用页面时立即关闭，否则等空闲后关闭）。"""
|
|
if self._warm_contexts.get(session_name) is warm:
|
|
del self._warm_contexts[session_name]
|
|
warm.retired = True
|
|
if warm.in_use <= 0:
|
|
await self._close_warm_context(warm)
|
|
|
|
async def _close_warm_context(self, warm: WarmContext) -> None:
|
|
"""关闭保活上下文（保证只关闭一次）。"""
|
|
if warm.close_done:
|
|
return
|
|
warm.close_done = True
|
|
await self._browser_pool.close_context(warm.context)
|
|
|
|
@staticmethod
|
|
def _warm_to_session(warm: WarmContext) -> VerifiedSession:
|
|
"""将保活上下文映射为对外的 VerifiedSession（storage_state 不再对外暴露）。"""
|
|
return VerifiedSession(
|
|
session_id=warm.session_id,
|
|
storage_state={},
|
|
cf_clearance=warm.cf_clearance,
|
|
html=None,
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
"""关闭全部保活上下文，应在浏览器池关闭前调用。"""
|
|
for warm in list(self._warm_contexts.values()):
|
|
await self._close_warm_context(warm)
|
|
self._warm_contexts.clear()
|
|
|
|
async def _capture_screenshot(self, page: object) -> str | None:
|
|
"""对当前页面截图用于排查，失败时返回 None。"""
|
|
try:
|
|
screenshot_dir = Path(__file__).resolve().parents[2] / self._settings.log_dir / "screenshots"
|
|
screenshot_dir.mkdir(parents=True, exist_ok=True)
|
|
screenshot_file = screenshot_dir / f"{uuid4().hex}.png"
|
|
await page.screenshot(path=str(screenshot_file), full_page=True)
|
|
return str(screenshot_file)
|
|
except Exception:
|
|
return None
|
|
|
|
async def invalidate(self, target_url: str) -> None:
|
|
"""主动让会话失效
|
|
|
|
当识别到被阻断/挑战失败时调用，下次请求将创建新会话。
|
|
|
|
Args:
|
|
target_url: 目标页面 URL，用于定位需要失效的会话
|
|
"""
|
|
session_name = self._session_name_for_url(target_url)
|
|
logger.warning("会话失效：session_name=%s", session_name)
|
|
await self._session_store.invalidate_session(session_name)
|
|
warm = self._warm_contexts.get(session_name)
|
|
if warm is not None:
|
|
await self._retire_warm(session_name, warm)
|
|
|
|
async def _create_verified_session(self, session_name: str, target_url: str) -> VerifiedSession:
|
|
"""通过真实浏览器访问目标页面，让 Cloudflare 挑战在浏览器侧完成
|
|
|
|
流程：
|
|
1. 从浏览器池获取页面会话
|
|
2. 访问目标页面
|
|
3. 等待 Cloudflare 挑战通过
|
|
4. 保存浏览器存储状态和 cf_clearance Cookie
|
|
5. 同时返回页面 HTML，避免调用方二次加载同一页面
|
|
"""
|
|
async with self._browser_pool.page_session(keep_open_on_success=False) as (context, page):
|
|
logger.debug("开始访问（用于通过挑战）：%s", target_url)
|
|
response = await page.goto(
|
|
target_url,
|
|
wait_until="domcontentloaded",
|
|
timeout=int(self._settings.request_timeout_seconds * 1000),
|
|
)
|
|
status = response.status
|
|
if status != 200:
|
|
try:
|
|
screenshot_dir = Path(__file__).resolve().parents[2] / self._settings.log_dir / "screenshots"
|
|
screenshot_dir.mkdir(parents=True, exist_ok=True)
|
|
screenshot_file = screenshot_dir / f"{uuid4().hex}.png"
|
|
await page.screenshot(path=str(screenshot_file), full_page=True)
|
|
screenshot_path = str(screenshot_file)
|
|
except Exception:
|
|
screenshot_path = None
|
|
|
|
raise UpstreamBlockedError(f"Upstream status={status}, 页面返回非200状态码, 截图地址：{screenshot_path}")
|
|
await self._wait_for_clearance(page)
|
|
html_content = await page.content()
|
|
storage_state = await context.storage_state()
|
|
cookies = await context.cookies()
|
|
cf_clearance = self._get_cookie_value(cookies, "cf_clearance")
|
|
logger.debug("挑战通过：session_name=%s cf_clearance=%s", session_name, bool(cf_clearance))
|
|
saved = await self._session_store.save_session(
|
|
session_name=session_name,
|
|
session_id=str(uuid4()),
|
|
user_agent=self._settings.browser_user_agent,
|
|
proxy_key=self._settings.proxy_server or "direct",
|
|
storage_state=storage_state,
|
|
cf_clearance=cf_clearance,
|
|
)
|
|
return VerifiedSession(
|
|
session_id=saved.session_id,
|
|
storage_state=saved.storage_state,
|
|
cf_clearance=saved.cf_clearance,
|
|
html=html_content,
|
|
)
|
|
|
|
async def _wait_for_clearance(self, page: object, timeout_seconds: float | None = None) -> None:
|
|
"""检测并等待 Cloudflare 挑战完成
|
|
|
|
循环检测页面标题和内容：
|
|
- 如果检测到阻断响应（如 1020/403），直接抛出异常
|
|
- 如果检测到挑战页面（如 "Just a moment"），继续等待
|
|
- 如果既非阻断也非挑战，认为已通过
|
|
|
|
Args:
|
|
page: Playwright 页面对象
|
|
timeout_seconds: 等待超时（秒）；为 None 时使用新建会话的较长超时。
|
|
复用会话时传入较短超时，避免在无法通过的挑战上长时间空等。
|
|
|
|
Raises:
|
|
UpstreamBlockedError: 检测到被 Cloudflare 阻断
|
|
CloudflareChallengeError: 等待挑战超时
|
|
"""
|
|
wait_timeout = (
|
|
self._settings.cloudflare_wait_timeout_seconds
|
|
if timeout_seconds is None
|
|
else timeout_seconds
|
|
)
|
|
timeout_at = asyncio.get_running_loop().time() + wait_timeout
|
|
while True:
|
|
title = (await page.title()).lower()
|
|
content = (await page.content()).lower()
|
|
|
|
if self._is_blocked_response(title, content):
|
|
logger.warning("检测到被阻断响应（可能是 1020/403）")
|
|
raise UpstreamBlockedError("Cloudflare returned a blocked response")
|
|
if not self._looks_like_challenge(title, content):
|
|
return
|
|
|
|
try:
|
|
cf_iframe = page.frame_locator('iframe[src*="turnstile"], iframe[src*="cloudflare"]')
|
|
if await cf_iframe.locator('input[type="checkbox"]').count() > 0:
|
|
checkbox = cf_iframe.locator('input[type="checkbox"]')
|
|
if await checkbox.is_visible():
|
|
logger.debug("检测到 Turnstile 复选框，尝试自动点击")
|
|
await checkbox.click()
|
|
await page.wait_for_timeout(2000)
|
|
except Exception as e:
|
|
logger.debug("自动点击 Turnstile 失败或未找到: %s", e)
|
|
|
|
if asyncio.get_running_loop().time() >= timeout_at:
|
|
logger.warning("等待挑战超时（秒）：%s", wait_timeout)
|
|
raise CloudflareChallengeError()
|
|
|
|
await page.wait_for_timeout(1500)
|
|
|
|
@staticmethod
|
|
def _get_cookie_value(cookies: Iterable[dict], name: str) -> str | None:
|
|
"""从 Cookie 列表中获取指定名称的 Cookie 值"""
|
|
for cookie in cookies:
|
|
if cookie.get("name") == name:
|
|
return cookie.get("value")
|
|
return None
|
|
|
|
@staticmethod
|
|
def _looks_like_challenge(title: str, content: str) -> bool:
|
|
"""判断页面是否仍处于 Cloudflare 挑战中
|
|
|
|
先排除正常页面（搜索列表/商品详情页中也可能包含 Cloudflare 脚本），
|
|
再检查标题和内容中是否包含挑战标志（如 "Just a moment"）。
|
|
"""
|
|
# 正常页面的 DOM 特征，出现这些说明挑战已通过
|
|
if (
|
|
'class="item_box' in content
|
|
or "product-name" in content
|
|
or 'id="item_title"' in content
|
|
or "h1_title_product" in content
|
|
or 'id="item_detailinfo"' in content
|
|
or "tbl_product_info" in content
|
|
or "商品詳細情報" in content
|
|
):
|
|
return False
|
|
|
|
challenge_markers = (
|
|
"just a moment",
|
|
"checking your browser",
|
|
"verify you are human",
|
|
"attention required",
|
|
"cf-challenge",
|
|
"cf-turnstile",
|
|
"cf_chl_opt",
|
|
"cf-please-wait",
|
|
"challenge-form",
|
|
)
|
|
return any(marker in title or marker in content for marker in challenge_markers)
|
|
|
|
@staticmethod
|
|
def _is_blocked_response(title: str, content: str) -> bool:
|
|
"""判断页面是否为 Cloudflare 阻断响应（如 1020 错误）"""
|
|
blocked_markers = (
|
|
"access denied",
|
|
"error code 1020",
|
|
"forbidden",
|
|
"temporarily blocked",
|
|
)
|
|
return any(marker in title or marker in content for marker in blocked_markers)
|
|
|
|
def _session_name_for_url(self, target_url: str) -> str:
|
|
"""根据 URL、代理和 User-Agent 生成会话名称
|
|
|
|
同一域名下仍共享会话，但会按当前代理与 UA 做进一步隔离，
|
|
避免不同运行身份误复用同一套 Cloudflare 验证状态。
|
|
"""
|
|
parsed = urlparse(target_url)
|
|
proxy_key = self._settings.proxy_server or "direct"
|
|
user_agent = self._settings.browser_user_agent
|
|
user_agent_hash = hashlib.sha1(user_agent.encode("utf-8")).hexdigest()[:12]
|
|
return f"{parsed.netloc}|{proxy_key}|{user_agent_hash}"
|
|
|
|
@staticmethod
|
|
def _to_verified_session(state: SessionState) -> VerifiedSession:
|
|
"""将 SessionState 转换为 VerifiedSession（复用场景，html 为空）"""
|
|
return VerifiedSession(
|
|
session_id=state.session_id,
|
|
storage_state=state.storage_state,
|
|
cf_clearance=state.cf_clearance,
|
|
html=None,
|
|
)
|