fix: wrap optional pipeline services in try/except for graceful degradation (#1004)

* fix: wrap optional pipeline services in try/except for graceful degradation

SearchService and SocialSentimentService init failures previously crashed
the entire StockAnalysisPipeline constructor, preventing any analysis from
running even when search/sentiment is non-essential.

Changes:
- Wrap SearchService init in try/except, fallback to None on failure
- Wrap SocialSentimentService init in try/except, fallback to None on failure
- Add null-safety guards at all 5 downstream call sites
- Upgrade progress callback log from debug to warning for better visibility

docs/CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [修复] 移除 `HistoryItem` 与 `ReportSummary` 响应 Schema 中 `sentiment_score` 的 `ge=0/le=100` 约束（fixes #942）——历史库中存储的超范围负值或大于 100 的情绪评分不再触发 Pydantic ValidationError，历史列表与详情接口恢复正常返回。
 - [改进] Agent IntelAgent 新增公司公告搜索维度（上交所/深交所/cninfo）与主力资金流工具（get_capital_flow），修复 Agent 模式下公告和资金流数据经常缺失的问题
 - [修复] webui_frontend.py 在 static/index.html 存在但 static/assets/ 缺失时发出明确警告，避免用户因 CSS/JS 资源缺失导致页面元素异常变大却无从排查
+- [修复] `StockAnalysisPipeline` 搜索服务与社交舆情服务改为可选降级初始化：任一服务初始化异常时记录 warning 并以禁用状态继续运行，避免外部依赖抖动阻塞主分析链路与 SSE 进度回调。
 - [文档] DEPLOY.md 和 deploy-webui-cloud.md 新增"UI 元素异常变大/布局错乱"排查步骤（重建 Docker 镜像或手动执行 npm run build）
 - [文档] 补充飞书 Webhook 配置说明：强调 `FEISHU_WEBHOOK_URL` 是群通知必填项、`FEISHU_WEBHOOK_SECRET` 与飞书机器人「签名校验」必须两端同时启用或同时关闭、`FEISHU_APP_SECRET` 仅用于应用/Stream Bot 模式不可替代 Webhook；同步完善英文指南并在 `.env.example` 为相关配置项补充内联说明注释
 

docs/full-guide.md

@@ -262,12 +262,16 @@ daily_stock_analysis/
 | `BOCHA_API_KEYS` | 博查搜索 API Key（中文优化） | 可选 |
 | `BRAVE_API_KEYS` | Brave Search API Key（美股优化） | 可选 |
 | `SERPAPI_API_KEYS` | SerpAPI 备用搜索 | 可选 |
+| `SOCIAL_SENTIMENT_API_KEY` | Stock Sentiment API Key（Reddit / X / Polymarket，可选） | 可选 |
+| `SOCIAL_SENTIMENT_API_URL` | Stock Sentiment API 地址（默认 `https://api.adanos.org`） | 可选 |
 | `SEARXNG_BASE_URLS` | SearXNG 自建实例（无配额兜底，需在 settings.yml 启用 format: json）；留空时默认自动发现公共实例 | 可选 |
 | `SEARXNG_PUBLIC_INSTANCES_ENABLED` | 是否在 `SEARXNG_BASE_URLS` 为空时自动从 `searx.space` 获取公共实例（默认 `true`） | 可选 |
 | `NEWS_STRATEGY_PROFILE` | 新闻策略窗口档位：`ultra_short`(1天)/`short`(3天)/`medium`(7天)/`long`(30天)；实际窗口取与 `NEWS_MAX_AGE_DAYS` 的最小值 | 默认 `short` |
 | `NEWS_MAX_AGE_DAYS` | 新闻最大时效（天），搜索时限制结果在近期内 | 默认 `3` |
 | `BIAS_THRESHOLD` | 乖离率阈值（%），超过提示不追高；强势趋势股自动放宽到 1.5 倍 | 默认 `5.0` |
 
+> 行为说明：搜索服务与社交舆情服务为可选增强链路。任一服务初始化失败时，系统会记录 warning 并降级为跳过该服务，仅影响对应环节，不会阻塞技术面主链路和主任务流。
+
 ### 数据源配置
 
 | 变量名 | 说明 | 默认值 | 必填 |
@@ -982,6 +986,7 @@ FastAPI 提供 RESTful API 服务，支持配置管理和触发分析。
 > 说明：`POST /api/v1/analysis/analyze` 在 `async_mode=false` 时仅支持单只股票；批量 `stock_codes` 需使用 `async_mode=true`。异步 `202` 响应对单股返回 `task_id`，对批量返回 `accepted` / `duplicates` 汇总结构。
 
 > 进度流说明：`GET /api/v1/analysis/tasks/stream` 除 `task_created / task_started / task_completed / task_failed` 外，新增 `task_progress` 事件。普通分析链路会在“行情准备 / 新闻检索 / 上下文整理 / LLM 生成 / 报告保存”等阶段持续更新 `progress` 与 `message`。LiteLLM 流式返回仅在服务端累积完整文本，最终 JSON 解析成功后才会持久化历史报告；若流式在首个 chunk 前不可用，会自动回退到原非流式调用；若已产生部分 chunk 后失败，系统先尝试同模型非流式重试，失败后再按既有主模型->备用模型顺序继续尝试。  
+> 如果任务进度回调异常，主链路不会中断，系统会提升告警为 warning 级别并在服务端日志中输出完整异常，便于排查 SSE 推送断点。
 >  
 > 说明：该特性属于运行时 SSE 与回退链路细节，优先记录于完整指南（`full-guide*.md`），不在 `README.md` 中展开详细行为分支。
 

docs/full-guide_EN.md

@@ -238,9 +238,13 @@ Default schedule: Every weekday at **18:00 (Beijing Time)** automatic execution.
 | `BOCHA_API_KEYS` | Bocha Search API Key (Chinese optimized) | Optional |
 | `BRAVE_API_KEYS` | Brave Search API Key (US stocks optimized) | Optional |
 | `SERPAPI_API_KEYS` | SerpAPI Backup search | Optional |
+| `SOCIAL_SENTIMENT_API_KEY` | Stock Sentiment API Key (Reddit / X / Polymarket, US stocks optional) | Optional |
+| `SOCIAL_SENTIMENT_API_URL` | Stock Sentiment API endpoint (default `https://api.adanos.org`) | Optional |
 | `SEARXNG_BASE_URLS` | SearXNG self-hosted instances (quota-free fallback, enable format: json in settings.yml); when empty the app auto-discovers public instances | Optional |
 | `SEARXNG_PUBLIC_INSTANCES_ENABLED` | Auto-discover public SearXNG instances from `searx.space` when `SEARXNG_BASE_URLS` is empty (default `true`) | Optional |
 
+> Behavior note: Search and social sentiment are optional enhancement services. If either service fails to initialize, the system logs a warning and degrades gracefully by skipping that stage without blocking the core analysis flow.
+
 ### Data Source Configuration
 
 | Variable | Description | Default | Required |
@@ -816,6 +820,7 @@ FastAPI provides RESTful API service for configuration management and triggering
 > Note: `POST /api/v1/analysis/analyze` supports only one stock when `async_mode=false`; batch `stock_codes` requires `async_mode=true`. The async `202` response returns a single `task_id` for one stock, or an `accepted` / `duplicates` summary for batch requests.
 
 > Progress-stream note: `GET /api/v1/analysis/tasks/stream` now emits `task_progress` in addition to `task_created / task_started / task_completed / task_failed`. The regular analysis path updates `progress` and `message` across quote preparation, news retrieval, context assembly, LLM generation, and report persistence. Streaming chunks are accumulated only on the server side; history is persisted only after the final JSON parses successfully. If streaming is unavailable before the first chunk, the system falls back to the previous non-stream request. If a stream fails after partial output has already arrived, the system first retries non-stream for the same model, then continues through existing fallback models in the original order (primary + fallback list).
+> If a progress callback fails, the analysis flow continues, and the exception is now logged at warning level to help troubleshoot SSE delivery gaps.
 
 > Note: This behavior is documented in the full guide (`full-guide*.md`) because it is detailed runtime SSE/fallback behavior and is therefore kept out of the README.
 

src/core/pipeline.py

@@ -102,18 +102,22 @@ class StockAnalysisPipeline:
         self.notifier = NotificationService(source_message=source_message)
         self._single_stock_notify_lock = threading.Lock()
         
-        # 初始化搜索服务
-        self.search_service = SearchService(
-            bocha_keys=self.config.bocha_api_keys,
-            tavily_keys=self.config.tavily_api_keys,
-            brave_keys=self.config.brave_api_keys,
-            serpapi_keys=self.config.serpapi_keys,
-            minimax_keys=self.config.minimax_api_keys,
-            searxng_base_urls=self.config.searxng_base_urls,
-            searxng_public_instances_enabled=self.config.searxng_public_instances_enabled,
-            news_max_age_days=self.config.news_max_age_days,
-            news_strategy_profile=getattr(self.config, "news_strategy_profile", "short"),
-        )
+        # 初始化搜索服务（可选，初始化失败不应阻断主分析流程）
+        try:
+            self.search_service = SearchService(
+                bocha_keys=self.config.bocha_api_keys,
+                tavily_keys=self.config.tavily_api_keys,
+                brave_keys=self.config.brave_api_keys,
+                serpapi_keys=self.config.serpapi_keys,
+                minimax_keys=self.config.minimax_api_keys,
+                searxng_base_urls=self.config.searxng_base_urls,
+                searxng_public_instances_enabled=self.config.searxng_public_instances_enabled,
+                news_max_age_days=self.config.news_max_age_days,
+                news_strategy_profile=getattr(self.config, "news_strategy_profile", "short"),
+            )
+        except Exception as exc:
+            logger.warning("搜索服务初始化失败，将以无搜索模式运行: %s", exc, exc_info=True)
+            self.search_service = None
         
         logger.info(f"调度器初始化完成，最大并发数: {self.max_workers}")
         logger.info("已启用技术分析引擎（均线/趋势/量价指标）")
@@ -126,18 +130,28 @@ class StockAnalysisPipeline:
             logger.info("筹码分布分析已启用")
         else:
             logger.info("筹码分布分析已禁用")
-        if self.search_service.is_available:
+        if self.search_service is None:
+            logger.warning("搜索服务未启用（初始化失败或依赖缺失）")
+        elif self.search_service.is_available:
             logger.info("搜索服务已启用")
         else:
             logger.warning("搜索服务未启用（未配置搜索能力）")
 
-        # 初始化社交舆情服务（仅美股）
-        self.social_sentiment_service = SocialSentimentService(
-            api_key=self.config.social_sentiment_api_key,
-            api_url=self.config.social_sentiment_api_url,
-        )
-        if self.social_sentiment_service.is_available:
-            logger.info("Social sentiment service enabled (Reddit/X/Polymarket, US stocks only)")
+        # 初始化社交舆情服务（仅美股，可选）
+        try:
+            self.social_sentiment_service = SocialSentimentService(
+                api_key=self.config.social_sentiment_api_key,
+                api_url=self.config.social_sentiment_api_url,
+            )
+            if self.social_sentiment_service.is_available:
+                logger.info("Social sentiment service enabled (Reddit/X/Polymarket, US stocks only)")
+        except Exception as exc:
+            logger.warning(
+                "社交舆情服务初始化失败，将跳过舆情分析: %s",
+                exc,
+                exc_info=True,
+            )
+            self.social_sentiment_service = None
 
     def _emit_progress(self, progress: int, message: str) -> None:
         """Best-effort bridge from pipeline stages to task SSE progress."""
@@ -147,7 +161,19 @@ class StockAnalysisPipeline:
         try:
             callback(progress, message)
         except Exception as exc:
-            logger.debug("[pipeline] progress callback skipped: %s", exc)
+            query_id = getattr(self, "query_id", None)
+            logger.warning(
+                "[pipeline] progress callback failed: %s (progress=%s, message=%r, query_id=%s)",
+                exc,
+                progress,
+                message,
+                query_id,
+                extra={
+                    "progress": progress,
+                    "progress_message": message,
+                    "query_id": query_id,
+                },
+            )
 
     def fetch_and_save_stock_data(
         self, 
@@ -349,7 +375,7 @@ class StockAnalysisPipeline:
             # Step 4: 多维度情报搜索（最新消息+风险排查+业绩预期）
             news_context = None
             self._emit_progress(46, f"{stock_name}：正在检索新闻与舆情")
-            if self.search_service.is_available:
+            if self.search_service is not None and self.search_service.is_available:
                 logger.info(f"{stock_name}({code}) 开始多维度情报搜索...")
 
                 # 使用多维度搜索（最多5次搜索）
@@ -387,7 +413,7 @@ class StockAnalysisPipeline:
                 logger.info(f"{stock_name}({code}) 搜索服务不可用，跳过情报搜索")
 
             # Step 4.5: Social sentiment intelligence (US stocks only)
-            if self.social_sentiment_service.is_available and is_us_stock_code(code):
+            if self.social_sentiment_service is not None and self.social_sentiment_service.is_available and is_us_stock_code(code):
                 try:
                     social_context = self.social_sentiment_service.get_social_context(code)
                     if social_context:
@@ -749,7 +775,7 @@ class StockAnalysisPipeline:
             # Agent path: inject social sentiment as news_context so both
             # executor (_build_user_message) and orchestrator (ctx.set_data)
             # can consume it through the existing news_context channel
-            if self.social_sentiment_service.is_available and is_us_stock_code(code):
+            if self.social_sentiment_service is not None and self.social_sentiment_service.is_available and is_us_stock_code(code):
                 try:
                     social_context = self.social_sentiment_service.get_social_context(code)
                     if social_context:
@@ -796,7 +822,7 @@ class StockAnalysisPipeline:
 
             # 保存新闻情报到数据库（Agent 工具结果仅用于 LLM 上下文，未持久化，Fixes #396）
             # 使用 search_stock_news（与 Agent 工具调用逻辑一致），仅 1 次 API 调用，无额外延迟
-            if self.search_service.is_available:
+            if self.search_service is not None and self.search_service.is_available:
                 try:
                     news_response = self.search_service.search_stock_news(
                         stock_code=code,

tests/test_pipeline_optional_service_resilience.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""Regression tests for optional pipeline service degradation logs."""
+
+import logging
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from src.core.pipeline import StockAnalysisPipeline
+
+
+def _make_config() -> SimpleNamespace:
+    return SimpleNamespace(
+        max_workers=2,
+        save_context_snapshot=False,
+        bocha_api_keys=[],
+        tavily_api_keys=[],
+        brave_api_keys=[],
+        serpapi_keys=[],
+        minimax_api_keys=[],
+        searxng_base_urls=[],
+        searxng_public_instances_enabled=False,
+        news_max_age_days=7,
+        news_strategy_profile="short",
+        enable_realtime_quote=False,
+        realtime_source_priority=[],
+        enable_chip_distribution=False,
+        social_sentiment_api_key="",
+        social_sentiment_api_url="https://example.invalid/social",
+    )
+
+
+def _build_pipeline(config: SimpleNamespace) -> StockAnalysisPipeline:
+    with patch("src.core.pipeline.get_db", return_value=MagicMock()), \
+         patch("src.core.pipeline.DataFetcherManager", return_value=MagicMock()), \
+         patch("src.core.pipeline.StockTrendAnalyzer", return_value=MagicMock()), \
+         patch("src.core.pipeline.GeminiAnalyzer", return_value=MagicMock()), \
+         patch("src.core.pipeline.NotificationService", return_value=MagicMock()):
+        return StockAnalysisPipeline(config=config)
+
+
+def test_search_service_init_failure_logs_traceback_and_failure_state(caplog):
+    config = _make_config()
+    social_service = MagicMock()
+    social_service.is_available = False
+
+    with patch("src.core.pipeline.SearchService", side_effect=RuntimeError("search init boom")), \
+         patch("src.core.pipeline.SocialSentimentService", return_value=social_service), \
+         caplog.at_level(logging.WARNING, logger="src.core.pipeline"):
+        pipeline = _build_pipeline(config)
+
+    assert pipeline.search_service is None
+
+    init_failure_records = [
+        record for record in caplog.records if "搜索服务初始化失败，将以无搜索模式运行" in record.message
+    ]
+    assert len(init_failure_records) == 1
+    assert init_failure_records[0].exc_info is not None
+    assert "搜索服务未启用（初始化失败或依赖缺失）" in caplog.text
+    assert "搜索服务未启用（未配置搜索能力）" not in caplog.text
+
+
+def test_social_sentiment_init_failure_logs_traceback(caplog):
+    config = _make_config()
+    search_service = MagicMock()
+    search_service.is_available = False
+
+    with patch("src.core.pipeline.SearchService", return_value=search_service), \
+         patch("src.core.pipeline.SocialSentimentService", side_effect=RuntimeError("social init boom")), \
+         caplog.at_level(logging.WARNING, logger="src.core.pipeline"):
+        pipeline = _build_pipeline(config)
+
+    assert pipeline.social_sentiment_service is None
+
+    init_failure_records = [
+        record for record in caplog.records if "社交舆情服务初始化失败，将跳过舆情分析" in record.message
+    ]
+    assert len(init_failure_records) == 1
+    assert init_failure_records[0].exc_info is not None
+
+
+def test_emit_progress_logs_context_when_callback_fails(caplog):
+    pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
+    pipeline.query_id = "query-123"
+
+    def _fail_callback(progress, message):
+        raise RuntimeError(f"cannot send {progress}:{message}")
+
+    pipeline.progress_callback = _fail_callback
+
+    with caplog.at_level(logging.WARNING, logger="src.core.pipeline"):
+        pipeline._emit_progress(55, "fetching news")
+
+    records = [record for record in caplog.records if "progress callback failed" in record.message]
+    assert len(records) == 1
+    record = records[0]
+    assert "progress=55" in record.message
+    assert "message='fetching news'" in record.message
+    assert "query_id=query-123" in record.message
+    assert record.progress == 55
+    assert record.progress_message == "fetching news"
+    assert record.query_id == "query-123"