cf逻辑复用

app/services/cloudflare_session.py

@@ -76,6 +76,71 @@ class CloudflareSessionManager:
         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 就判死。
+
+        Returns:
+            通过验证的页面 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
+            if status != 200:
+                # 携带已有会话仍被判非 200，通常意味着 cf_clearance 已失效，交由上层重建
+                logger.debug("复用会话导航返回非 200：url=%s status=%s", target_url, status)
+                return None
+            try:
+                await self._wait_for_clearance(page)
+            except (CloudflareChallengeError, UpstreamBlockedError):
+                logger.debug("复用会话仍处于挑战/阻断状态，判定会话失效：url=%s", target_url)
+                return None
+            return await page.content()
+
     async def invalidate(self, target_url: str) -> None:
         """主动让会话失效
 

app/services/surugaya_client.py

@@ -91,21 +91,7 @@ class SurugayaClient:
         logger.debug("开始抓取搜索页：payload=%s", json.dumps(payload.model_dump(), ensure_ascii=False))
         url = str(payload.search_url) if payload.search_url else self._build_search_url(payload)
 
-        session = await self._session_manager.get_verified_session(url)
+        html, session = await self._session_manager.fetch_html(url)
-        if session.html is not None:
-            # 新创建的会话已携带页面 HTML，直接使用，避免二次加载
-            html = session.html
-        else:
-            async with self._browser_pool.page_session(storage_state=session.storage_state) as (_, page):
-                response = await page.goto(
-                    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:
-                    await self._raise_upstream_error(page=page, response=response, url=url)
-                html = await page.content()
 
         tree = HTMLParser(html)
 
@@ -165,21 +151,8 @@ class SurugayaClient:
         logger.info("开始抓取详情页：goods_id=%s", goods_id)
         logger.debug("详情页URL：url=%s", url)
 
-        # 抓取页面 HTML
+        # 抓取页面 HTML（会话复用失败时由 fetch_html 内部自动重建，避免成功/失败交替）
-        session = await self._session_manager.get_verified_session(url)
+        html, _ = await self._session_manager.fetch_html(url)
-        if session.html is not None:
-            html = session.html
-        else:
-            async with self._browser_pool.page_session(storage_state=session.storage_state) as (_, page):
-                response = await page.goto(
-                    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:
-                    await self._raise_upstream_error(page=page, response=response, url=url)
-                html = await page.content()
 
         tree = HTMLParser(html)