feat: 收敛 SerpAPI 正文抓取范围，降低搜索链路尾延迟 (#882) (#898)

* fix(issue-882): [feature]-收敛-serpapi-正文抓取范围，降低搜索链路尾延迟

docs/CHANGELOG.md

@@ -37,6 +37,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ### 改进
 
+- 🖥️ **核心页面统一到同一套工作台视觉语言** — `Home / Chat / Backtest / Portfolio / Settings` 进一步收口到共享设计 token、`input-surface` 输入体系、空态/错误态表达和抽屉遮罩语义，减少页面之间的视觉割裂与局部私有样式漂移。
+- 💬 **问股交互可达性与反馈增强** — 问股页补强了会话导出、通知发送、消息复制、历史删除与追问上下文提示；AI 回复操作不再过度依赖 hover，触屏设备和小屏场景下也能直接触达关键按钮。
+- 📊 **回测与持仓页表面和状态表达继续标准化** — 回测页筛选控件、布尔状态、结果表格与汇总卡片统一到共享输入/状态原语；持仓页的导入反馈、汇率刷新提示、空态与警示信息进一步归口到共享组件，减少页面级重复实现。
+- 🧭 **导航与页面壳层协同优化** — 侧边栏主题切换、问股完成角标、移动端抽屉遮罩和主内容滚动契约进一步统一，首页、问股和回测在桌面端与移动端的切页体验更稳定。
+- 🔎 **SerpAPI 正文补抓范围收敛** — 自然搜索结果不再逐条同步抓取网页正文；现在仅对极少数高位且摘要明显不足的结果，在更短超时预算内做延迟补抓，并优先复用 SerpAPI 已返回的结构化摘要，降低搜索链路尾延迟与慢站点放大风险。
+
+### 测试
+
+- 🧪 **UI 治理与关键路径回归补强** — 补充 `SidebarNav`、`ChatPage`、`BacktestPage` 等组件测试，并新增 UI governance 守卫，持续防止交互元素重新引入原生 `title` 属性或旧 `input-terminal` 样式回流。同步更新 smoke / markdown drawer 相关验证，覆盖主题升级后的关键主链路。
+
+- [修复] 🐳 **Docker WebUI 运行时优先复用预构建静态资源** — `prepare_webui_frontend_assets()` 现在会先检查镜像内已有的 `static/index.html` 是否可直接复用；当容器运行时不包含 `apps/dsa-web` 源码目录且未安装 `npm` 时，也不会误报“未找到前端项目，无法自动构建”，从而恢复 Docker 部署后的 WebUI 打开能力。
+- [改进] 🔎 **SerpAPI 正文补抓范围收敛** — 自然搜索结果不再逐条同步抓取网页正文；现在仅对极少数高位且摘要明显不足的结果，在更短超时预算内做延迟补抓，并优先复用 SerpAPI 已返回的结构化摘要，降低搜索链路尾延迟与慢站点放大风险。
+
+## [3.11.0] - 2026-03-27
+
+### 发布亮点
+
+- 🎨 **Web 工作台完成一轮 UI 统一与双主题升级** — 首页、问股、回测、持仓和设置页进一步收口到统一设计 token、输入表面和状态表达；新增完整浅色主题，并支持浅色 / 深色一键切换与持久化保存。
+- 🤖 **Bot / Agent 能力重新补回主分支** — 恢复 `/history`、`/strategies`、`/research` 等命令，`/ask` 继续支持多股对比与组合视角；Deep Research、事件监控与 schedule 轮询链路重新接回主线能力。
+- 🔒 **安全性与运行稳态同步补强** — 修复 `X-Forwarded-For` 限流绕过风险，恢复 LiteLLM 官方 PyPI 安装路径，Tushare 初始化不再依赖本地 SDK，降低 Docker、桌面打包和环境重建时的脆弱点。
+- 🖥️ **日常使用细节继续打磨** — 修复首页港股自动补全提交、登录页首屏主题闪烁、历史长股票名重叠，以及 Telegram Markdown 解析失败时整条通知发送中断等问题。
+
+### 新功能
+
+- 🎨 **全新浅色主题与双主题切换上线** — Web 工作台新增完整浅色主题，并支持在侧边栏中一键切换浅色 / 深色模式；主题选择会持久化保存，刷新页面后仍保持当前偏好。此次升级不是局部配色微调，而是对卡片层级、边界对比、输入表面、状态提示和页面背景做了一整套 light theme 重绘。
+- 🤖 **补回主分支缺失的 Agent / Bot 能力** — `#648` / `#649` 已重新补回 `main`：Bot 恢复 `/history`、`/strategies`、`/research`，`/ask` 保留多股对比与组合视角；Deep Research 与 Event Monitor 的配置重新在 Web 设置页可见并可编辑，schedule 模式也重新接入事件告警轮询。
+
+### 改进
+
 - 🖥️ **核心页面统一到同一套工作台视觉语言** — `Home / Chat / Backtest / Portfolio / Settings` 进一步收口到共享设计 token、`input-surface` 输入体系、空态/错误态表达和抽屉遮罩语义，减少页面之间的视觉割裂与局部私有样式漂移。
 - 💬 **问股交互可达性与反馈增强** — 问股页补强了会话导出、通知发送、消息复制、历史删除与追问上下文提示；AI 回复操作不再过度依赖 hover，触屏设备和小屏场景下也能直接触达关键按钮。
 - 📊 **回测与持仓页表面和状态表达继续标准化** — 回测页筛选控件、布尔状态、结果表格与汇总卡片统一到共享输入/状态原语；持仓页的导入反馈、汇率刷新提示、空态与警示信息进一步归口到共享组件，减少页面级重复实现。

src/search_service.py

@@ -21,6 +21,7 @@ from datetime import date, datetime, timedelta, timezone
 from email.utils import parsedate_to_datetime
 from typing import List, Dict, Any, Optional, Tuple
 from itertools import cycle
+from urllib.parse import parse_qsl, unquote, urlparse
 import requests
 from newspaper import Article, Config
 from tenacity import (
@@ -419,6 +420,46 @@ class SerpAPISearchProvider(BaseSearchProvider):
     
     文档：https://serpapi.com/baidu-search-api?utm_source=github_daily_stock_analysis
     """
+
+    _ORGANIC_CONTENT_FETCH_LIMIT = 1
+    _ORGANIC_CONTENT_FETCH_RANK_LIMIT = 2
+    _ORGANIC_CONTENT_FETCH_TIMEOUT = 2
+    _ORGANIC_SNIPPET_SUFFICIENT_LENGTH = 140
+    _ORGANIC_FETCHED_PREVIEW_LENGTH = 320
+    _SKIPPED_CONTENT_FETCH_SUFFIXES = (
+        ".pdf",
+        ".jpg",
+        ".jpeg",
+        ".png",
+        ".gif",
+        ".svg",
+        ".webp",
+        ".zip",
+        ".rar",
+        ".7z",
+        ".doc",
+        ".docx",
+        ".ppt",
+        ".pptx",
+        ".xls",
+        ".xlsx",
+        ".csv",
+    )
+    _SKIPPED_CONTENT_FETCH_QUERY_KEYS = {
+        "attachment",
+        "attachment_file",
+        "doc",
+        "document",
+        "download",
+        "download_file",
+        "file",
+        "file_name",
+        "filename",
+        "file_path",
+        "filepath",
+        "resource",
+        "resource_file",
+    }
     
     def __init__(self, api_keys: List[str]):
         super().__init__(api_keys, "SerpAPI")
@@ -553,28 +594,33 @@ class SerpAPISearchProvider(BaseSearchProvider):
 
             # 4. 解析 Organic Results (自然搜索结果)
             organic_results = response.get('organic_results', [])
+            organic_content_fetch_attempts = 0
 
-            for item in organic_results[:max_results]:
+            for rank, item in enumerate(organic_results[:max_results]):
                 link = item.get('link', '')
-                snippet = item.get('snippet', '')
+                rich_extensions = self._extract_rich_snippet_extensions(item)
+                snippet = self._build_organic_snippet(item, rich_extensions=rich_extensions)
 
-                # 增强：如果需要，解析网页正文
-                # 策略：如果摘要太短，或者为了获取更多信息，可以请求网页
-                # 这里我们对所有结果尝试获取正文，但为了性能，仅获取前1000字符
-                content = ""
-                if link:
-                   try:
-                       fetched_content = fetch_url_content(link, timeout=5)
-                       if fetched_content:
-                           # 如果获取到了正文，将其拼接到 snippet 中，或者替换 snippet
-                           # 这里选择拼接，保留原摘要
-                           content = fetched_content
-                           if len(content) > 500:
-                               snippet = f"{snippet}\n\n【网页详情】\n{content[:500]}..."
-                           else:
-                               snippet = f"{snippet}\n\n【网页详情】\n{content}"
-                   except Exception as e:
-                       logger.debug(f"[SerpAPI] Fetch content failed: {e}")
+                if self._should_fetch_organic_content(
+                    link=link,
+                    snippet=snippet,
+                    rank=rank,
+                    fetched_count=organic_content_fetch_attempts,
+                    has_structured_summary=bool(rich_extensions),
+                ):
+                    organic_content_fetch_attempts += 1
+                    try:
+                        fetched_content = fetch_url_content(
+                            link,
+                            timeout=self._ORGANIC_CONTENT_FETCH_TIMEOUT,
+                        )
+                        if fetched_content:
+                            snippet = self._merge_organic_snippet_with_content(
+                                snippet,
+                                fetched_content,
+                            )
+                    except Exception as e:
+                        logger.debug(f"[SerpAPI] Fetch content failed: {e}")
 
                 results.append(SearchResult(
                     title=item.get('title', ''),
@@ -605,12 +651,203 @@ class SerpAPISearchProvider(BaseSearchProvider):
     def _extract_domain(url: str) -> str:
         """从 URL 提取域名"""
         try:
-            from urllib.parse import urlparse
             parsed = urlparse(url)
             return parsed.netloc.replace('www.', '') or '未知来源'
         except Exception:
             return '未知来源'
 
+    @classmethod
+    def _normalize_organic_text(cls, value: Any) -> str:
+        """标准化 SerpAPI organic 文本字段。"""
+        text = "" if value is None else str(value)
+        return re.sub(r"\s+", " ", text).strip()
+
+    @classmethod
+    def _extract_rich_snippet_extensions(cls, item: Dict[str, Any]) -> List[str]:
+        """提取 rich_snippet 中已有的结构化摘要，优先复用 API 原始返回。"""
+        rich_snippet = item.get("rich_snippet")
+        if not isinstance(rich_snippet, dict):
+            return []
+
+        extensions: List[str] = []
+        seen: set[str] = set()
+
+        for section in ("top", "bottom"):
+            section_data = rich_snippet.get(section)
+            if not isinstance(section_data, dict):
+                continue
+
+            raw_extensions = section_data.get("extensions")
+            if isinstance(raw_extensions, (list, tuple, set)):
+                for raw_value in raw_extensions:
+                    value = cls._normalize_organic_text(raw_value)
+                    if not value or value in seen:
+                        continue
+                    seen.add(value)
+                    extensions.append(value)
+
+            for raw_value in cls._flatten_rich_snippet_values(
+                section_data.get("detected_extensions")
+            ):
+                if raw_value in seen:
+                    continue
+                seen.add(raw_value)
+                extensions.append(raw_value)
+
+        return extensions
+
+    @classmethod
+    def _flatten_rich_snippet_values(
+        cls,
+        value: Any,
+        *,
+        label: Optional[str] = None,
+        allow_unlabeled_scalar: bool = False,
+    ) -> List[str]:
+        """把 rich_snippet.detected_extensions 展平为可读文本。"""
+        if isinstance(value, dict):
+            flattened: List[str] = []
+            for key, nested_value in value.items():
+                flattened.extend(
+                    cls._flatten_rich_snippet_values(
+                        nested_value,
+                        label=cls._normalize_organic_text(str(key)).replace("_", " "),
+                    )
+                )
+            return flattened
+
+        if isinstance(value, (list, tuple, set)):
+            flattened: List[str] = []
+            for nested_value in value:
+                flattened.extend(
+                    cls._flatten_rich_snippet_values(
+                        nested_value,
+                        label=label,
+                        allow_unlabeled_scalar=True,
+                    )
+                )
+            return flattened
+
+        text = cls._normalize_organic_text(value)
+        if not text:
+            return []
+
+        if label:
+            return [f"{label}: {text}"]
+
+        if allow_unlabeled_scalar:
+            return [text]
+
+        return []
+
+    @classmethod
+    def _build_organic_snippet(
+        cls,
+        item: Dict[str, Any],
+        *,
+        rich_extensions: Optional[List[str]] = None,
+    ) -> str:
+        """构建 organic result 摘要，尽量先消费 SerpAPI 已返回的信息。"""
+        snippet = cls._normalize_organic_text(item.get("snippet", ""))
+        if rich_extensions is None:
+            rich_extensions = cls._extract_rich_snippet_extensions(item)
+
+        if rich_extensions:
+            rich_text = " | ".join(rich_extensions)
+            if rich_text and rich_text not in snippet:
+                snippet = f"{snippet}\n{rich_text}".strip() if snippet else rich_text
+
+        return snippet
+
+    @classmethod
+    def _matches_skipped_content_fetch_suffix(cls, value: Any) -> bool:
+        """判断链接片段是否指向附件或其他非 HTML 资源。"""
+        normalized_value = cls._normalize_organic_text(value).lower()
+        if not normalized_value:
+            return False
+
+        decoded_value = unquote(normalized_value)
+        if decoded_value.endswith(cls._SKIPPED_CONTENT_FETCH_SUFFIXES):
+            return True
+
+        return urlparse(decoded_value).path.lower().endswith(
+            cls._SKIPPED_CONTENT_FETCH_SUFFIXES
+        )
+
+    @classmethod
+    def _matches_skipped_content_fetch_query_param(
+        cls, key: Any, value: Any
+    ) -> bool:
+        """仅对少数显式附件参数跳过正文抓取，避免误伤普通 HTML 页面。"""
+        normalized_key = cls._normalize_organic_text(key)
+        if not normalized_key:
+            return False
+
+        snake_key = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", normalized_key)
+        canonical_key = re.sub(r"[^a-z0-9]+", "_", snake_key.lower()).strip("_")
+        if canonical_key not in cls._SKIPPED_CONTENT_FETCH_QUERY_KEYS:
+            return False
+
+        return cls._matches_skipped_content_fetch_suffix(value)
+
+    @classmethod
+    def _should_fetch_organic_content(
+        cls,
+        *,
+        link: Any,
+        snippet: str,
+        rank: int,
+        fetched_count: int,
+        has_structured_summary: bool,
+    ) -> bool:
+        """仅对极少量高位且摘要明显不足的结果补抓正文。"""
+        if fetched_count >= cls._ORGANIC_CONTENT_FETCH_LIMIT:
+            return False
+
+        if rank >= cls._ORGANIC_CONTENT_FETCH_RANK_LIMIT:
+            return False
+
+        if has_structured_summary:
+            return False
+
+        if len(snippet) >= cls._ORGANIC_SNIPPET_SUFFICIENT_LENGTH:
+            return False
+
+        if not isinstance(link, str):
+            return False
+
+        if not link or not link.startswith(("http://", "https://")):
+            return False
+
+        parsed_link = urlparse(link)
+        if parsed_link.scheme not in {"http", "https"}:
+            return False
+
+        if cls._matches_skipped_content_fetch_suffix(parsed_link.path):
+            return False
+
+        for key, value in parse_qsl(parsed_link.query, keep_blank_values=True):
+            if cls._matches_skipped_content_fetch_query_param(key, value):
+                return False
+
+        return True
+
+    @classmethod
+    def _merge_organic_snippet_with_content(cls, snippet: str, content: str) -> str:
+        """用较短正文预览补强 snippet，避免拉长单次搜索耗时和返回体积。"""
+        normalized = cls._normalize_organic_text(content)
+        if not normalized:
+            return snippet
+
+        preview = normalized[:cls._ORGANIC_FETCHED_PREVIEW_LENGTH]
+        if len(normalized) > cls._ORGANIC_FETCHED_PREVIEW_LENGTH:
+            preview = f"{preview}..."
+
+        if snippet:
+            return f"{snippet}\n\n【网页详情】\n{preview}"
+
+        return f"【网页详情】\n{preview}"
+
 
 class BochaSearchProvider(BaseSearchProvider):
     """

tests/test_search_serpapi_provider.py

@@ -0,0 +1,539 @@
+# -*- coding: utf-8 -*-
+"""
+Regression tests for SerpAPI organic content fetch throttling (Issue #882).
+"""
+
+import sys
+import unittest
+from types import ModuleType
+from unittest.mock import MagicMock, patch
+
+# Mock newspaper before search_service import (optional dependency)
+if "newspaper" not in sys.modules:
+    mock_np = MagicMock()
+    mock_np.Article = MagicMock()
+    mock_np.Config = MagicMock()
+    sys.modules["newspaper"] = mock_np
+
+from src.search_service import SerpAPISearchProvider
+
+
+class _FakeGoogleSearch:
+    response_payload = {}
+    init_params = []
+
+    def __init__(self, params):
+        type(self).init_params.append(params)
+
+    def get_dict(self):
+        return type(self).response_payload
+
+    @classmethod
+    def reset(cls) -> None:
+        cls.response_payload = {}
+        cls.init_params = []
+
+
+def _fake_serpapi_module() -> ModuleType:
+    module = ModuleType("serpapi")
+    module.GoogleSearch = _FakeGoogleSearch
+    return module
+
+
+class TestSerpAPISearchProvider(unittest.TestCase):
+    """Tests for provider-specific organic content fetch behavior."""
+
+    def _patch_serpapi(self, payload):
+        _FakeGoogleSearch.reset()
+        _FakeGoogleSearch.response_payload = payload
+        return patch.dict(sys.modules, {"serpapi": _fake_serpapi_module()})
+
+    def test_provider_skips_body_fetch_when_snippet_is_sufficient(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+        long_snippet = "这是一段已经足够长的摘要。 " * 12
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Long summary result",
+                        "link": "https://example.com/long-summary",
+                        "snippet": long_snippet,
+                        "source": "Example",
+                        "date": "2026-03-20",
+                    }
+                ]
+            }
+        ), patch("src.search_service.fetch_url_content") as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        self.assertEqual(resp.results[0].snippet, long_snippet.strip())
+        self.assertEqual(resp.results[0].published_date, "2026-03-20")
+        mock_fetch.assert_not_called()
+        self.assertEqual(_FakeGoogleSearch.init_params[0]["num"], 3)
+
+    def test_provider_uses_rich_snippet_extensions_without_fetching(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Structured summary result",
+                        "link": "https://example.com/structured-summary",
+                        "source": "Example",
+                        "rich_snippet": {
+                            "top": {
+                                "extensions": [
+                                    "Q4 revenue grows 22% year over year and margin keeps improving",
+                                    "Management raises full-year guidance after demand stays strong",
+                                ]
+                            },
+                            "bottom": {
+                                "extensions": [
+                                    "Brokerages lift target prices and keep overweight ratings",
+                                ]
+                            },
+                        },
+                    }
+                ]
+            }
+        ), patch("src.search_service.fetch_url_content") as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        self.assertIn("Q4 revenue grows 22%", resp.results[0].snippet)
+        self.assertIn("Brokerages lift target prices", resp.results[0].snippet)
+        mock_fetch.assert_not_called()
+
+    def test_provider_uses_detected_rich_snippet_fields_without_fetching(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Detected extensions result",
+                        "link": "https://example.com/detected-extensions",
+                        "source": "Example",
+                        "rich_snippet": {
+                            "top": {
+                                "detected_extensions": {
+                                    "price": "$125.30",
+                                    "updated_at": "1 hour ago",
+                                }
+                            },
+                            "bottom": {
+                                "detected_extensions": {
+                                    "rating": 4.5,
+                                    "votes": 1200,
+                                }
+                            },
+                        },
+                    }
+                ]
+            }
+        ), patch("src.search_service.fetch_url_content") as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        self.assertIn("price: $125.30", resp.results[0].snippet)
+        self.assertIn("updated at: 1 hour ago", resp.results[0].snippet)
+        self.assertIn("rating: 4.5", resp.results[0].snippet)
+        self.assertIn("votes: 1200", resp.results[0].snippet)
+        mock_fetch.assert_not_called()
+
+    def test_provider_preserves_falsy_detected_extensions_without_fetching(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Zero-like extensions result",
+                        "link": "https://example.com/zero-like-extensions",
+                        "source": "Example",
+                        "rich_snippet": {
+                            "top": {
+                                "detected_extensions": {
+                                    "price": 0,
+                                    "market_open": False,
+                                }
+                            },
+                            "bottom": {
+                                "detected_extensions": {
+                                    "votes": 0,
+                                }
+                            },
+                        },
+                    }
+                ]
+            }
+        ), patch("src.search_service.fetch_url_content") as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        self.assertIn("price: 0", resp.results[0].snippet)
+        self.assertIn("market open: False", resp.results[0].snippet)
+        self.assertIn("votes: 0", resp.results[0].snippet)
+        mock_fetch.assert_not_called()
+
+    def test_provider_ignores_scalar_detected_extensions_and_still_fetches(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Malformed detected extensions result",
+                        "link": "https://example.com/malformed-detected-extensions",
+                        "snippet": "摘要过短",
+                        "source": "Example",
+                        "rich_snippet": {
+                            "top": {
+                                "detected_extensions": True,
+                            },
+                        },
+                    }
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            return_value="网页正文补充信息 " * 40,
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/malformed-detected-extensions",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertNotIn("True", resp.results[0].snippet)
+        self.assertIn("【网页详情】", resp.results[0].snippet)
+
+    def test_provider_ignores_malformed_rich_snippet_sections(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+        long_enough_snippet = "已有摘要，足够避免补抓。 " * 16
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Malformed rich snippet",
+                        "link": "https://example.com/malformed-rich-snippet",
+                        "snippet": long_enough_snippet,
+                        "source": "Example",
+                        "rich_snippet": {
+                            "top": "unexpected string payload",
+                            "bottom": ["unexpected", "list"],
+                        },
+                    }
+                ]
+            }
+        ), patch("src.search_service.fetch_url_content") as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        self.assertEqual(resp.results[0].snippet, long_enough_snippet.strip())
+        mock_fetch.assert_not_called()
+
+    def test_provider_ignores_non_list_rich_snippet_extensions(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Malformed extensions payload",
+                        "link": "https://example.com/malformed-extensions",
+                        "source": "Example",
+                        "rich_snippet": {
+                            "top": {
+                                "extensions": True,
+                            },
+                            "bottom": {
+                                "extensions": 1,
+                                "detected_extensions": {
+                                    "rating": 4.5,
+                                },
+                            },
+                        },
+                    }
+                ]
+            }
+        ), patch("src.search_service.fetch_url_content") as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 1)
+        self.assertIn("rating: 4.5", resp.results[0].snippet)
+        mock_fetch.assert_not_called()
+
+    def test_extract_rich_snippet_extensions_handles_non_dict_payloads(self) -> None:
+        self.assertEqual(
+            SerpAPISearchProvider._extract_rich_snippet_extensions(
+                {"rich_snippet": "unexpected string payload"}
+            ),
+            [],
+        )
+        self.assertEqual(
+            SerpAPISearchProvider._extract_rich_snippet_extensions(
+                {
+                    "rich_snippet": {
+                        "top": "unexpected string payload",
+                        "bottom": ["unexpected", "list"],
+                    }
+                }
+            ),
+            [],
+        )
+
+    def test_merge_organic_snippet_uses_normalized_length_for_ellipsis(self) -> None:
+        merged = SerpAPISearchProvider._merge_organic_snippet_with_content(
+            "原始摘要",
+            "A" + ("\n" * (SerpAPISearchProvider._ORGANIC_FETCHED_PREVIEW_LENGTH + 20)),
+        )
+
+        self.assertFalse(merged.endswith("..."))
+
+    def test_provider_fetches_only_one_top_short_snippet_candidate(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Need extra context",
+                        "link": "https://example.com/need-extra-context",
+                        "snippet": "摘要过短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "Second short result",
+                        "link": "https://example.com/second-short",
+                        "snippet": "也很短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "Third short result",
+                        "link": "https://example.com/third-short",
+                        "snippet": "还是很短",
+                        "source": "Example",
+                    },
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            return_value="网页正文补充信息 " * 40,
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 3)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/need-extra-context",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertIn("【网页详情】", resp.results[0].snippet)
+        self.assertEqual(resp.results[1].snippet, "也很短")
+        self.assertEqual(resp.results[2].snippet, "还是很短")
+
+    def test_provider_skips_asset_link_and_fetches_next_eligible_result(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "PDF attachment",
+                        "link": "https://example.com/report.PDF?download=1",
+                        "snippet": "附件摘要很短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "HTML article",
+                        "link": "https://example.com/article",
+                        "snippet": "正文摘要也短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "Third short result",
+                        "link": "https://example.com/third-short",
+                        "snippet": "还是很短",
+                        "source": "Example",
+                    },
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            return_value="网页正文补充信息 " * 40,
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 3)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/article",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertEqual(resp.results[0].snippet, "附件摘要很短")
+        self.assertIn("【网页详情】", resp.results[1].snippet)
+        self.assertEqual(resp.results[2].snippet, "还是很短")
+
+    def test_provider_skips_query_encoded_attachment_and_fetches_next_result(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Attachment behind download endpoint",
+                        "link": "https://example.com/download?file=report.pdf",
+                        "snippet": "附件摘要很短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "HTML article",
+                        "link": "https://example.com/article",
+                        "snippet": "正文摘要也短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "Third short result",
+                        "link": "https://example.com/third-short",
+                        "snippet": "还是很短",
+                        "source": "Example",
+                    },
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            return_value="网页正文补充信息 " * 40,
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=3)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 3)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/article",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertEqual(resp.results[0].snippet, "附件摘要很短")
+        self.assertIn("【网页详情】", resp.results[1].snippet)
+        self.assertEqual(resp.results[2].snippet, "还是很短")
+
+    def test_provider_keeps_html_fetch_for_asset_like_query_param(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "HTML article with media param",
+                        "link": "https://example.com/article?thumbnail=cover.jpg",
+                        "snippet": "摘要过短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "Second short result",
+                        "link": "https://example.com/second-short",
+                        "snippet": "也很短",
+                        "source": "Example",
+                    },
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            return_value="网页正文补充信息 " * 40,
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=2)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 2)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/article?thumbnail=cover.jpg",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertIn("【网页详情】", resp.results[0].snippet)
+        self.assertEqual(resp.results[1].snippet, "也很短")
+
+    def test_provider_skips_non_string_link_and_keeps_fetch_budget(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Malformed link result",
+                        "link": {"href": "https://example.com/broken"},
+                        "snippet": "摘要过短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "HTML article",
+                        "link": "https://example.com/article",
+                        "snippet": "正文摘要也短",
+                        "source": "Example",
+                    },
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            return_value="网页正文补充信息 " * 40,
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=2)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 2)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/article",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertEqual(resp.results[0].snippet, "摘要过短")
+        self.assertIn("【网页详情】", resp.results[1].snippet)
+
+    def test_provider_fetch_failure_stays_fail_open_and_stops_after_budget(self) -> None:
+        provider = SerpAPISearchProvider(["dummy_key"])
+
+        with self._patch_serpapi(
+            {
+                "organic_results": [
+                    {
+                        "title": "Slow result",
+                        "link": "https://example.com/slow",
+                        "snippet": "摘要过短",
+                        "source": "Example",
+                    },
+                    {
+                        "title": "Another short result",
+                        "link": "https://example.com/another-short",
+                        "snippet": "仍然很短",
+                        "source": "Example",
+                    },
+                ]
+            }
+        ), patch(
+            "src.search_service.fetch_url_content",
+            side_effect=TimeoutError("slow site"),
+        ) as mock_fetch:
+            resp = provider.search("阿里巴巴 财报", max_results=2)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(len(resp.results), 2)
+        mock_fetch.assert_called_once_with(
+            "https://example.com/slow",
+            timeout=SerpAPISearchProvider._ORGANIC_CONTENT_FETCH_TIMEOUT,
+        )
+        self.assertEqual(resp.results[0].snippet, "摘要过短")
+        self.assertEqual(resp.results[1].snippet, "仍然很短")
+
+
+if __name__ == "__main__":
+    unittest.main()