343 lines
15 KiB
Python
343 lines
15 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 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
|
|
|
|
|
|
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
|
|
|
|
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._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，并屏蔽会话复用失败的细节
|
|
|
|
统一处理三种情形，确保调用方拿到的要么是有效页面，要么是明确的上游异常：
|
|
1. 新建会话：直接复用创建时携带的页面 HTML，无需二次导航
|
|
2. 复用会话：带 storage_state 重新导航，并消化可能出现的 Cloudflare 挑战
|
|
3. 复用失败：主动失效后强制重建一次，对调用方透明（避免「一次成功一次失败」交替）
|
|
|
|
Args:
|
|
target_url: 目标页面 URL
|
|
|
|
Returns:
|
|
(html, session) 元组，session 为最终生效的会话
|
|
|
|
Raises:
|
|
UpstreamBlockedError: 会话重建后仍无法获取有效页面
|
|
CloudflareChallengeError: 等待 Cloudflare 挑战超时
|
|
"""
|
|
session = await self.get_verified_session(target_url)
|
|
if session.html is not None:
|
|
return session.html, session
|
|
|
|
html = await self._try_fetch_with_session(target_url, session)
|
|
if html is not None:
|
|
return html, session
|
|
|
|
# 复用会话已失效：失效并强制重建，新会话自带页面 HTML
|
|
logger.info("复用会话失败，强制重建会话：url=%s", target_url)
|
|
await self.invalidate(target_url)
|
|
session = await self.get_verified_session(target_url, force_refresh=True)
|
|
if session.html is not None:
|
|
return session.html, session
|
|
|
|
# 兜底：force_refresh 理论上必带 HTML，缺失时再导航一次
|
|
html = await self._try_fetch_with_session(target_url, session)
|
|
if html is None:
|
|
raise UpstreamBlockedError(f"会话重建后仍无法获取页面：url={target_url}")
|
|
return html, session
|
|
|
|
async def _try_fetch_with_session(self, target_url: str, session: VerifiedSession) -> str | None:
|
|
"""带已有会话导航并消化 Cloudflare 挑战
|
|
|
|
与新建路径保持一致：导航后等待挑战自动通过，而非一遇非 200 就判死。
|
|
CF 对携带过期 cf_clearance 的请求常直接返回挑战页（首响应可能非 200），
|
|
因此即便首响应非 200，也先给页面一次自动通过 Turnstile 的机会，
|
|
仅当挑战超时/被硬阻断、或通过后仍无有效内容时才判定会话失效。
|
|
|
|
Returns:
|
|
通过验证的页面 HTML；若判定会话已失效，返回 None
|
|
"""
|
|
async with self._browser_pool.page_session(storage_state=session.storage_state) 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
|
|
try:
|
|
# 复用使用短超时：快速验证能否通过，不行就放弃复用交由上层重建，
|
|
# 避免在无法自动通过的挑战上空等整个 cloudflare_wait_timeout_seconds。
|
|
await self._wait_for_clearance(
|
|
page, timeout_seconds=self._settings.cloudflare_reuse_wait_timeout_seconds
|
|
)
|
|
except (CloudflareChallengeError, UpstreamBlockedError):
|
|
logger.info(
|
|
"复用会话未通过挑战，判定失效：url=%s status=%s had_cf_clearance=%s",
|
|
target_url, status, bool(session.cf_clearance),
|
|
)
|
|
return None
|
|
|
|
html = await page.content()
|
|
# 挑战已通过但页面仍是挑战壳（极少数 wait 提前返回的情况）：判失效，触发重建
|
|
if self._looks_like_challenge((await page.title()).lower(), html.lower()):
|
|
logger.info(
|
|
"复用会话通过等待后仍无有效内容，判定失效：url=%s status=%s had_cf_clearance=%s",
|
|
target_url, status, bool(session.cf_clearance),
|
|
)
|
|
return None
|
|
return html
|
|
|
|
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)
|
|
|
|
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,
|
|
)
|