ZhuLinsen · ZhuLinsen · Apr 1, 2026 · Mar 27, 2026 · Mar 27, 2026 · Mar 28, 2026
@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [测试] 🧪 **补充前端变更验证命令** — 对应前端资源变更同步执行 `cd apps/dsa-web && npm ci && npm run lint && npm run build`，作为版本信息展示与 Docker 重建生效验证的最小验证闭环记录。
 - [修复] 内置定时调度器现在会在运行中感知 WebUI 保存后的 `SCHEDULE_TIME` 变化，并在下一轮检查时重绑 daily job，避免 `python main.py --serve --schedule` 仍固定按启动时的 `18:00` 触发；`.env.example` 也同步删除了重复的定时任务配置示例。
 - [修复] 🪟 **Windows Release 渠道编辑器保留 MiniMax 模型前缀** — 渠道模式下填写 `minimax/<模型名>` 时，后端归一化与 Web 设置页运行时模型列表都会保留该值原样，不再误改写成 `openai/minimax/<模型名>`，从而恢复 MiniMax 模型在 Win 客户端里的保存、选择与使用。
+- [修复] 🐳 **Docker WebUI 运行时优先复用预构建静态资源** — `prepare_webui_frontend_assets()` 现在会先检查镜像内已有的 `static/index.html` 是否可直接复用；当容器运行时不包含 `apps/dsa-web` 源码目录且未安装 `npm` 时，也不会误报“未找到前端项目，无法自动构建”，从而恢复 Docker 部署后的 WebUI 打开能力。
+- [修复] 📨 **单股推送模式不再并发复用共享通知实例** — `StockAnalysisPipeline.run()` 现在会保留个股分析并发，但把 `SINGLE_STOCK_NOTIFY=true` 下的即时通知挪到结果收集侧串行发送；同时 `_send_single_stock_notification()` 为同一个 pipeline 实例补上实例级临界区，避免直接调用 `process_single_stock(..., single_stock_notify=True)` 时多个线程继续共享同一个 `NotificationService` 进入报告生成与发送链路，导致通知乱序、重复发送或状态污染。
+- [修复] 📨 **单股推送模式不再并发复用共享通知实例** — `StockAnalysisPipeline.run()` 现在会保留个股分析并发，但把 `SINGLE_STOCK_NOTIFY=true` 下的即时通知挪到结果收集侧串行发送；同时 `_send_single_stock_notification()` 为同一个 pipeline 实例补上实例级临界区，避免直接调用 `process_single_stock(..., single_stock_notify=True)` 时多个线程继续共享同一个 `NotificationService` 进入报告生成与发送链路，导致通知乱序、重复发送或状态污染。补充了 `tests/test_pipeline_single_stock_notify.py` 与 `tests/test_pipeline_single_notify_thread_safety.py` 的回归场景以覆盖并发单股推送与直接单股入口的串行化行为。
 - [修复] 🔇 **实时行情降级提示收口为单次告警** — 分析主流程获取股票名称时不再提前触发一次实时行情查询，避免每只股票重复命中 quote 链路；当某个前置实时数据源失败但后续 fallback 成功时，不再输出“实时行情获取失败”级别提示，只有在实时行情开关关闭或全部数据源都不可用时，才提示已降级为历史收盘价继续分析。
 
 ## [3.11.0] - 2026-03-27
@@ -80,6 +83,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - 🌗 **Web 首屏默认主题预设为深色** — `apps/dsa-web/index.html` 现在会在 React 挂载前读取本地保存的主题偏好；若没有已保存值，则立即给 `<html>` 预设 `dark` 并同步 `color-scheme`，避免首页和登录页首屏先闪出浅色主题。
 - 🔐 **登录页独立主题层收口** — 登录页输入框、标签、切换按钮和按钮文案现在使用独立的 `--login-*` 视觉 token，不再继承全局浅/深主题文字色；即使浏览器缓存了浅色主题，登录页仍保持稳定的深色视觉与青色密码输入表现，避免密码圆点和文案落成黑色。
 - 🖥️ **首页港股代码输入修复** — Web 首页分析输入框现在可正确接受港股代码与自动完成选中的港股项，补齐 `00700.HK` / `HK00700` 等格式识别，避免提交时误报“请输入有效的股票代码或股票名称”。
+
 - 🔒 **认证限流 X-Forwarded-For 取值修复（CWE-345）**（#841 / #842）— `get_client_ip()` 从取 `X-Forwarded-For` 最左值改为最右值，防止攻击者通过伪造首部旋转限流桶绕过暴力破解保护；仅影响 `TRUST_X_FORWARDED_FOR=true` 且单层可信反向代理的部署场景，多级代理环境需按部署文档评估配置。
 - 📦 **恢复 LiteLLM 官方 PyPI 安装并锁定安全上限** — `requirements.txt` 重新使用 `pip install litellm` 的官方 PyPI 安装路径，并在保留历史最低要求 `>=1.80.10` 的同时增加 `<1.82.7` 的安全上限，避免误装已被移除的 `1.82.7` / `1.82.8` 风险版本；Windows 桌面打包脚本也同步回退到标准 `pip install -r requirements.txt` 链路，减少特殊下载分支带来的维护成本。
 - 📨 **Telegram Markdown 解析失败回退纯文本**（fixes #850）— `src/notification_sender/telegram_sender.py` 现在会在 Telegram 返回 `HTTP 400` 且包含 `can't parse entities` / Markdown 解析错误时，自动去掉 `parse_mode` 后重试纯文本发送，避免 `*ST` 等正文内容直接导致整条通知失败。

@@ -12,6 +12,7 @@
 """
 
 import logging
+import threading
 import time
 import uuid
 from collections import defaultdict
@@ -45,6 +46,10 @@
 
 logger = logging.getLogger(__name__)
 
+# 防御性 guard：当实例绕过 __init__（如测试中 __new__）构造时，
+# double-check 初始化 _single_stock_notify_lock 仍然线程安全。
+_SINGLE_STOCK_NOTIFY_LOCK_INIT_GUARD = threading.Lock()
+
 
 class StockAnalysisPipeline:
     """
@@ -88,6 +93,7 @@ def __init__(
         self.trend_analyzer = StockTrendAnalyzer()  # 技术分析器
         self.analyzer = GeminiAnalyzer(config=self.config)
         self.notifier = NotificationService(source_message=source_message)
+        self._single_stock_notify_lock = threading.Lock()
 
         # 初始化搜索服务
         self.search_service = SearchService(
@@ -1138,25 +1144,12 @@ def process_single_stock(
                     )
 
                 # 单股推送模式（#55）：每分析完一只股票立即推送
-                if single_stock_notify and self.notifier.is_available():
-                    try:
-                        # 根据报告类型选择生成方法
-                        if report_type == ReportType.FULL:
-                            report_content = self.notifier.generate_dashboard_report([result])
-                            logger.info(f"[{code}] 使用完整报告格式")
-                        elif report_type == ReportType.BRIEF:
-                            report_content = self.notifier.generate_brief_report([result])
-                            logger.info(f"[{code}] 使用简洁报告格式")
-                        else:
-                            report_content = self.notifier.generate_single_stock_report(result)
-                            logger.info(f"[{code}] 使用精简报告格式")
-
-                        if self.notifier.send(report_content, email_stock_codes=[code]):
-                            logger.info(f"[{code}] 单股推送成功")
-                        else:
-                            logger.warning(f"[{code}] 单股推送失败")
-                    except Exception as e:
-                        logger.error(f"[{code}] 单股推送异常: {e}")
+                if single_stock_notify:
+                    self._send_single_stock_notification(
+                        result,
+                        report_type=report_type,
+                        fallback_code=code,
+                    )
 
             return result
 
@@ -1231,7 +1224,10 @@ def run(
         analysis_delay = getattr(self.config, 'analysis_delay', 0)
 
         if single_stock_notify:
-            logger.info(f"已启用单股推送模式：每分析完一只股票立即推送（报告类型: {report_type_str}）")
+            logger.info(
+                "已启用单股推送模式：分析仍并发执行，通知改为在结果收集侧串行发送（报告类型: %s）",
+                report_type_str,
+            )
 
         results: List[AnalysisResult] = []
 
@@ -1244,7 +1240,7 @@ def run(
                     self.process_single_stock,
                     code,
                     skip_analysis=dry_run,
-                    single_stock_notify=single_stock_notify and send_notification,
+                    single_stock_notify=False,
                     report_type=report_type,  # Issue #119: 传递报告类型
                     analysis_query_id=uuid.uuid4().hex,
                 ): code
@@ -1258,6 +1254,12 @@ def run(
                     result = future.result()
                     if result:
                         results.append(result)
+                        if single_stock_notify and send_notification and not dry_run:
+                            self._send_single_stock_notification(
+                                result,
+                                report_type=report_type,
+                                fallback_code=code,
+                            )
 
                     # Issue #128: 分析间隔 - 在个股分析和大盘分析之间添加延迟
                     if idx < len(stock_codes) - 1 and analysis_delay > 0:
@@ -1304,7 +1306,45 @@ def run(
                 self._send_notifications(results, report_type)
 
         return results
-
+
+    def _send_single_stock_notification(
+        self,
+        result: AnalysisResult,
+        report_type: ReportType = ReportType.SIMPLE,
+        fallback_code: Optional[str] = None,
+    ) -> None:
+        """发送单股通知，供直接单股入口和批量串行推送共用。"""
+        if not self.notifier.is_available():
+            return
+
+        stock_code = getattr(result, "code", None) or fallback_code or "unknown"
+        notify_lock = getattr(self, "_single_stock_notify_lock", None)
+        if notify_lock is None:
+            with _SINGLE_STOCK_NOTIFY_LOCK_INIT_GUARD:
+                notify_lock = getattr(self, "_single_stock_notify_lock", None)
+                if notify_lock is None:
+                    notify_lock = threading.Lock()
+                    setattr(self, "_single_stock_notify_lock", notify_lock)
+
+        with notify_lock:
+            try:
+                if report_type == ReportType.FULL:
+                    report_content = self.notifier.generate_dashboard_report([result])
+                    logger.info(f"[{stock_code}] 使用完整报告格式")
+                elif report_type == ReportType.BRIEF:
+                    report_content = self.notifier.generate_brief_report([result])
+                    logger.info(f"[{stock_code}] 使用简洁报告格式")
+                else:
+                    report_content = self.notifier.generate_single_stock_report(result)
+                    logger.info(f"[{stock_code}] 使用精简报告格式")
+
+                if self.notifier.send(report_content, email_stock_codes=[stock_code]):
+                    logger.info(f"[{stock_code}] 单股推送成功")
+                else:
+                    logger.warning(f"[{stock_code}] 单股推送失败")
+            except Exception as e:
+                logger.error(f"[{stock_code}] 单股推送异常: {e}")
+
     def _save_local_report(
         self,
         results: List[AnalysisResult],

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+"""
+Regression tests for single-stock notification thread safety.
+"""
+
+import os
+import sys
+import threading
+import time
+import unittest
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from tests.litellm_stub import ensure_litellm_stub
+
+ensure_litellm_stub()
+
+from src.analyzer import AnalysisResult
+from src.core.pipeline import StockAnalysisPipeline
+
+
+def _make_result(code: str) -> AnalysisResult:
+    return AnalysisResult(
+        code=code,
+        name=f"股票{code}",
+        sentiment_score=80,
+        trend_prediction="看多",
+        operation_advice="持有",
+        analysis_summary="测试结果",
+    )
+
+
+class _CriticalSectionTrackingNotifier:
+    def __init__(self):
+        self._state_lock = threading.Lock()
+        self._inflight = 0
+        self.max_inflight = 0
+        self.calls = []
+        self.is_available = MagicMock(return_value=True)
+        self.generate_single_stock_report = MagicMock(
+            side_effect=self._generate_single_stock_report
+        )
+        self.send = MagicMock(side_effect=self._send)
+
+    def _enter(self, stage: str, code: str) -> None:
+        with self._state_lock:
+            self._inflight += 1
+            self.max_inflight = max(self.max_inflight, self._inflight)
+
+        self.calls.append((stage, code, threading.current_thread().name))
+        time.sleep(0.02)
+
+        with self._state_lock:
+            self._inflight -= 1
+
+    def _generate_single_stock_report(self, result: AnalysisResult) -> str:
+        self._enter("generate", result.code)
+        return f"single:{result.code}"
+
+    def _send(self, content: str, email_stock_codes=None) -> bool:
+        stock_code = (email_stock_codes or ["unknown"])[0]
+        self._enter("send", stock_code)
+        return True
+
+
+class TestPipelineSingleNotifyThreadSafety(unittest.TestCase):
+    def test_process_single_stock_serializes_direct_notification_path(self):
+        pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
+        pipeline.fetch_and_save_stock_data = MagicMock(return_value=(True, None))
+        pipeline.notifier = _CriticalSectionTrackingNotifier()
+
+        notify_barrier = threading.Barrier(2)
+
+        def _analyze(code, report_type, query_id):
+            notify_barrier.wait(timeout=10)
+            return _make_result(code)
+
+        pipeline.analyze_stock = MagicMock(side_effect=_analyze)
+
+        results = []
+        result_lock = threading.Lock()
+
+        def _worker(code: str) -> None:
+            result = pipeline.process_single_stock(
+                code=code,
+                single_stock_notify=True,
+                analysis_query_id=f"query-{code}",
+            )
+            with result_lock:
+                results.append(result)
+
+        threads = [
+            threading.Thread(target=_worker, args=(code,), name=f"notify-{code}")
+            for code in ("000001", "600519")
+        ]
+
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(len(results), 2)
+        self.assertTrue(all(result is not None for result in results))
+        self.assertEqual(pipeline.notifier.generate_single_stock_report.call_count, 2)
+        self.assertEqual(pipeline.notifier.send.call_count, 2)
+        self.assertEqual(pipeline.notifier.max_inflight, 1)
+        self.assertCountEqual(
+            [(stage, code) for stage, code, _ in pipeline.notifier.calls],
+            [
+                ("generate", "000001"),
+                ("send", "000001"),
+                ("generate", "600519"),
+                ("send", "600519"),
+            ],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()