diff --git a/app/main.py b/app/main.py index 81ab5b8..4674439 100644 --- a/app/main.py +++ b/app/main.py @@ -80,6 +80,7 @@ async def lifespan(app: FastAPI): try: yield finally: + await container.session_manager.close() await container.browser_pool.close() await container.session_store.close() diff --git a/app/services/browser_pool.py b/app/services/browser_pool.py index 7d4d1b6..56d9a26 100644 --- a/app/services/browser_pool.py +++ b/app/services/browser_pool.py @@ -250,6 +250,107 @@ class BrowserPool: except Exception: logger.exception("关闭旧的保留调试上下文失败") + async def _build_context_obj(self, storage_state: dict[str, Any] | None) -> Any: + """创建浏览器上下文(含断链重试与 stealth),仅负责创建本身。 + + 计入回收阈值统计;不负责并发信号量与关闭,由调用方管理生命周期。 + """ + context = None + for attempt in range(2): + try: + context = await self._browser.new_context( + locale=self._settings.browser_locale, + timezone_id=self._settings.browser_timezone_id, + user_agent=self._settings.browser_user_agent, + viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900}, + storage_state=storage_state, + ) + break + except Exception as exc: + if attempt == 0 and self._should_retry_context_creation(exc): + logger.warning("创建浏览器上下文失败,尝试重建浏览器后重试一次:err=%s", exc) + await self._restart_browser(f"new_context failed: {exc}") + continue + raise + + # 上下文创建成功,计入回收阈值统计 + self._contexts_since_start += 1 + + if self._settings.browser_stealth_enabled: + try: + from playwright_stealth import Stealth + + await Stealth().apply_stealth_async(context) + except Exception as e: + logger.warning("应用 playwright-stealth 失败,隐身模式未生效, err=%s", e) + + return context + + @staticmethod + async def _configure_page(context: Any) -> Any: + """在上下文中创建页面并设置反检测请求头。""" + page = await context.new_page() + await page.set_extra_http_headers( + { + "Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7", + "Cache-Control": "no-cache", + "Pragma": "no-cache", + } + ) + return page + + async def create_persistent_context(self, storage_state: dict[str, Any] | None = None) -> Any: + """创建一个由调用方负责生命周期的长存上下文。 + + 用于 cf_clearance 保活复用:通过挑战后保留该上下文,后续请求在其内 + 开新页面访问目标页(同一上下文内 cf_clearance 有效)。调用方需在不再 + 使用时调用 close_context 释放。 + """ + await self._ensure_browser_ready() + return await self._build_context_obj(storage_state) + + @asynccontextmanager + async def page_in_context(self, context: Any) -> AsyncIterator[Any]: + """在既有(长存)上下文中开一个页面,复用并发信号量。 + + 结束时仅关闭页面,不关闭上下文(上下文生命周期由调用方维护)。 + + Raises: + ResourceBusyError: 等待可用槽位超时 + """ + await self._ensure_browser_ready() + try: + await asyncio.wait_for( + self._semaphore.acquire(), + timeout=self._settings.page_acquire_timeout_seconds, + ) + except TimeoutError as exc: + raise ResourceBusyError() from exc + + page = None + counted = False + try: + self._active_contexts += 1 + counted = True + page = await self._configure_page(context) + yield page + finally: + if page is not None: + try: + await page.close() + except Exception: + logger.debug("关闭复用上下文页面失败", exc_info=True) + if counted: + self._active_contexts -= 1 + self._semaphore.release() + + async def close_context(self, context: Any) -> None: + """关闭一个长存上下文(忽略已关闭等异常)。""" + try: + await context.close() + except Exception: + logger.debug("关闭长存上下文失败", exc_info=True) + @asynccontextmanager async def page_session( self, @@ -295,45 +396,12 @@ class BrowserPool: try: # 创建上下文前检查是否需要主动回收(仅在空闲时执行,不打断进行中的会话) await self._maybe_recycle() - for attempt in range(2): - try: - context = await self._browser.new_context( - locale=self._settings.browser_locale, - timezone_id=self._settings.browser_timezone_id, - user_agent=self._settings.browser_user_agent, - viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900}, - storage_state=storage_state, - ) - break - except Exception as exc: - if attempt == 0 and self._should_retry_context_creation(exc): - logger.warning("创建浏览器上下文失败,尝试重建浏览器后重试一次:err=%s", exc) - await self._restart_browser(f"new_context failed: {exc}") - continue - raise - - # 上下文创建成功,计入回收阈值统计并标记为在用 - self._contexts_since_start += 1 + context = await self._build_context_obj(storage_state) + # 上下文创建成功,标记为在用(回收阈值计数已在 _build_context_obj 内完成) self._active_contexts += 1 context_counted = True - if self._settings.browser_stealth_enabled: - try: - from playwright_stealth import Stealth - - await Stealth().apply_stealth_async(context) - # logger.debug("已为浏览器上下文应用 Stealth 模式") - except Exception as e: - logger.warning("应用 playwright-stealth 失败,隐身模式未生效, err=%s", e) - - page = await context.new_page() - await page.set_extra_http_headers( - { - "Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7", - "Cache-Control": "no-cache", - "Pragma": "no-cache", - } - ) + page = await self._configure_page(context) yield context, page except Exception: if self._settings.browser_keep_open_on_error_effective and context is not None: diff --git a/app/services/cloudflare_session.py b/app/services/cloudflare_session.py index 63dd8e5..41d11bf 100644 --- a/app/services/cloudflare_session.py +++ b/app/services/cloudflare_session.py @@ -14,6 +14,7 @@ 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 @@ -35,6 +36,24 @@ class VerifiedSession: html: str | None = None # 新创建会话时携带的页面 HTML,复用场景为 None +@dataclass(slots=True) +class WarmContext: + """保活的浏览器上下文:通过 Cloudflare 挑战后长期驻留,供后续请求复用 + + cf_clearance 无法跨上下文重放,因此保留通过挑战的上下文本身, + 后续请求在同一上下文内开新页面访问目标页(上下文内 clearance 有效)。 + 带引用计数以保证并发使用时安全退役。 + """ + 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 会话管理器,负责会话的获取、验证和失效""" @@ -48,6 +67,10 @@ class CloudflareSessionManager: 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 验证的会话 @@ -77,84 +100,171 @@ class CloudflareSessionManager: return session async def fetch_html(self, target_url: str) -> tuple[str, VerifiedSession]: - """获取目标页面 HTML,并屏蔽会话复用失败的细节 + """获取目标页面 HTML,基于「保活上下文」复用通过验证的浏览器会话 - 统一处理三种情形,确保调用方拿到的要么是有效页面,要么是明确的上游异常: - 1. 新建会话:直接复用创建时携带的页面 HTML,无需二次导航 - 2. 复用会话:带 storage_state 重新导航,并消化可能出现的 Cloudflare 挑战 - 3. 复用失败:主动失效后强制重建一次,对调用方透明(避免「一次成功一次失败」交替) + cf_clearance 无法跨上下文重放,因此保留通过挑战的上下文本身: + 1. 已有保活上下文且未过期:在其内开新页面访问目标页(同上下文 clearance 有效) + 2. 复用失败或无保活上下文:在构建锁内创建新的保活上下文(顺带完成本次访问) Args: target_url: 目标页面 URL Returns: - (html, session) 元组,session 为最终生效的会话 + (html, session) 元组,session 携带当前生效会话的 session_id 等信息 Raises: - UpstreamBlockedError: 会话重建后仍无法获取有效页面 + UpstreamBlockedError: 重建后仍无法获取有效页面 CloudflareChallengeError: 等待 Cloudflare 挑战超时 + ResourceBusyError: 等待页面槽位超时 """ - session = await self.get_verified_session(target_url) - if session.html is not None: - return session.html, session + session_name = self._session_name_for_url(target_url) - html = await self._try_fetch_with_session(target_url, session) - if html is not None: - return html, session + # 1. 优先复用已有保活上下文 + warm = self._warm_contexts.get(session_name) + if warm is not None and not self._warm_expired(warm): + html = await self._fetch_via_warm(warm, target_url) + if html is not None: + return html, self._warm_to_session(warm) + await self._retire_warm(session_name, warm) - # 复用会话已失效:失效并强制重建,新会话自带页面 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 + # 2. 在构建锁内创建/复用保活上下文,避免并发重建风暴 + lock = self._build_locks.setdefault(session_name, asyncio.Lock()) + async with lock: + # 双重检查:可能已被其他协程在等待锁期间建好 + warm = self._warm_contexts.get(session_name) + if warm is not None and not self._warm_expired(warm): + html = await self._fetch_via_warm(warm, target_url) + if html is not None: + return html, self._warm_to_session(warm) + await self._retire_warm(session_name, warm) - # 兜底: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 + logger.info("创建保活上下文:session_name=%s", session_name) + warm, html = await self._build_warm_context(session_name, target_url) + return html, self._warm_to_session(warm) - async def _try_fetch_with_session(self, target_url: str, session: VerifiedSession) -> str | None: - """带已有会话导航并消化 Cloudflare 挑战 + 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 - 与新建路径保持一致:导航后等待挑战自动通过,而非一遇非 200 就判死。 - CF 对携带过期 cf_clearance 的请求常直接返回挑战页(首响应可能非 200), - 因此即便首响应非 200,也先给页面一次自动通过 Turnstile 的机会, - 仅当挑战超时/被硬阻断、或通过后仍无有效内容时才判定会话失效。 + async def _fetch_via_warm(self, warm: WarmContext, target_url: str) -> str | None: + """在保活上下文内开新页面访问目标页。 Returns: - 通过验证的页面 HTML;若判定会话已失效,返回 None + 通过验证的页面 HTML;若判定上下文已失效(非 200 / 挑战未过 / 上下文异常),返回 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 + 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), ) - except (CloudflareChallengeError, UpstreamBlockedError): - logger.info( - "复用会话未通过挑战,判定失效:url=%s status=%s had_cf_clearance=%s", - target_url, status, bool(session.cf_clearance), - ) - return None + 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 - 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), + # 首响应 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。""" + 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), ) - return None - return html + 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: """主动让会话失效 @@ -167,6 +277,9 @@ class CloudflareSessionManager: 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 挑战在浏览器侧完成 diff --git a/tests/test_warm_context.py b/tests/test_warm_context.py new file mode 100644 index 0000000..4fc487e --- /dev/null +++ b/tests/test_warm_context.py @@ -0,0 +1,124 @@ +"""保活上下文(cf_clearance 复用)逻辑的单元测试(不启动真实 Chromium)。""" +from __future__ import annotations + +from contextlib import asynccontextmanager +from types import SimpleNamespace + +import pytest + +from app.core.config import Settings +from app.services.cloudflare_session import CloudflareSessionManager +from app.services.session_store import SessionStore + +_DETAIL_URL = "https://www.suruga-ya.jp/product/detail/{}" + + +class _FakePage: + """最小页面替身:goto 状态由所属上下文是否被标记失败决定。""" + + def __init__(self, pool: "_FakeBrowserPool", context: "_FakeContext"): + self._pool = pool + self._context = context + + async def goto(self, url: str, **_kw): + status = 403 if self._context.idx in self._pool.fail_idxs else 200 + return SimpleNamespace(status=status) + + async def title(self) -> str: + return "ok" + + async def content(self) -> str: + return self._pool.content_html + + async def screenshot(self, **_kw): # pragma: no cover - 非 200 分支才用到 + return None + + async def wait_for_timeout(self, _ms): # pragma: no cover + return None + + +class _FakeContext: + def __init__(self, idx: int): + self.idx = idx + self.closed = False + + async def cookies(self): + return [{"name": "cf_clearance", "value": f"cv-{self.idx}"}] + + async def close(self): + self.closed = True + + +class _FakeBrowserPool: + """记录上下文创建次数,并允许把指定上下文标记为「复用必失败」。""" + + def __init__(self): + self.content_html = "ok" + self.create_calls = 0 + self.contexts: list[_FakeContext] = [] + self.fail_idxs: set[int] = set() + + async def create_persistent_context(self, storage_state=None) -> _FakeContext: + self.create_calls += 1 + ctx = _FakeContext(self.create_calls) + self.contexts.append(ctx) + return ctx + + @asynccontextmanager + async def page_in_context(self, context: _FakeContext): + yield _FakePage(self, context) + + async def close_context(self, context: _FakeContext): + await context.close() + + +def _make_manager(pool: _FakeBrowserPool) -> CloudflareSessionManager: + settings = Settings(session_ttl_seconds=900, cloudflare_reuse_wait_timeout_seconds=8.0) + return CloudflareSessionManager(settings, pool, SessionStore(settings)) + + +@pytest.mark.asyncio +async def test_warm_context_is_reused_without_rebuild(): + pool = _FakeBrowserPool() + mgr = _make_manager(pool) + + html1, sess1 = await mgr.fetch_html(_DETAIL_URL.format(1)) + assert html1 == pool.content_html + assert pool.create_calls == 1 # 首次构建 + + # 第二次请求(不同商品)应复用同一保活上下文,不再新建 + html2, sess2 = await mgr.fetch_html(_DETAIL_URL.format(2)) + assert html2 == pool.content_html + assert pool.create_calls == 1 + assert sess2.session_id == sess1.session_id + + +@pytest.mark.asyncio +async def test_warm_context_rebuilds_and_closes_on_failure(): + pool = _FakeBrowserPool() + mgr = _make_manager(pool) + + _, sess1 = await mgr.fetch_html(_DETAIL_URL.format(1)) + assert pool.create_calls == 1 + + # 让首个上下文在后续复用导航时返回 403(模拟 clearance 失效) + pool.fail_idxs.add(1) + + html2, sess2 = await mgr.fetch_html(_DETAIL_URL.format(2)) + assert html2 == pool.content_html + assert pool.create_calls == 2 # 复用失败后重建 + assert pool.contexts[0].closed is True # 旧上下文已退役关闭 + assert sess2.session_id != sess1.session_id # 换了新会话 + + +@pytest.mark.asyncio +async def test_close_releases_all_warm_contexts(): + pool = _FakeBrowserPool() + mgr = _make_manager(pool) + + await mgr.fetch_html(_DETAIL_URL.format(1)) + assert pool.contexts[0].closed is False + + await mgr.close() + assert pool.contexts[0].closed is True + assert mgr._warm_contexts == {}