feature: 机器人命令交互触发分析功能

.env.example

@@ -97,6 +97,13 @@ SERPAPI_API_KEYS=your_serpapi_key_here
 # FEISHU_MAX_BYTES=20000    # 飞书限制约 20KB，默认 20000 字节
 # WECHAT_MAX_BYTES=4000     # 企业微信限制 4096 字节，默认 4000 字节
 
+# 应用 AppKey（与 Webhook 模式共用）
+DINGTALK_APP_KEY=xxxx
+# 应用 AppSecret（与 Webhook 模式共用）
+DINGTALK_APP_SECRET=xxxx
+# 启用 Stream 模式
+DINGTALK_STREAM_ENABLED=true
+
 # 数据库路径
 DATABASE_PATH=./data/stock_analysis.db
 

Dockerfile

@@ -22,12 +22,13 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 COPY requirements.txt .
 
 # 安装 Python 依赖
-RUN pip install --no-cache-dir -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/
+RUN pip install --no-cache-dir -r requirements.txt
 
 # 复制应用代码
 COPY *.py ./
 COPY data_provider/ ./data_provider/
 COPY web/ ./web/
+COPY bot/ ./bot/
 
 # 创建数据目录
 RUN mkdir -p /app/data /app/logs /app/reports

bot/__init__.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+机器人命令触发系统
+===================================
+
+通过 @机器人 或发送命令触发股票分析等功能。
+支持飞书、钉钉、企业微信、Telegram 等多平台。
+
+模块结构：
+- models.py: 统一的消息/响应模型
+- dispatcher.py: 命令分发器
+- commands/: 命令处理器
+- platforms/: 平台适配器
+- handler.py: Webhook 处理器
+
+使用方式：
+1. 配置环境变量（各平台的 Token 等）
+2. 启动 WebUI 服务
+3. 在各平台配置 Webhook URL：
+   - 飞书: http://your-server/bot/feishu
+   - 钉钉: http://your-server/bot/dingtalk
+   - 企业微信: http://your-server/bot/wecom
+   - Telegram: http://your-server/bot/telegram
+
+支持的命令：
+- /analyze <股票代码>  - 分析指定股票
+- /market             - 大盘复盘
+- /batch              - 批量分析自选股
+- /help               - 显示帮助
+- /status             - 系统状态
+"""
+
+from bot.models import BotMessage, BotResponse, ChatType, WebhookResponse
+from bot.dispatcher import CommandDispatcher, get_dispatcher
+
+__all__ = [
+    'BotMessage',
+    'BotResponse',
+    'ChatType',
+    'WebhookResponse',
+    'CommandDispatcher',
+    'get_dispatcher',
+]

bot/commands/__init__.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+命令处理器模块
+===================================
+
+包含所有机器人命令的实现。
+"""
+
+from bot.commands.base import BotCommand
+from bot.commands.help import HelpCommand
+from bot.commands.status import StatusCommand
+from bot.commands.analyze import AnalyzeCommand
+from bot.commands.market import MarketCommand
+from bot.commands.batch import BatchCommand
+
+# 所有可用命令（用于自动注册）
+ALL_COMMANDS = [
+    HelpCommand,
+    StatusCommand,
+    AnalyzeCommand,
+    MarketCommand,
+    BatchCommand,
+]
+
+__all__ = [
+    'BotCommand',
+    'HelpCommand',
+    'StatusCommand',
+    'AnalyzeCommand',
+    'MarketCommand',
+    'BatchCommand',
+    'ALL_COMMANDS',
+]

bot/commands/analyze.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+股票分析命令
+===================================
+
+分析指定股票，调用 AI 生成分析报告。
+"""
+
+import re
+import logging
+from typing import List, Optional
+
+from bot.commands.base import BotCommand
+from bot.models import BotMessage, BotResponse
+
+logger = logging.getLogger(__name__)
+
+
+class AnalyzeCommand(BotCommand):
+    """
+    股票分析命令
+    
+    分析指定股票代码，生成 AI 分析报告并推送。
+    
+    用法：
+        /analyze 600519       - 分析贵州茅台
+        /analyze 600519 full  - 分析并生成完整报告
+    """
+    
+    @property
+    def name(self) -> str:
+        return "analyze"
+    
+    @property
+    def aliases(self) -> List[str]:
+        return ["a", "分析", "查"]
+    
+    @property
+    def description(self) -> str:
+        return "分析指定股票"
+    
+    @property
+    def usage(self) -> str:
+        return "/analyze <股票代码> [full]"
+    
+    def validate_args(self, args: List[str]) -> Optional[str]:
+        """验证参数"""
+        if not args:
+            return "请输入股票代码"
+        
+        code = args[0].lower()
+        
+        # 验证股票代码格式
+        # A股：6位数字
+        # 港股：hk + 5位数字
+        if not (re.match(r'^\d{6}$', code) or re.match(r'^hk\d{5}$', code)):
+            return f"无效的股票代码: {code}（A股6位数字，港股hk+5位数字）"
+        
+        return None
+    
+    def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """执行分析命令"""
+        code = args[0].lower()
+        
+        # 检查是否需要完整报告
+        report_type = "full"
+        # if len(args) > 1 and args[1].lower() in ["full", "完整", "详细"]:
+        #     report_type = "full"
+        logger.info(f"[AnalyzeCommand] 分析股票: {code}, 报告类型: {report_type}")
+        
+        try:
+            # 调用分析服务
+            from web.services import get_analysis_service
+            from enums import ReportType
+            
+            service = get_analysis_service()
+            
+            # 提交异步分析任务
+            result = service.submit_analysis(
+                code=code,
+                report_type=ReportType.from_str(report_type),
+                source_message=message
+            )
+            
+            if result.get("success"):
+                task_id = result.get("task_id", "")
+                return BotResponse.markdown_response(
+                    f"✅ **分析任务已提交**\n\n"
+                    f"• 股票代码: `{code}`\n"
+                    f"• 报告类型: {ReportType.from_str(report_type).display_name}\n"
+                    f"• 任务 ID: `{task_id[:20]}...`\n\n"
+                    f"分析完成后将自动推送结果。"
+                )
+            else:
+                error = result.get("error", "未知错误")
+                return BotResponse.error_response(f"提交分析任务失败: {error}")
+                
+        except Exception as e:
+            logger.error(f"[AnalyzeCommand] 执行失败: {e}")
+            return BotResponse.error_response(f"分析失败: {str(e)[:100]}")

bot/commands/base.py

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+命令基类
+===================================
+
+定义命令处理器的抽象基类，所有命令都必须继承此类。
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Optional
+
+from bot.models import BotMessage, BotResponse
+
+
+class BotCommand(ABC):
+    """
+    命令处理器抽象基类
+    
+    所有命令都必须继承此类并实现抽象方法。
+    
+    使用示例：
+        class MyCommand(BotCommand):
+            @property
+            def name(self) -> str:
+                return "mycommand"
+            
+            @property
+            def aliases(self) -> List[str]:
+                return ["mc", "我的命令"]
+            
+            @property
+            def description(self) -> str:
+                return "这是我的命令"
+            
+            @property
+            def usage(self) -> str:
+                return "/mycommand [参数]"
+            
+            def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+                return BotResponse.text_response("命令执行成功")
+    """
+    
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """
+        命令名称（不含前缀）
+        
+        例如 "analyze"，用户输入 "/analyze" 触发
+        """
+        pass
+    
+    @property
+    @abstractmethod
+    def aliases(self) -> List[str]:
+        """
+        命令别名列表
+        
+        例如 ["a", "分析"]，用户输入 "/a" 或 "分析" 也能触发
+        """
+        pass
+    
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """命令描述（用于帮助信息）"""
+        pass
+    
+    @property
+    @abstractmethod
+    def usage(self) -> str:
+        """
+        使用说明（用于帮助信息）
+        
+        例如 "/analyze <股票代码>"
+        """
+        pass
+    
+    @property
+    def hidden(self) -> bool:
+        """
+        是否在帮助列表中隐藏
+        
+        默认 False，设为 True 则不显示在 /help 列表中
+        """
+        return False
+    
+    @property
+    def admin_only(self) -> bool:
+        """
+        是否仅管理员可用
+        
+        默认 False，设为 True 则需要管理员权限
+        """
+        return False
+    
+    @abstractmethod
+    def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """
+        执行命令
+        
+        Args:
+            message: 原始消息对象
+            args: 命令参数列表（已分割）
+            
+        Returns:
+            BotResponse 响应对象
+        """
+        pass
+    
+    def validate_args(self, args: List[str]) -> Optional[str]:
+        """
+        验证参数
+        
+        子类可重写此方法进行参数校验。
+        
+        Args:
+            args: 命令参数列表
+            
+        Returns:
+            如果参数有效返回 None，否则返回错误信息
+        """
+        return None
+    
+    def get_help_text(self) -> str:
+        """获取帮助文本"""
+        return f"**{self.name}** - {self.description}\n用法: `{self.usage}`"

bot/commands/batch.py

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+批量分析命令
+===================================
+
+批量分析自选股列表中的所有股票。
+"""
+
+import logging
+import threading
+from typing import List
+
+from bot.commands.base import BotCommand
+from bot.models import BotMessage, BotResponse
+
+logger = logging.getLogger(__name__)
+
+
+class BatchCommand(BotCommand):
+    """
+    批量分析命令
+    
+    批量分析配置中的自选股列表，生成汇总报告。
+    
+    用法：
+        /batch      - 分析所有自选股
+        /batch 3    - 只分析前3只
+    """
+    
+    @property
+    def name(self) -> str:
+        return "batch"
+    
+    @property
+    def aliases(self) -> List[str]:
+        return ["b", "批量", "全部"]
+    
+    @property
+    def description(self) -> str:
+        return "批量分析自选股"
+    
+    @property
+    def usage(self) -> str:
+        return "/batch [数量]"
+    
+    @property
+    def admin_only(self) -> bool:
+        """批量分析需要管理员权限（防止滥用）"""
+        return False  # 可以根据需要设为 True
+    
+    def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """执行批量分析命令"""
+        from config import get_config
+        
+        config = get_config()
+        config.refresh_stock_list()
+        
+        stock_list = config.stock_list
+        
+        if not stock_list:
+            return BotResponse.error_response(
+                "自选股列表为空，请先配置 STOCK_LIST"
+            )
+        
+        # 解析数量参数
+        limit = None
+        if args:
+            try:
+                limit = int(args[0])
+                if limit <= 0:
+                    return BotResponse.error_response("数量必须大于0")
+            except ValueError:
+                return BotResponse.error_response(f"无效的数量: {args[0]}")
+        
+        # 限制分析数量
+        if limit:
+            stock_list = stock_list[:limit]
+        
+        logger.info(f"[BatchCommand] 开始批量分析 {len(stock_list)} 只股票")
+        
+        # 在后台线程中执行分析
+        thread = threading.Thread(
+            target=self._run_batch_analysis,
+            args=(stock_list, message),
+            daemon=True
+        )
+        thread.start()
+        
+        return BotResponse.markdown_response(
+            f"✅ **批量分析任务已启动**\n\n"
+            f"• 分析数量: {len(stock_list)} 只\n"
+            f"• 股票列表: {', '.join(stock_list[:5])}"
+            f"{'...' if len(stock_list) > 5 else ''}\n\n"
+            f"分析完成后将自动推送汇总报告。"
+        )
+    
+    def _run_batch_analysis(self, stock_list: List[str], message: BotMessage) -> None:
+        """后台执行批量分析"""
+        try:
+            from config import get_config
+            from main import StockAnalysisPipeline
+            
+            config = get_config()
+            
+            # 创建分析管道
+            pipeline = StockAnalysisPipeline(config=config)
+            
+            # 执行分析（会自动推送汇总报告）
+            results = pipeline.run(
+                stock_codes=stock_list,
+                dry_run=False,
+                send_notification=True
+            )
+            
+            logger.info(f"[BatchCommand] 批量分析完成，成功 {len(results)} 只")
+            
+        except Exception as e:
+            logger.error(f"[BatchCommand] 批量分析失败: {e}")
+            logger.exception(e)

bot/commands/help.py

@@ -0,0 +1,127 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+帮助命令
+===================================
+
+显示可用命令列表和使用说明。
+"""
+
+from typing import List
+
+from bot.commands.base import BotCommand
+from bot.models import BotMessage, BotResponse
+
+
+class HelpCommand(BotCommand):
+    """
+    帮助命令
+    
+    显示所有可用命令的列表和使用说明。
+    也可以查看特定命令的详细帮助。
+    
+    用法：
+        /help         - 显示所有命令
+        /help analyze - 显示 analyze 命令的详细帮助
+    """
+    
+    @property
+    def name(self) -> str:
+        return "help"
+    
+    @property
+    def aliases(self) -> List[str]:
+        return ["h", "帮助", "?"]
+    
+    @property
+    def description(self) -> str:
+        return "显示帮助信息"
+    
+    @property
+    def usage(self) -> str:
+        return "/help [命令名]"
+    
+    def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """执行帮助命令"""
+        # 延迟导入避免循环依赖
+        from bot.dispatcher import get_dispatcher
+        
+        dispatcher = get_dispatcher()
+        
+        # 如果指定了命令名，显示该命令的详细帮助
+        if args:
+            cmd_name = args[0]
+            command = dispatcher.get_command(cmd_name)
+            
+            if command is None:
+                return BotResponse.error_response(f"未知命令: {cmd_name}")
+            
+            # 构建详细帮助
+            help_text = self._format_command_help(command, dispatcher.command_prefix)
+            return BotResponse.markdown_response(help_text)
+        
+        # 显示所有命令列表
+        commands = dispatcher.list_commands(include_hidden=False)
+        prefix = dispatcher.command_prefix
+        
+        help_text = self._format_help_list(commands, prefix)
+        return BotResponse.markdown_response(help_text)
+    
+    def _format_help_list(self, commands: List[BotCommand], prefix: str) -> str:
+        """格式化命令列表"""
+        lines = [
+            "📚 **股票分析助手 - 命令帮助**",
+            "",
+            "可用命令：",
+            "",
+        ]
+        
+        for cmd in commands:
+            # 命令名和别名
+            aliases_str = ""
+            if cmd.aliases:
+                # 过滤掉中文别名，只显示英文别名
+                en_aliases = [a for a in cmd.aliases if a.isascii()]
+                if en_aliases:
+                    aliases_str = f" ({', '.join(prefix + a for a in en_aliases[:2])})"
+            
+            lines.append(f"• {prefix}{cmd.name}{aliases_str} - {cmd.description}")
+            lines.append("")
+
+        lines.extend([
+            "",
+            "---",
+            f"💡 输入 {prefix}help <命令名> 查看详细用法",
+            "",
+            "**示例：**",
+            "",
+            f"• {prefix}analyze 301023 - 奕帆传动",
+            "",
+            f"• {prefix}market - 查看大盘复盘",
+            "",
+            f"• {prefix}batch - 批量分析自选股",
+        ])
+        
+        return "\n".join(lines)
+    
+    def _format_command_help(self, command: BotCommand, prefix: str) -> str:
+        """格式化单个命令的详细帮助"""
+        lines = [
+            f"📖 **{prefix}{command.name}** - {command.description}",
+            "",
+            f"**用法：** `{command.usage}`",
+            "",
+        ]
+        
+        # 别名
+        if command.aliases:
+            aliases = [f"`{prefix}{a}`" if a.isascii() else f"`{a}`" for a in command.aliases]
+            lines.append(f"**别名：** {', '.join(aliases)}")
+            lines.append("")
+        
+        # 权限
+        if command.admin_only:
+            lines.append("⚠️ **需要管理员权限**")
+            lines.append("")
+        
+        return "\n".join(lines)

bot/commands/market.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+大盘复盘命令
+===================================
+
+执行大盘复盘分析，生成市场概览报告。
+"""
+
+import logging
+import threading
+from typing import List
+
+from bot.commands.base import BotCommand
+from bot.models import BotMessage, BotResponse
+
+logger = logging.getLogger(__name__)
+
+
+class MarketCommand(BotCommand):
+    """
+    大盘复盘命令
+    
+    执行大盘复盘分析，包括：
+    - 主要指数表现
+    - 板块热点
+    - 市场情绪
+    - 后市展望
+    
+    用法：
+        /market - 执行大盘复盘
+    """
+
+    @property
+    def name(self) -> str:
+        return "market"
+
+    @property
+    def aliases(self) -> List[str]:
+        return ["m", "大盘", "复盘", "行情"]
+
+    @property
+    def description(self) -> str:
+        return "大盘复盘分析"
+
+    @property
+    def usage(self) -> str:
+        return "/market"
+
+    def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """执行大盘复盘命令"""
+        logger.info(f"[MarketCommand] 开始大盘复盘分析")
+
+        # 在后台线程中执行复盘（避免阻塞）
+        thread = threading.Thread(
+            target=self._run_market_review,
+            args=(message,),
+            daemon=True
+        )
+        thread.start()
+
+        return BotResponse.markdown_response(
+            "✅ **大盘复盘任务已启动**\n\n"
+            "正在分析：\n"
+            "• 主要指数表现\n"
+            "• 板块热点分析\n"
+            "• 市场情绪判断\n"
+            "• 后市展望\n\n"
+            "分析完成后将自动推送结果。"
+        )
+
+    def _run_market_review(self, message: BotMessage) -> None:
+        """后台执行大盘复盘"""
+        try:
+            from config import get_config
+            from notification import NotificationService
+            from market_analyzer import MarketAnalyzer
+            from search_service import SearchService
+            from analyzer import GeminiAnalyzer
+
+            config = get_config()
+            notifier = NotificationService(source_message=message)
+
+            # 初始化搜索服务
+            search_service = None
+            if config.bocha_api_keys or config.tavily_api_keys or config.serpapi_keys:
+                search_service = SearchService(
+                    bocha_keys=config.bocha_api_keys,
+                    tavily_keys=config.tavily_api_keys,
+                    serpapi_keys=config.serpapi_keys
+                )
+
+            # 初始化 AI 分析器
+            analyzer = None
+            if config.gemini_api_key or config.openai_api_key:
+                analyzer = GeminiAnalyzer()
+
+            # 执行复盘
+            market_analyzer = MarketAnalyzer(
+                search_service=search_service,
+                analyzer=analyzer
+            )
+
+            review_report = market_analyzer.run_daily_review()
+
+            if review_report:
+                # 推送结果
+                report_content = f"🎯 **大盘复盘**\n\n{review_report}"
+                notifier.send(report_content)
+                logger.info("[MarketCommand] 大盘复盘完成并已推送")
+            else:
+                logger.warning("[MarketCommand] 大盘复盘返回空结果")
+
+        except Exception as e:
+            logger.error(f"[MarketCommand] 大盘复盘失败: {e}")
+            logger.exception(e)

bot/commands/status.py

@@ -0,0 +1,145 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+状态命令
+===================================
+
+显示系统运行状态和配置信息。
+"""
+
+import platform
+import sys
+from datetime import datetime
+from typing import List
+
+from bot.commands.base import BotCommand
+from bot.models import BotMessage, BotResponse
+
+
+class StatusCommand(BotCommand):
+    """
+    状态命令
+    
+    显示系统运行状态，包括：
+    - 服务状态
+    - 配置信息
+    - 可用功能
+    """
+    
+    @property
+    def name(self) -> str:
+        return "status"
+    
+    @property
+    def aliases(self) -> List[str]:
+        return ["s", "状态", "info"]
+    
+    @property
+    def description(self) -> str:
+        return "显示系统状态"
+    
+    @property
+    def usage(self) -> str:
+        return "/status"
+    
+    def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """执行状态命令"""
+        from config import get_config
+        
+        config = get_config()
+        
+        # 收集状态信息
+        status_info = self._collect_status(config)
+        
+        # 格式化输出
+        text = self._format_status(status_info, message.platform)
+        
+        return BotResponse.markdown_response(text)
+    
+    def _collect_status(self, config) -> dict:
+        """收集系统状态信息"""
+        status = {
+            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "python_version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
+            "platform": platform.system(),
+            "stock_count": len(config.stock_list),
+            "stock_list": config.stock_list[:5],  # 只显示前5个
+        }
+        
+        # AI 配置状态
+        status["ai_gemini"] = bool(config.gemini_api_key)
+        status["ai_openai"] = bool(config.openai_api_key)
+        
+        # 搜索服务状态
+        status["search_bocha"] = len(config.bocha_api_keys) > 0
+        status["search_tavily"] = len(config.tavily_api_keys) > 0
+        status["search_serpapi"] = len(config.serpapi_keys) > 0
+        
+        # 通知渠道状态
+        status["notify_wechat"] = bool(config.wechat_webhook_url)
+        status["notify_feishu"] = bool(config.feishu_webhook_url)
+        status["notify_telegram"] = bool(config.telegram_bot_token and config.telegram_chat_id)
+        status["notify_email"] = bool(config.email_sender and config.email_password)
+        
+        return status
+    
+    def _format_status(self, status: dict, platform: str) -> str:
+        """格式化状态信息"""
+        # 状态图标
+        def icon(enabled: bool) -> str:
+            return "✅" if enabled else "❌"
+        
+        lines = [
+            "📊 **股票分析助手 - 系统状态**",
+            "",
+            f"🕐 时间: {status['timestamp']}",
+            f"🐍 Python: {status['python_version']}",
+            f"💻 平台: {status['platform']}",
+            "",
+            "---",
+            "",
+            "**📈 自选股配置**",
+            f"• 股票数量: {status['stock_count']} 只",
+        ]
+        
+        if status['stock_list']:
+            stocks_preview = ", ".join(status['stock_list'])
+            if status['stock_count'] > 5:
+                stocks_preview += f" ... 等 {status['stock_count']} 只"
+            lines.append(f"• 股票列表: {stocks_preview}")
+        
+        lines.extend([
+            "",
+            "**🤖 AI 分析服务**",
+            f"• Gemini API: {icon(status['ai_gemini'])}",
+            f"• OpenAI API: {icon(status['ai_openai'])}",
+            "",
+            "**🔍 搜索服务**",
+            f"• Bocha: {icon(status['search_bocha'])}",
+            f"• Tavily: {icon(status['search_tavily'])}",
+            f"• SerpAPI: {icon(status['search_serpapi'])}",
+            "",
+            "**📢 通知渠道**",
+            f"• 企业微信: {icon(status['notify_wechat'])}",
+            f"• 飞书: {icon(status['notify_feishu'])}",
+            f"• Telegram: {icon(status['notify_telegram'])}",
+            f"• 邮件: {icon(status['notify_email'])}",
+        ])
+        
+        # AI 服务总体状态
+        ai_available = status['ai_gemini'] or status['ai_openai']
+        if ai_available:
+            lines.extend([
+                "",
+                "---",
+                "✅ **系统就绪，可以开始分析！**",
+            ])
+        else:
+            lines.extend([
+                "",
+                "---",
+                "⚠️ **AI 服务未配置，分析功能不可用**",
+                "请配置 Gemini 或 OpenAI API Key",
+            ])
+        
+        return "\n".join(lines)

bot/dispatcher.py

@@ -0,0 +1,342 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+命令分发器
+===================================
+
+负责解析命令、匹配处理器、分发执行。
+"""
+
+import logging
+import time
+from collections import defaultdict
+from typing import Dict, List, Optional, Type, Callable
+
+from bot.models import BotMessage, BotResponse
+from bot.commands.base import BotCommand
+
+logger = logging.getLogger(__name__)
+
+
+class RateLimiter:
+    """
+    简单的频率限制器
+    
+    基于滑动窗口算法，限制每个用户的请求频率。
+    """
+    
+    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
+        """
+        Args:
+            max_requests: 窗口内最大请求数
+            window_seconds: 窗口时间（秒）
+        """
+        self.max_requests = max_requests
+        self.window_seconds = window_seconds
+        self._requests: Dict[str, List[float]] = defaultdict(list)
+    
+    def is_allowed(self, user_id: str) -> bool:
+        """
+        检查用户是否允许请求
+        
+        Args:
+            user_id: 用户标识
+            
+        Returns:
+            是否允许
+        """
+        now = time.time()
+        window_start = now - self.window_seconds
+        
+        # 清理过期记录
+        self._requests[user_id] = [
+            t for t in self._requests[user_id] 
+            if t > window_start
+        ]
+        
+        # 检查是否超限
+        if len(self._requests[user_id]) >= self.max_requests:
+            return False
+        
+        # 记录本次请求
+        self._requests[user_id].append(now)
+        return True
+    
+    def get_remaining(self, user_id: str) -> int:
+        """获取剩余可用请求数"""
+        now = time.time()
+        window_start = now - self.window_seconds
+        
+        # 清理过期记录
+        self._requests[user_id] = [
+            t for t in self._requests[user_id] 
+            if t > window_start
+        ]
+        
+        return max(0, self.max_requests - len(self._requests[user_id]))
+
+
+class CommandDispatcher:
+    """
+    命令分发器
+    
+    职责：
+    1. 注册和管理命令处理器
+    2. 解析消息中的命令和参数
+    3. 分发命令到对应处理器
+    4. 处理未知命令和错误
+    
+    使用示例：
+        dispatcher = CommandDispatcher()
+        dispatcher.register(AnalyzeCommand())
+        dispatcher.register(HelpCommand())
+        
+        response = dispatcher.dispatch(message)
+    """
+    
+    def __init__(
+        self, 
+        command_prefix: str = "/",
+        rate_limit_requests: int = 10,
+        rate_limit_window: int = 60,
+        admin_users: Optional[List[str]] = None
+    ):
+        """
+        Args:
+            command_prefix: 命令前缀，默认 "/"
+            rate_limit_requests: 频率限制：窗口内最大请求数
+            rate_limit_window: 频率限制：窗口时间（秒）
+            admin_users: 管理员用户 ID 列表
+        """
+        self.command_prefix = command_prefix
+        self.admin_users = set(admin_users or [])
+        
+        self._commands: Dict[str, BotCommand] = {}
+        self._aliases: Dict[str, str] = {}
+        self._rate_limiter = RateLimiter(rate_limit_requests, rate_limit_window)
+        
+        # 回调函数：获取帮助命令的命令列表
+        self._help_command_getter: Optional[Callable] = None
+    
+    def register(self, command: BotCommand) -> None:
+        """
+        注册命令
+        
+        Args:
+            command: 命令实例
+        """
+        name = command.name.lower()
+        
+        if name in self._commands:
+            logger.warning(f"[Dispatcher] 命令 '{name}' 已存在，将被覆盖")
+        
+        self._commands[name] = command
+        logger.debug(f"[Dispatcher] 注册命令: {name}")
+        
+        # 注册别名
+        for alias in command.aliases:
+            alias_lower = alias.lower()
+            if alias_lower in self._aliases:
+                logger.warning(f"[Dispatcher] 别名 '{alias_lower}' 已存在，将被覆盖")
+            self._aliases[alias_lower] = name
+            logger.debug(f"[Dispatcher] 注册别名: {alias_lower} -> {name}")
+    
+    def register_class(self, command_class: Type[BotCommand]) -> None:
+        """
+        注册命令类（自动实例化）
+        
+        Args:
+            command_class: 命令类
+        """
+        self.register(command_class())
+    
+    def unregister(self, name: str) -> bool:
+        """
+        注销命令
+        
+        Args:
+            name: 命令名称
+            
+        Returns:
+            是否成功注销
+        """
+        name = name.lower()
+        
+        if name not in self._commands:
+            return False
+        
+        command = self._commands.pop(name)
+        
+        # 移除别名
+        for alias in command.aliases:
+            self._aliases.pop(alias.lower(), None)
+        
+        logger.debug(f"[Dispatcher] 注销命令: {name}")
+        return True
+    
+    def get_command(self, name: str) -> Optional[BotCommand]:
+        """
+        获取命令
+        
+        支持命令名和别名查询。
+        
+        Args:
+            name: 命令名或别名
+            
+        Returns:
+            命令实例，或 None
+        """
+        name = name.lower()
+        
+        # 先查命令名
+        if name in self._commands:
+            return self._commands[name]
+        
+        # 再查别名
+        if name in self._aliases:
+            return self._commands.get(self._aliases[name])
+        
+        return None
+    
+    def list_commands(self, include_hidden: bool = False) -> List[BotCommand]:
+        """
+        列出所有命令
+        
+        Args:
+            include_hidden: 是否包含隐藏命令
+            
+        Returns:
+            命令列表
+        """
+        commands = list(self._commands.values())
+        
+        if not include_hidden:
+            commands = [c for c in commands if not c.hidden]
+        
+        return sorted(commands, key=lambda c: c.name)
+    
+    def is_admin(self, user_id: str) -> bool:
+        """检查用户是否是管理员"""
+        return user_id in self.admin_users
+    
+    def add_admin(self, user_id: str) -> None:
+        """添加管理员"""
+        self.admin_users.add(user_id)
+    
+    def remove_admin(self, user_id: str) -> None:
+        """移除管理员"""
+        self.admin_users.discard(user_id)
+    
+    def dispatch(self, message: BotMessage) -> BotResponse:
+        """
+        分发消息到对应命令
+        
+        Args:
+            message: 消息对象
+            
+        Returns:
+            响应对象
+        """
+        # 1. 检查频率限制
+        if not self._rate_limiter.is_allowed(message.user_id):
+            remaining_time = self._rate_limiter.window_seconds
+            return BotResponse.error_response(
+                f"请求过于频繁，请 {remaining_time} 秒后再试"
+            )
+        
+        # 2. 解析命令和参数
+        cmd_name, args = message.get_command_and_args(self.command_prefix)
+        
+        if cmd_name is None:
+            # 不是命令，检查是否 @了机器人
+            if message.mentioned:
+                return BotResponse.text_response(
+                    "你好！我是股票分析助手。\n"
+                    f"发送 `{self.command_prefix}help` 查看可用命令。"
+                )
+            # 非命令消息，不处理
+            return BotResponse.text_response("")
+        
+        logger.info(f"[Dispatcher] 收到命令: {cmd_name}, 参数: {args}, 用户: {message.user_name}")
+        
+        # 3. 查找命令处理器
+        command = self.get_command(cmd_name)
+        
+        if command is None:
+            return BotResponse.error_response(
+                f"未知命令: {cmd_name}\n"
+                f"发送 `{self.command_prefix}help` 查看可用命令。"
+            )
+        
+        # 4. 检查权限
+        if command.admin_only and not self.is_admin(message.user_id):
+            return BotResponse.error_response("此命令需要管理员权限")
+        
+        # 5. 验证参数
+        error_msg = command.validate_args(args)
+        if error_msg:
+            return BotResponse.error_response(
+                f"{error_msg}\n用法: `{command.usage}`"
+            )
+        
+        # 6. 执行命令
+        try:
+            response = command.execute(message, args)
+            logger.info(f"[Dispatcher] 命令 {cmd_name} 执行成功")
+            return response
+        except Exception as e:
+            logger.error(f"[Dispatcher] 命令 {cmd_name} 执行失败: {e}")
+            logger.exception(e)
+            return BotResponse.error_response(f"命令执行失败: {str(e)[:100]}")
+    
+    def set_help_command_getter(self, getter: Callable) -> None:
+        """
+        设置帮助命令的命令列表获取器
+        
+        用于让 HelpCommand 获取命令列表。
+        
+        Args:
+            getter: 回调函数，返回命令列表
+        """
+        self._help_command_getter = getter
+
+
+# 全局分发器实例
+_dispatcher: Optional[CommandDispatcher] = None
+
+
+def get_dispatcher() -> CommandDispatcher:
+    """
+    获取全局分发器实例
+    
+    使用单例模式，首次调用时自动初始化并注册所有命令。
+    """
+    global _dispatcher
+    
+    if _dispatcher is None:
+        from config import get_config
+        
+        config = get_config()
+        
+        # 创建分发器
+        _dispatcher = CommandDispatcher(
+            command_prefix=getattr(config, 'bot_command_prefix', '/'),
+            rate_limit_requests=getattr(config, 'bot_rate_limit_requests', 10),
+            rate_limit_window=getattr(config, 'bot_rate_limit_window', 60),
+            admin_users=getattr(config, 'bot_admin_users', []),
+        )
+        
+        # 自动注册所有命令
+        from bot.commands import ALL_COMMANDS
+        for command_class in ALL_COMMANDS:
+            _dispatcher.register_class(command_class)
+        
+        logger.info(f"[Dispatcher] 初始化完成，已注册 {len(_dispatcher._commands)} 个命令")
+    
+    return _dispatcher
+
+
+def reset_dispatcher() -> None:
+    """重置全局分发器（主要用于测试）"""
+    global _dispatcher
+    _dispatcher = None

bot/handler.py

@@ -0,0 +1,138 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+Bot Webhook 处理器
+===================================
+
+处理各平台的 Webhook 回调，分发到命令处理器。
+"""
+
+import json
+import logging
+from typing import Dict, Any, Optional, TYPE_CHECKING
+
+from bot.models import WebhookResponse
+from bot.dispatcher import get_dispatcher
+from bot.platforms import ALL_PLATFORMS
+
+if TYPE_CHECKING:
+    from bot.platforms.base import BotPlatform
+
+logger = logging.getLogger(__name__)
+
+# 平台实例缓存
+_platform_instances: Dict[str, 'BotPlatform'] = {}
+
+
+def get_platform(platform_name: str) -> Optional['BotPlatform']:
+    """
+    获取平台适配器实例
+    
+    使用缓存避免重复创建。
+    
+    Args:
+        platform_name: 平台名称
+        
+    Returns:
+        平台适配器实例，或 None
+    """
+    if platform_name not in _platform_instances:
+        platform_class = ALL_PLATFORMS.get(platform_name)
+        if platform_class:
+            _platform_instances[platform_name] = platform_class()
+        else:
+            logger.warning(f"[BotHandler] 未知平台: {platform_name}")
+            return None
+    
+    return _platform_instances[platform_name]
+
+
+def handle_webhook(
+    platform_name: str,
+    headers: Dict[str, str],
+    body: bytes,
+    query_params: Optional[Dict[str, list]] = None
+) -> WebhookResponse:
+    """
+    处理 Webhook 请求
+    
+    这是所有平台 Webhook 的统一入口。
+    
+    Args:
+        platform_name: 平台名称 (feishu, dingtalk, wecom, telegram)
+        headers: HTTP 请求头
+        body: 请求体原始字节
+        query_params: URL 查询参数（用于某些平台的验证）
+        
+    Returns:
+        WebhookResponse 响应对象
+    """
+    logger.info(f"[BotHandler] 收到 {platform_name} Webhook 请求")
+    
+    # 检查机器人功能是否启用
+    from config import get_config
+    config = get_config()
+    
+    if not getattr(config, 'bot_enabled', True):
+        logger.info("[BotHandler] 机器人功能未启用")
+        return WebhookResponse.success()
+    
+    # 获取平台适配器
+    platform = get_platform(platform_name)
+    if not platform:
+        return WebhookResponse.error(f"Unknown platform: {platform_name}", 400)
+    
+    # 解析 JSON 数据
+    try:
+        data = json.loads(body.decode('utf-8')) if body else {}
+    except json.JSONDecodeError as e:
+        logger.error(f"[BotHandler] JSON 解析失败: {e}")
+        return WebhookResponse.error("Invalid JSON", 400)
+    
+    logger.debug(f"[BotHandler] 请求数据: {json.dumps(data, ensure_ascii=False)[:500]}")
+    
+    # 处理 Webhook
+    message, challenge_response = platform.handle_webhook(headers, body, data)
+    
+    # 如果是验证请求，直接返回验证响应
+    if challenge_response:
+        logger.info(f"[BotHandler] 返回验证响应")
+        return challenge_response
+    
+    # 如果没有消息需要处理，返回空响应
+    if not message:
+        logger.debug("[BotHandler] 无需处理的消息")
+        return WebhookResponse.success()
+    
+    logger.info(f"[BotHandler] 解析到消息: user={message.user_name}, content={message.content[:50]}")
+    
+    # 分发到命令处理器
+    dispatcher = get_dispatcher()
+    response = dispatcher.dispatch(message)
+    
+    # 格式化响应
+    if response.text:
+        webhook_response = platform.format_response(response, message)
+        return webhook_response
+    
+    return WebhookResponse.success()
+
+
+def handle_feishu_webhook(headers: Dict[str, str], body: bytes) -> WebhookResponse:
+    """处理飞书 Webhook"""
+    return handle_webhook('feishu', headers, body)
+
+
+def handle_dingtalk_webhook(headers: Dict[str, str], body: bytes) -> WebhookResponse:
+    """处理钉钉 Webhook"""
+    return handle_webhook('dingtalk', headers, body)
+
+
+def handle_wecom_webhook(headers: Dict[str, str], body: bytes) -> WebhookResponse:
+    """处理企业微信 Webhook"""
+    return handle_webhook('wecom', headers, body)
+
+
+def handle_telegram_webhook(headers: Dict[str, str], body: bytes) -> WebhookResponse:
+    """处理 Telegram Webhook"""
+    return handle_webhook('telegram', headers, body)

bot/models.py

@@ -0,0 +1,179 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+机器人消息模型
+===================================
+
+定义统一的消息和响应模型，屏蔽各平台差异。
+"""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from typing import Dict, Any, Optional, List
+
+
+class ChatType(str, Enum):
+    """会话类型"""
+    GROUP = "group"      # 群聊
+    PRIVATE = "private"  # 私聊
+    UNKNOWN = "unknown"  # 未知
+
+
+class Platform(str, Enum):
+    """平台类型"""
+    FEISHU = "feishu"        # 飞书
+    DINGTALK = "dingtalk"    # 钉钉
+    WECOM = "wecom"          # 企业微信
+    TELEGRAM = "telegram"    # Telegram
+    UNKNOWN = "unknown"      # 未知
+
+
+@dataclass
+class BotMessage:
+    """
+    统一的机器人消息模型
+    
+    将各平台的消息格式统一为此模型，便于命令处理器处理。
+    
+    Attributes:
+        platform: 平台标识
+        message_id: 消息 ID（平台原始 ID）
+        user_id: 发送者 ID
+        user_name: 发送者名称
+        chat_id: 会话 ID（群聊 ID 或私聊 ID）
+        chat_type: 会话类型
+        content: 消息文本内容（已去除 @机器人 部分）
+        raw_content: 原始消息内容
+        mentioned: 是否 @了机器人
+        mentions: @的用户列表
+        timestamp: 消息时间戳
+        raw_data: 原始请求数据（平台特定，用于调试）
+    """
+    platform: str
+    message_id: str
+    user_id: str
+    user_name: str
+    chat_id: str
+    chat_type: ChatType
+    content: str
+    raw_content: str = ""
+    mentioned: bool = False
+    mentions: List[str] = field(default_factory=list)
+    timestamp: datetime = field(default_factory=datetime.now)
+    raw_data: Dict[str, Any] = field(default_factory=dict)
+    
+    def get_command_and_args(self, prefix: str = "/") -> tuple:
+        """
+        解析命令和参数
+        
+        Args:
+            prefix: 命令前缀，默认 "/"
+            
+        Returns:
+            (command, args) 元组，如 ("analyze", ["600519"])
+            如果不是命令，返回 (None, [])
+        """
+        text = self.content.strip()
+        
+        # 检查是否以命令前缀开头
+        if not text.startswith(prefix):
+            # 尝试匹配中文命令（无前缀）
+            chinese_commands = {
+                '分析': 'analyze',
+                '大盘': 'market',
+                '批量': 'batch',
+                '帮助': 'help',
+                '状态': 'status',
+            }
+            for cn_cmd, en_cmd in chinese_commands.items():
+                if text.startswith(cn_cmd):
+                    args = text[len(cn_cmd):].strip().split()
+                    return en_cmd, args
+            return None, []
+        
+        # 去除前缀
+        text = text[len(prefix):]
+        
+        # 分割命令和参数
+        parts = text.split()
+        if not parts:
+            return None, []
+        
+        command = parts[0].lower()
+        args = parts[1:] if len(parts) > 1 else []
+        
+        return command, args
+    
+    def is_command(self, prefix: str = "/") -> bool:
+        """检查消息是否是命令"""
+        cmd, _ = self.get_command_and_args(prefix)
+        return cmd is not None
+
+
+@dataclass
+class BotResponse:
+    """
+    统一的机器人响应模型
+    
+    命令处理器返回此模型，由平台适配器转换为平台特定格式。
+    
+    Attributes:
+        text: 回复文本
+        markdown: 是否为 Markdown 格式
+        at_user: 是否 @发送者
+        reply_to_message: 是否回复原消息
+        extra: 额外数据（平台特定）
+    """
+    text: str
+    markdown: bool = False
+    at_user: bool = True
+    reply_to_message: bool = True
+    extra: Dict[str, Any] = field(default_factory=dict)
+    
+    @classmethod
+    def text_response(cls, text: str, at_user: bool = True) -> 'BotResponse':
+        """创建纯文本响应"""
+        return cls(text=text, markdown=False, at_user=at_user)
+    
+    @classmethod
+    def markdown_response(cls, text: str, at_user: bool = True) -> 'BotResponse':
+        """创建 Markdown 响应"""
+        return cls(text=text, markdown=True, at_user=at_user)
+    
+    @classmethod
+    def error_response(cls, message: str) -> 'BotResponse':
+        """创建错误响应"""
+        return cls(text=f"❌ 错误：{message}", markdown=False, at_user=True)
+
+
+@dataclass
+class WebhookResponse:
+    """
+    Webhook 响应模型
+    
+    平台适配器返回此模型，包含 HTTP 响应内容。
+    
+    Attributes:
+        status_code: HTTP 状态码
+        body: 响应体（字典，将被 JSON 序列化）
+        headers: 额外的响应头
+    """
+    status_code: int = 200
+    body: Dict[str, Any] = field(default_factory=dict)
+    headers: Dict[str, str] = field(default_factory=dict)
+    
+    @classmethod
+    def success(cls, body: Optional[Dict] = None) -> 'WebhookResponse':
+        """创建成功响应"""
+        return cls(status_code=200, body=body or {})
+    
+    @classmethod
+    def challenge(cls, challenge: str) -> 'WebhookResponse':
+        """创建验证响应（用于平台 URL 验证）"""
+        return cls(status_code=200, body={"challenge": challenge})
+    
+    @classmethod
+    def error(cls, message: str, status_code: int = 400) -> 'WebhookResponse':
+        """创建错误响应"""
+        return cls(status_code=status_code, body={"error": message})

bot/platforms/__init__.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+平台适配器模块
+===================================
+
+包含各平台的 Webhook 处理和消息解析逻辑。
+
+支持两种接入模式：
+1. Webhook 模式：需要公网 IP，配置回调 URL
+2. Stream 模式：无需公网 IP，通过 WebSocket 长连接（钉钉支持）
+"""
+
+from bot.platforms.base import BotPlatform
+from bot.platforms.feishu import FeishuPlatform
+from bot.platforms.dingtalk import DingtalkPlatform
+from bot.platforms.wecom import WecomPlatform
+from bot.platforms.telegram import TelegramPlatform
+
+# 所有可用平台（Webhook 模式）
+ALL_PLATFORMS = {
+    'feishu': FeishuPlatform,
+    'dingtalk': DingtalkPlatform,
+    'wecom': WecomPlatform,
+    'telegram': TelegramPlatform,
+}
+
+# 钉钉 Stream 模式（可选）
+try:
+    from bot.platforms.dingtalk_stream import (
+        DingtalkStreamClient,
+        DingtalkStreamHandler,
+        get_dingtalk_stream_client,
+        start_dingtalk_stream_background,
+        DINGTALK_STREAM_AVAILABLE,
+    )
+except ImportError:
+    DINGTALK_STREAM_AVAILABLE = False
+    DingtalkStreamClient = None
+    DingtalkStreamHandler = None
+    get_dingtalk_stream_client = lambda: None
+    start_dingtalk_stream_background = lambda: False
+
+__all__ = [
+    'BotPlatform',
+    'FeishuPlatform',
+    'DingtalkPlatform',
+    'WecomPlatform',
+    'TelegramPlatform',
+    'ALL_PLATFORMS',
+    # Stream 模式
+    'DingtalkStreamClient',
+    'DingtalkStreamHandler',
+    'get_dingtalk_stream_client',
+    'start_dingtalk_stream_background',
+    'DINGTALK_STREAM_AVAILABLE',
+]

bot/platforms/base.py

@@ -0,0 +1,153 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+平台适配器基类
+===================================
+
+定义平台适配器的抽象基类，各平台必须继承此类。
+"""
+
+from abc import ABC, abstractmethod
+from typing import Dict, Any, Optional, Tuple
+
+from bot.models import BotMessage, BotResponse, WebhookResponse
+
+
+class BotPlatform(ABC):
+    """
+    平台适配器抽象基类
+    
+    负责：
+    1. 验证 Webhook 请求签名
+    2. 解析平台消息为统一格式
+    3. 将响应转换为平台格式
+    
+    使用示例：
+        class MyPlatform(BotPlatform):
+            @property
+            def platform_name(self) -> str:
+                return "myplatform"
+            
+            def verify_request(self, headers, body) -> bool:
+                # 验证签名逻辑
+                return True
+            
+            def parse_message(self, data) -> Optional[BotMessage]:
+                # 解析消息逻辑
+                return BotMessage(...)
+            
+            def format_response(self, response, message) -> WebhookResponse:
+                # 格式化响应逻辑
+                return WebhookResponse.success({"text": response.text})
+    """
+    
+    @property
+    @abstractmethod
+    def platform_name(self) -> str:
+        """
+        平台标识名称
+        
+        用于路由匹配和日志标识，如 "feishu", "dingtalk"
+        """
+        pass
+    
+    @abstractmethod
+    def verify_request(self, headers: Dict[str, str], body: bytes) -> bool:
+        """
+        验证请求签名
+        
+        各平台有不同的签名验证机制，需要单独实现。
+        
+        Args:
+            headers: HTTP 请求头
+            body: 请求体原始字节
+            
+        Returns:
+            签名是否有效
+        """
+        pass
+    
+    @abstractmethod
+    def parse_message(self, data: Dict[str, Any]) -> Optional[BotMessage]:
+        """
+        解析平台消息为统一格式
+        
+        将平台特定的消息格式转换为 BotMessage。
+        如果不是需要处理的消息类型（如事件回调），返回 None。
+        
+        Args:
+            data: 解析后的 JSON 数据
+            
+        Returns:
+            BotMessage 对象，或 None（不需要处理）
+        """
+        pass
+    
+    @abstractmethod
+    def format_response(
+        self, 
+        response: BotResponse, 
+        message: BotMessage
+    ) -> WebhookResponse:
+        """
+        将统一响应转换为平台格式
+        
+        Args:
+            response: 统一响应对象
+            message: 原始消息对象（用于获取回复目标等信息）
+            
+        Returns:
+            WebhookResponse 对象
+        """
+        pass
+    
+    def handle_challenge(self, data: Dict[str, Any]) -> Optional[WebhookResponse]:
+        """
+        处理平台验证请求
+        
+        部分平台在配置 Webhook 时会发送验证请求，需要返回特定响应。
+        子类可重写此方法。
+        
+        Args:
+            data: 请求数据
+            
+        Returns:
+            验证响应，或 None（不是验证请求）
+        """
+        return None
+    
+    def handle_webhook(
+        self, 
+        headers: Dict[str, str], 
+        body: bytes,
+        data: Dict[str, Any]
+    ) -> Tuple[Optional[BotMessage], Optional[WebhookResponse]]:
+        """
+        处理 Webhook 请求
+        
+        这是主入口方法，协调验证、解析等流程。
+        
+        Args:
+            headers: HTTP 请求头
+            body: 请求体原始字节
+            data: 解析后的 JSON 数据
+            
+        Returns:
+            (BotMessage, WebhookResponse) 元组
+            - 如果是验证请求：(None, challenge_response)
+            - 如果是普通消息：(message, None) - 响应将在命令处理后生成
+            - 如果验证失败或无需处理：(None, error_response 或 None)
+        """
+        # 1. 检查是否是验证请求
+        challenge_response = self.handle_challenge(data)
+        if challenge_response:
+            return None, challenge_response
+        
+        # 2. 验证请求签名
+        if not self.verify_request(headers, body):
+            return None, WebhookResponse.error("Invalid signature", 403)
+        
+        # 3. 解析消息
+        message = self.parse_message(data)
+        
+        return message, None

bot/platforms/dingtalk.py

@@ -0,0 +1,315 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+钉钉平台适配器
+===================================
+
+处理钉钉机器人的 Webhook 回调。
+
+钉钉机器人文档：
+https://open.dingtalk.com/document/robots/robot-overview
+"""
+
+import hashlib
+import hmac
+import base64
+import time
+import logging
+import json
+from datetime import datetime
+from typing import Dict, Any, Optional
+from urllib.parse import quote_plus
+
+from bot.platforms.base import BotPlatform
+from bot.models import BotMessage, BotResponse, WebhookResponse, ChatType
+
+logger = logging.getLogger(__name__)
+
+
+class DingtalkPlatform(BotPlatform):
+    """
+    钉钉平台适配器
+    
+    支持：
+    - 企业内部机器人回调
+    - 群机器人 Outgoing 回调
+    - 消息签名验证
+    
+    配置要求：
+    - DINGTALK_APP_KEY: 应用 AppKey
+    - DINGTALK_APP_SECRET: 应用 AppSecret（用于签名验证）
+    """
+    
+    def __init__(self):
+        from config import get_config
+        config = get_config()
+        
+        self._app_key = getattr(config, 'dingtalk_app_key', None)
+        self._app_secret = getattr(config, 'dingtalk_app_secret', None)
+    
+    @property
+    def platform_name(self) -> str:
+        return "dingtalk"
+    
+    def verify_request(self, headers: Dict[str, str], body: bytes) -> bool:
+        """
+        验证钉钉请求签名
+        
+        钉钉签名算法：
+        1. 获取 timestamp 和 sign
+        2. 计算：base64(hmac_sha256(timestamp + "\n" + app_secret))
+        3. 比对签名
+        """
+        if not self._app_secret:
+            logger.warning("[DingTalk] 未配置 app_secret，跳过签名验证")
+            return True
+        
+        timestamp = headers.get('timestamp', '')
+        sign = headers.get('sign', '')
+        
+        if not timestamp or not sign:
+            logger.warning("[DingTalk] 缺少签名参数")
+            return True  # 可能是不需要签名的请求
+        
+        # 验证时间戳（1小时内有效）
+        try:
+            request_time = int(timestamp)
+            current_time = int(time.time() * 1000)
+            if abs(current_time - request_time) > 3600 * 1000:
+                logger.warning("[DingTalk] 时间戳过期")
+                return False
+        except ValueError:
+            logger.warning("[DingTalk] 无效的时间戳")
+            return False
+        
+        # 计算签名
+        string_to_sign = f"{timestamp}\n{self._app_secret}"
+        hmac_code = hmac.new(
+            self._app_secret.encode('utf-8'),
+            string_to_sign.encode('utf-8'),
+            digestmod=hashlib.sha256
+        ).digest()
+        expected_sign = base64.b64encode(hmac_code).decode('utf-8')
+        
+        if sign != expected_sign:
+            logger.warning(f"[DingTalk] 签名验证失败")
+            return False
+        
+        return True
+    
+    def handle_challenge(self, data: Dict[str, Any]) -> Optional[WebhookResponse]:
+        """钉钉不需要 URL 验证"""
+        return None
+    
+    def parse_message(self, data: Dict[str, Any]) -> Optional[BotMessage]:
+        """
+        解析钉钉消息
+        
+        钉钉 Outgoing 机器人消息格式：
+        {
+            "msgtype": "text",
+            "text": {
+                "content": "@机器人 /analyze 600519"
+            },
+            "msgId": "xxx",
+            "createAt": "1234567890",
+            "conversationType": "2",  # 1=单聊, 2=群聊
+            "conversationId": "xxx",
+            "conversationTitle": "群名",
+            "senderId": "xxx",
+            "senderNick": "用户昵称",
+            "senderCorpId": "xxx",
+            "senderStaffId": "xxx",
+            "chatbotUserId": "xxx",
+            "atUsers": [{"dingtalkId": "xxx", "staffId": "xxx"}],
+            "isAdmin": false,
+            "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=xxx",
+            "sessionWebhookExpiredTime": 1234567890
+        }
+        """
+        # 检查消息类型
+        msg_type = data.get('msgtype', '')
+        if msg_type != 'text':
+            logger.debug(f"[DingTalk] 忽略非文本消息: {msg_type}")
+            return None
+        
+        # 获取消息内容
+        text_content = data.get('text', {})
+        raw_content = text_content.get('content', '')
+        
+        # 提取命令（去除 @机器人）
+        content = self._extract_command(raw_content)
+        
+        # 检查是否 @了机器人
+        at_users = data.get('atUsers', [])
+        mentioned = len(at_users) > 0
+        
+        # 会话类型
+        conversation_type = data.get('conversationType', '')
+        if conversation_type == '1':
+            chat_type = ChatType.PRIVATE
+        elif conversation_type == '2':
+            chat_type = ChatType.GROUP
+        else:
+            chat_type = ChatType.UNKNOWN
+        
+        # 创建时间
+        create_at = data.get('createAt', '')
+        try:
+            timestamp = datetime.fromtimestamp(int(create_at) / 1000)
+        except (ValueError, TypeError):
+            timestamp = datetime.now()
+        
+        # 保存 session webhook 用于回复
+        session_webhook = data.get('sessionWebhook', '')
+        
+        return BotMessage(
+            platform=self.platform_name,
+            message_id=data.get('msgId', ''),
+            user_id=data.get('senderId', ''),
+            user_name=data.get('senderNick', ''),
+            chat_id=data.get('conversationId', ''),
+            chat_type=chat_type,
+            content=content,
+            raw_content=raw_content,
+            mentioned=mentioned,
+            mentions=[u.get('dingtalkId', '') for u in at_users],
+            timestamp=timestamp,
+            raw_data={
+                **data,
+                '_session_webhook': session_webhook,
+            },
+        )
+    
+    def _extract_command(self, text: str) -> str:
+        """
+        提取命令内容（去除 @机器人）
+        
+        钉钉的 @用户 格式通常是 @昵称 后跟空格
+        """
+        # 简单处理：移除开头的 @xxx 部分
+        import re
+        # 匹配开头的 @xxx（中英文都可能）
+        text = re.sub(r'^@[\S]+\s*', '', text.strip())
+        return text.strip()
+    
+    def format_response(
+        self, 
+        response: BotResponse, 
+        message: BotMessage
+    ) -> WebhookResponse:
+        """
+        格式化钉钉响应
+        
+        钉钉 Outgoing 机器人可以直接在响应中返回消息。
+        也可以使用 sessionWebhook 异步发送。
+        
+        响应格式：
+        {
+            "msgtype": "text" | "markdown",
+            "text": {"content": "xxx"},
+            "markdown": {"title": "xxx", "text": "xxx"},
+            "at": {"atUserIds": ["xxx"], "isAtAll": false}
+        }
+        """
+        if not response.text:
+            return WebhookResponse.success()
+        
+        # 构建响应
+        if response.markdown:
+            body = {
+                "msgtype": "markdown",
+                "markdown": {
+                    "title": "股票分析助手",
+                    "text": response.text,
+                }
+            }
+        else:
+            body = {
+                "msgtype": "text",
+                "text": {
+                    "content": response.text,
+                }
+            }
+        
+        # @发送者
+        if response.at_user and message.user_id:
+            body["at"] = {
+                "atUserIds": [message.user_id],
+                "isAtAll": False,
+            }
+        
+        return WebhookResponse.success(body)
+    
+    def send_by_session_webhook(
+        self, 
+        session_webhook: str, 
+        response: BotResponse,
+        message: BotMessage
+    ) -> bool:
+        """
+        通过 sessionWebhook 发送消息
+        
+        适用于需要异步发送或多条消息的场景。
+        
+        Args:
+            session_webhook: 钉钉提供的会话 Webhook URL
+            response: 响应对象
+            message: 原始消息对象
+            
+        Returns:
+            是否发送成功
+        """
+        if not session_webhook:
+            logger.warning("[DingTalk] 没有可用的 sessionWebhook")
+            return False
+        
+        import requests
+        
+        try:
+            # 构建消息
+            if response.markdown:
+                payload = {
+                    "msgtype": "markdown",
+                    "markdown": {
+                        "title": "股票分析助手",
+                        "text": response.text,
+                    }
+                }
+            else:
+                payload = {
+                    "msgtype": "text",
+                    "text": {
+                        "content": response.text,
+                    }
+                }
+            
+            # @发送者
+            if response.at_user and message.user_id:
+                payload["at"] = {
+                    "atUserIds": [message.user_id],
+                    "isAtAll": False,
+                }
+            
+            # 发送请求
+            resp = requests.post(
+                session_webhook,
+                json=payload,
+                timeout=10
+            )
+            
+            if resp.status_code == 200:
+                result = resp.json()
+                if result.get('errcode') == 0:
+                    logger.info("[DingTalk] sessionWebhook 发送成功")
+                    return True
+                else:
+                    logger.error(f"[DingTalk] sessionWebhook 发送失败: {result}")
+                    return False
+            else:
+                logger.error(f"[DingTalk] sessionWebhook 请求失败: {resp.status_code}")
+                return False
+                
+        except Exception as e:
+            logger.error(f"[DingTalk] sessionWebhook 发送异常: {e}")
+            return False

bot/platforms/dingtalk_stream.py

@@ -0,0 +1,346 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+钉钉 Stream 模式适配器
+===================================
+
+使用钉钉官方 Stream SDK 接入机器人，无需公网 IP 和 Webhook 配置。
+
+优势：
+- 不需要公网 IP 或域名
+- 不需要配置 Webhook URL
+- 通过 WebSocket 长连接接收消息
+- 更简单的接入方式
+
+依赖：
+pip install dingtalk-stream
+
+钉钉 Stream SDK：
+https://github.com/open-dingtalk/dingtalk-stream-sdk-python
+"""
+
+import logging
+import asyncio
+import threading
+from datetime import datetime
+from typing import Optional, Callable, Any
+
+logger = logging.getLogger(__name__)
+
+# 尝试导入钉钉 Stream SDK
+try:
+    import dingtalk_stream
+    from dingtalk_stream import AckMessage
+
+    DINGTALK_STREAM_AVAILABLE = True
+except ImportError:
+    DINGTALK_STREAM_AVAILABLE = False
+    logger.warning("[DingTalk Stream] dingtalk-stream SDK 未安装，Stream 模式不可用")
+    logger.warning("[DingTalk Stream] 请运行: pip install dingtalk-stream")
+
+from bot.models import BotMessage, BotResponse, ChatType
+
+
+class DingtalkStreamHandler:
+    """
+    钉钉 Stream 模式消息处理器
+    
+    将 Stream SDK 的回调转换为统一的 BotMessage 格式，
+    并调用命令分发器处理。
+    """
+
+    def __init__(self, on_message: Callable[[BotMessage], BotResponse]):
+        """
+        Args:
+            on_message: 消息处理回调函数，接收 BotMessage 返回 BotResponse
+        """
+        self._on_message = on_message
+        self._logger = logger
+
+    @staticmethod
+    def _truncate_log_content(text: str, max_len: int = 200) -> str:
+        cleaned = text.replace("\n", " ").strip()
+        if len(cleaned) > max_len:
+            return f"{cleaned[:max_len]}..."
+        return cleaned
+
+    def _log_incoming_message(self, message: BotMessage) -> None:
+        content = message.raw_content or message.content or ""
+        summary = self._truncate_log_content(content)
+        self._logger.info(
+            "[DingTalk Stream] Incoming message: msg_id=%s user_id=%s chat_id=%s chat_type=%s content=%s",
+            message.message_id,
+            message.user_id,
+            message.chat_id,
+            getattr(message.chat_type, "value", message.chat_type),
+            summary,
+        )
+
+    if DINGTALK_STREAM_AVAILABLE:
+        class _ChatbotHandler(dingtalk_stream.ChatbotHandler):
+            """内部消息处理器"""
+
+            def __init__(self, parent: 'DingtalkStreamHandler'):
+                super().__init__()
+                self._parent = parent
+                self.logger = logger
+
+            async def process(self, callback: dingtalk_stream.CallbackMessage):
+                """处理收到的消息"""
+                try:
+                    # 解析消息
+                    incoming = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
+
+                    # 转换为统一格式
+                    bot_message = self._parent._parse_stream_message(incoming, callback.data)
+
+                    if bot_message:
+                        self._parent._log_incoming_message(bot_message)
+                        # 调用消息处理回调
+                        response = self._parent._on_message(bot_message)
+
+                        # 发送回复
+                        if response and response.text:
+                            # 构建 @用户 前缀（群聊场景下需要在文本中包含 @用户名）
+                            if response.at_user and incoming.sender_nick:
+                                if response.markdown:
+                                    self.reply_markdown(
+                                        title="股票分析助手",
+                                        text=f"@{incoming.sender_nick} " + response.text,
+                                        incoming_message=incoming
+                                    )
+                                else:
+                                    self.reply_text(response.text, incoming)
+
+                    return AckMessage.STATUS_OK, 'OK'
+
+                except Exception as e:
+                    self.logger.error(f"[DingTalk Stream] 处理消息失败: {e}")
+                    self.logger.exception(e)
+                    return AckMessage.STATUS_SYSTEM_EXCEPTION, str(e)
+
+        def create_handler(self) -> '_ChatbotHandler':
+            """创建 SDK 需要的处理器实例"""
+            return self._ChatbotHandler(self)
+
+    def _parse_stream_message(self, incoming: Any, raw_data: dict) -> Optional[BotMessage]:
+        """
+        解析 Stream 消息为统一格式
+        
+        Args:
+            incoming: ChatbotMessage 对象
+            raw_data: 原始回调数据
+        """
+        try:
+            raw_data = dict(raw_data or {})
+
+            # 获取消息内容
+            raw_content = incoming.text.content if incoming.text else ''
+
+            # 提取命令（去除 @机器人）
+            content = self._extract_command(raw_content)
+
+            # 会话类型
+            conversation_type = getattr(incoming, 'conversation_type', None)
+            if conversation_type == '1':
+                chat_type = ChatType.PRIVATE
+            elif conversation_type == '2':
+                chat_type = ChatType.GROUP
+            else:
+                chat_type = ChatType.UNKNOWN
+
+            # 是否 @了机器人（Stream 模式下收到的消息一般都是 @机器人的）
+            mentioned = True
+
+            # 提取 sessionWebhook，便于异步推送
+            session_webhook = (
+                    getattr(incoming, 'session_webhook', None)
+                    or raw_data.get('sessionWebhook')
+                    or raw_data.get('session_webhook')
+            )
+            if session_webhook:
+                raw_data['_session_webhook'] = session_webhook
+
+            return BotMessage(
+                platform='dingtalk',
+                message_id=getattr(incoming, 'msg_id', '') or '',
+                user_id=getattr(incoming, 'sender_id', '') or '',
+                user_name=getattr(incoming, 'sender_nick', '') or '',
+                chat_id=getattr(incoming, 'conversation_id', '') or '',
+                chat_type=chat_type,
+                content=content,
+                raw_content=raw_content,
+                mentioned=mentioned,
+                mentions=[],
+                timestamp=datetime.now(),
+                raw_data=raw_data,
+            )
+
+        except Exception as e:
+            logger.error(f"[DingTalk Stream] 解析消息失败: {e}")
+            return None
+
+    def _extract_command(self, text: str) -> str:
+        """提取命令内容（去除 @机器人）"""
+        import re
+        text = re.sub(r'^@[\S]+\s*', '', text.strip())
+        return text.strip()
+
+
+class DingtalkStreamClient:
+    """
+    钉钉 Stream 模式客户端
+    
+    封装 dingtalk-stream SDK，提供简单的启动接口。
+    
+    使用方式：
+        client = DingtalkStreamClient()
+        client.start()  # 阻塞运行
+        
+        # 或者在后台运行
+        client.start_background()
+    """
+
+    def __init__(
+            self,
+            client_id: Optional[str] = None,
+            client_secret: Optional[str] = None
+    ):
+        """
+        Args:
+            client_id: 应用 AppKey（不传则从配置读取）
+            client_secret: 应用 AppSecret（不传则从配置读取）
+        """
+        if not DINGTALK_STREAM_AVAILABLE:
+            raise ImportError(
+                "dingtalk-stream SDK 未安装。\n"
+                "请运行: pip install dingtalk-stream"
+            )
+
+        from config import get_config
+        config = get_config()
+
+        self._client_id = client_id or getattr(config, 'dingtalk_app_key', None)
+        self._client_secret = client_secret or getattr(config, 'dingtalk_app_secret', None)
+
+        if not self._client_id or not self._client_secret:
+            raise ValueError(
+                "钉钉 Stream 模式需要配置 DINGTALK_APP_KEY 和 DINGTALK_APP_SECRET"
+            )
+
+        self._client: Optional[dingtalk_stream.DingTalkStreamClient] = None
+        self._background_thread: Optional[threading.Thread] = None
+        self._running = False
+
+    def _create_message_handler(self) -> Callable[[BotMessage], BotResponse]:
+        """创建消息处理函数"""
+
+        def handle_message(message: BotMessage) -> BotResponse:
+            from bot.dispatcher import get_dispatcher
+            dispatcher = get_dispatcher()
+            return dispatcher.dispatch(message)
+
+        return handle_message
+
+    def start(self) -> None:
+        """
+        启动 Stream 客户端（阻塞）
+        
+        此方法会阻塞当前线程，直到客户端停止。
+        """
+        logger.info("[DingTalk Stream] 正在启动...")
+
+        # 创建凭证
+        credential = dingtalk_stream.Credential(
+            self._client_id,
+            self._client_secret
+        )
+
+        # 创建客户端
+        self._client = dingtalk_stream.DingTalkStreamClient(credential)
+
+        # 注册消息处理器
+        handler = DingtalkStreamHandler(self._create_message_handler())
+        self._client.register_callback_handler(
+            dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
+            handler.create_handler()
+        )
+
+        self._running = True
+        logger.info("[DingTalk Stream] 客户端已启动，等待消息...")
+
+        # 启动（阻塞）
+        self._client.start_forever()
+
+    def start_background(self) -> None:
+        """
+        在后台线程启动 Stream 客户端（非阻塞）
+        
+        适用于与其他服务（如 WebUI）同时运行的场景。
+        """
+        if self._background_thread and self._background_thread.is_alive():
+            logger.warning("[DingTalk Stream] 客户端已在运行")
+            return
+
+        self._running = True
+        self._background_thread = threading.Thread(
+            target=self._run_in_background,
+            daemon=True,
+            name="DingtalkStreamClient"
+        )
+        self._background_thread.start()
+        logger.info("[DingTalk Stream] 后台客户端已启动")
+
+    def _run_in_background(self) -> None:
+        """后台运行（处理异常和重连）"""
+        while self._running:
+            try:
+                self.start()
+            except Exception as e:
+                logger.error(f"[DingTalk Stream] 运行异常: {e}")
+                if self._running:
+                    logger.info("[DingTalk Stream] 5 秒后重连...")
+                    import time
+                    time.sleep(5)
+
+    def stop(self) -> None:
+        """停止客户端"""
+        self._running = False
+        logger.info("[DingTalk Stream] 客户端已停止")
+
+    @property
+    def is_running(self) -> bool:
+        """是否正在运行"""
+        return self._running
+
+
+# 全局客户端实例
+_stream_client: Optional[DingtalkStreamClient] = None
+
+
+def get_dingtalk_stream_client() -> Optional[DingtalkStreamClient]:
+    """获取全局 Stream 客户端实例"""
+    global _stream_client
+
+    if _stream_client is None and DINGTALK_STREAM_AVAILABLE:
+        try:
+            _stream_client = DingtalkStreamClient()
+        except (ImportError, ValueError) as e:
+            logger.warning(f"[DingTalk Stream] 无法创建客户端: {e}")
+            return None
+
+    return _stream_client
+
+
+def start_dingtalk_stream_background() -> bool:
+    """
+    在后台启动钉钉 Stream 客户端
+    
+    Returns:
+        是否成功启动
+    """
+    client = get_dingtalk_stream_client()
+    if client:
+        client.start_background()
+        return True
+    return False

bot/platforms/feishu.py

@@ -0,0 +1,278 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+飞书平台适配器
+===================================
+
+处理飞书机器人的 Webhook 回调。
+
+飞书机器人文档：
+https://open.feishu.cn/document/ukTMukTMukTM/ucTM5YjL3ETO24yNxkjN
+"""
+
+import hashlib
+import hmac
+import base64
+import json
+import logging
+import time
+from datetime import datetime
+from typing import Dict, Any, Optional
+
+from bot.platforms.base import BotPlatform
+from bot.models import BotMessage, BotResponse, WebhookResponse, ChatType
+
+logger = logging.getLogger(__name__)
+
+
+class FeishuPlatform(BotPlatform):
+    """
+    飞书平台适配器
+    
+    支持：
+    - 事件订阅回调（机器人收到消息）
+    - URL 验证（配置回调地址时的验证请求）
+    - 消息签名验证
+    
+    配置要求：
+    - FEISHU_APP_ID: 应用 ID
+    - FEISHU_APP_SECRET: 应用密钥
+    - FEISHU_VERIFICATION_TOKEN: 事件订阅验证 Token
+    - FEISHU_ENCRYPT_KEY: 加密密钥（可选）
+    """
+    
+    def __init__(self):
+        from config import get_config
+        config = get_config()
+        
+        self._app_id = getattr(config, 'feishu_app_id', None)
+        self._app_secret = getattr(config, 'feishu_app_secret', None)
+        self._verification_token = getattr(config, 'feishu_verification_token', None)
+        self._encrypt_key = getattr(config, 'feishu_encrypt_key', None)
+    
+    @property
+    def platform_name(self) -> str:
+        return "feishu"
+    
+    def verify_request(self, headers: Dict[str, str], body: bytes) -> bool:
+        """
+        验证飞书请求签名
+        
+        飞书使用 X-Lark-Signature 头进行签名验证。
+        签名算法：sha256(timestamp + nonce + encrypt_key + body)
+        """
+        if not self._verification_token:
+            # 未配置验证 Token，跳过验证（开发环境）
+            logger.warning("[Feishu] 未配置 verification_token，跳过签名验证")
+            return True
+        
+        # 获取签名相关头
+        timestamp = headers.get('X-Lark-Request-Timestamp', '')
+        nonce = headers.get('X-Lark-Request-Nonce', '')
+        signature = headers.get('X-Lark-Signature', '')
+        
+        if not signature:
+            # 没有签名头，可能是旧版本或验证请求
+            return True
+        
+        # 计算签名
+        if self._encrypt_key:
+            sign_string = f"{timestamp}{nonce}{self._encrypt_key}{body.decode('utf-8')}"
+        else:
+            sign_string = f"{timestamp}{nonce}{body.decode('utf-8')}"
+        
+        expected_signature = hashlib.sha256(sign_string.encode('utf-8')).hexdigest()
+        
+        if signature != expected_signature:
+            logger.warning(f"[Feishu] 签名验证失败: expected={expected_signature}, got={signature}")
+            return False
+        
+        return True
+    
+    def handle_challenge(self, data: Dict[str, Any]) -> Optional[WebhookResponse]:
+        """
+        处理飞书 URL 验证请求
+        
+        配置事件订阅时，飞书会发送验证请求：
+        {
+            "challenge": "xxx",
+            "token": "xxx",
+            "type": "url_verification"
+        }
+        
+        需要返回：
+        {"challenge": "xxx"}
+        """
+        if data.get('type') == 'url_verification':
+            challenge = data.get('challenge', '')
+            token = data.get('token', '')
+            
+            # 验证 token
+            if self._verification_token and token != self._verification_token:
+                logger.warning(f"[Feishu] 验证 token 不匹配")
+                return WebhookResponse.error("Invalid token", 403)
+            
+            logger.info(f"[Feishu] URL 验证成功")
+            return WebhookResponse.challenge(challenge)
+        
+        return None
+    
+    def parse_message(self, data: Dict[str, Any]) -> Optional[BotMessage]:
+        """
+        解析飞书消息
+        
+        飞书事件格式（v2.0）：
+        {
+            "schema": "2.0",
+            "header": {
+                "event_id": "xxx",
+                "event_type": "im.message.receive_v1",
+                "create_time": "1234567890",
+                "token": "xxx",
+                "app_id": "xxx"
+            },
+            "event": {
+                "sender": {
+                    "sender_id": {"open_id": "xxx", "user_id": "xxx"},
+                    "sender_type": "user"
+                },
+                "message": {
+                    "message_id": "xxx",
+                    "chat_id": "xxx",
+                    "chat_type": "group",
+                    "message_type": "text",
+                    "content": "{\"text\":\"@机器人 /analyze 600519\"}"
+                }
+            }
+        }
+        """
+        # 检查事件类型
+        header = data.get('header', {})
+        event_type = header.get('event_type', '')
+        
+        if event_type != 'im.message.receive_v1':
+            logger.debug(f"[Feishu] 忽略非消息事件: {event_type}")
+            return None
+        
+        event = data.get('event', {})
+        message_data = event.get('message', {})
+        sender_data = event.get('sender', {})
+        
+        # 只处理文本消息
+        message_type = message_data.get('message_type', '')
+        if message_type != 'text':
+            logger.debug(f"[Feishu] 忽略非文本消息: {message_type}")
+            return None
+        
+        # 解析消息内容
+        content_str = message_data.get('content', '{}')
+        try:
+            content_json = json.loads(content_str)
+            raw_content = content_json.get('text', '')
+        except json.JSONDecodeError:
+            raw_content = content_str
+        
+        # 提取 @机器人 后的内容
+        content = self._extract_command(raw_content, event)
+        mentioned = '@' in raw_content or bool(message_data.get('mentions'))
+        
+        # 获取发送者信息
+        sender_id = sender_data.get('sender_id', {})
+        user_id = sender_id.get('open_id', '') or sender_id.get('user_id', '')
+        
+        # 获取会话类型
+        chat_type_str = message_data.get('chat_type', '')
+        if chat_type_str == 'group':
+            chat_type = ChatType.GROUP
+        elif chat_type_str == 'p2p':
+            chat_type = ChatType.PRIVATE
+        else:
+            chat_type = ChatType.UNKNOWN
+        
+        # 创建时间
+        create_time = header.get('create_time', '')
+        try:
+            timestamp = datetime.fromtimestamp(int(create_time) / 1000)
+        except (ValueError, TypeError):
+            timestamp = datetime.now()
+        
+        return BotMessage(
+            platform=self.platform_name,
+            message_id=message_data.get('message_id', ''),
+            user_id=user_id,
+            user_name=sender_id.get('user_id', user_id),  # 飞书不直接返回用户名
+            chat_id=message_data.get('chat_id', ''),
+            chat_type=chat_type,
+            content=content,
+            raw_content=raw_content,
+            mentioned=mentioned,
+            mentions=[m.get('key', '') for m in message_data.get('mentions', [])],
+            timestamp=timestamp,
+            raw_data=data,
+        )
+    
+    def _extract_command(self, text: str, event: Dict) -> str:
+        """
+        提取命令内容（去除 @机器人）
+        
+        飞书的 @用户 格式是：@_user_1
+        """
+        # 移除 @提及
+        mentions = event.get('message', {}).get('mentions', [])
+        for mention in mentions:
+            key = mention.get('key', '')
+            if key:
+                text = text.replace(key, '')
+        
+        # 清理多余空格
+        return ' '.join(text.split())
+    
+    def format_response(
+        self, 
+        response: BotResponse, 
+        message: BotMessage
+    ) -> WebhookResponse:
+        """
+        格式化飞书响应
+        
+        飞书 Webhook 只需要返回空响应，实际回复需要调用 API。
+        这里我们返回空响应，然后通过 NotificationService 发送消息。
+        """
+        # 飞书事件回调只需要返回空 200 响应
+        # 实际的消息回复需要调用飞书 API（通过 NotificationService）
+        
+        if response.text:
+            # 通过通知服务发送响应
+            self._send_reply(response, message)
+        
+        return WebhookResponse.success()
+    
+    def _send_reply(self, response: BotResponse, message: BotMessage) -> None:
+        """
+        发送回复消息
+        
+        通过飞书 API 发送回复（异步，不阻塞 Webhook 响应）
+        """
+        import threading
+        
+        def _send():
+            try:
+                from notification import NotificationService
+                
+                notifier = NotificationService()
+                
+                # 构建回复内容
+                text = response.text
+                if response.at_user and message.user_id:
+                    # 飞书的 @用户 需要使用 open_id
+                    text = f"<at user_id=\"{message.user_id}\"></at> {text}"
+                
+                # 发送到飞书
+                notifier.send_to_feishu(text)
+                
+            except Exception as e:
+                logger.error(f"[Feishu] 发送回复失败: {e}")
+        
+        # 异步发送
+        thread = threading.Thread(target=_send, daemon=True)
+        thread.start()

bot/platforms/telegram.py

@@ -0,0 +1,391 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+Telegram 平台适配器
+===================================
+
+处理 Telegram Bot 的 Webhook 更新。
+
+Telegram Bot API 文档：
+https://core.telegram.org/bots/api
+"""
+
+import hashlib
+import hmac
+import logging
+import threading
+from datetime import datetime
+from typing import Dict, Any, Optional
+
+from bot.platforms.base import BotPlatform
+from bot.models import BotMessage, BotResponse, WebhookResponse, ChatType
+
+logger = logging.getLogger(__name__)
+
+
+class TelegramPlatform(BotPlatform):
+    """
+    Telegram 平台适配器
+    
+    支持：
+    - Webhook 更新处理
+    - 消息解析
+    - 回复消息发送
+    
+    配置要求：
+    - TELEGRAM_BOT_TOKEN: Bot Token（从 @BotFather 获取）
+    - TELEGRAM_WEBHOOK_SECRET: Webhook 密钥（可选，用于验证请求）
+    
+    Webhook 设置：
+    使用 setWebhook API 设置回调 URL：
+    https://api.telegram.org/bot<token>/setWebhook?url=<webhook_url>&secret_token=<secret>
+    """
+    
+    def __init__(self):
+        from config import get_config
+        config = get_config()
+        
+        self._bot_token = getattr(config, 'telegram_bot_token', None)
+        self._webhook_secret = getattr(config, 'telegram_webhook_secret', None)
+        self._chat_id = getattr(config, 'telegram_chat_id', None)
+        
+        # Bot 用户名（用于识别 @提及）
+        self._bot_username = None
+    
+    @property
+    def platform_name(self) -> str:
+        return "telegram"
+    
+    def verify_request(self, headers: Dict[str, str], body: bytes) -> bool:
+        """
+        验证 Telegram Webhook 请求
+        
+        Telegram 使用 X-Telegram-Bot-Api-Secret-Token 头验证请求。
+        """
+        if not self._webhook_secret:
+            logger.debug("[Telegram] 未配置 webhook_secret，跳过验证")
+            return True
+        
+        secret_token = headers.get('X-Telegram-Bot-Api-Secret-Token', '')
+        
+        if secret_token != self._webhook_secret:
+            logger.warning("[Telegram] Webhook 密钥验证失败")
+            return False
+        
+        return True
+    
+    def handle_challenge(self, data: Dict[str, Any]) -> Optional[WebhookResponse]:
+        """Telegram 不需要 URL 验证"""
+        return None
+    
+    def parse_message(self, data: Dict[str, Any]) -> Optional[BotMessage]:
+        """
+        解析 Telegram 更新
+        
+        Telegram Update 格式：
+        {
+            "update_id": 123456,
+            "message": {
+                "message_id": 123,
+                "from": {
+                    "id": 123456789,
+                    "is_bot": false,
+                    "first_name": "John",
+                    "last_name": "Doe",
+                    "username": "johndoe"
+                },
+                "chat": {
+                    "id": -123456789,
+                    "title": "群名",
+                    "type": "group"  # private, group, supergroup, channel
+                },
+                "date": 1234567890,
+                "text": "/analyze 600519",
+                "entities": [
+                    {"type": "bot_command", "offset": 0, "length": 8},
+                    {"type": "mention", "offset": 10, "length": 5}
+                ]
+            }
+        }
+        """
+        # 获取消息对象
+        message_data = data.get('message') or data.get('edited_message')
+        
+        if not message_data:
+            # 可能是其他类型的更新（callback_query, inline_query 等）
+            logger.debug("[Telegram] 忽略非消息更新")
+            return None
+        
+        # 只处理文本消息
+        text = message_data.get('text', '')
+        if not text:
+            logger.debug("[Telegram] 忽略非文本消息")
+            return None
+        
+        # 获取发送者信息
+        from_user = message_data.get('from', {})
+        
+        # 获取用户名
+        user_name = from_user.get('first_name', '')
+        if from_user.get('last_name'):
+            user_name += f" {from_user['last_name']}"
+        if not user_name:
+            user_name = from_user.get('username', str(from_user.get('id', '')))
+        
+        # 获取会话信息
+        chat = message_data.get('chat', {})
+        chat_type_str = chat.get('type', '')
+        
+        if chat_type_str == 'private':
+            chat_type = ChatType.PRIVATE
+        elif chat_type_str in ('group', 'supergroup'):
+            chat_type = ChatType.GROUP
+        else:
+            chat_type = ChatType.UNKNOWN
+        
+        # 检查是否 @了机器人
+        entities = message_data.get('entities', [])
+        mentioned = self._check_mention(text, entities)
+        
+        # 提取命令
+        content = self._extract_command(text, entities)
+        
+        # 创建时间
+        date = message_data.get('date', 0)
+        try:
+            timestamp = datetime.fromtimestamp(date)
+        except (ValueError, TypeError):
+            timestamp = datetime.now()
+        
+        return BotMessage(
+            platform=self.platform_name,
+            message_id=str(message_data.get('message_id', '')),
+            user_id=str(from_user.get('id', '')),
+            user_name=user_name,
+            chat_id=str(chat.get('id', '')),
+            chat_type=chat_type,
+            content=content,
+            raw_content=text,
+            mentioned=mentioned,
+            mentions=self._extract_mentions(text, entities),
+            timestamp=timestamp,
+            raw_data=data,
+        )
+    
+    def _check_mention(self, text: str, entities: list) -> bool:
+        """检查是否 @了机器人"""
+        for entity in entities:
+            if entity.get('type') == 'mention':
+                offset = entity.get('offset', 0)
+                length = entity.get('length', 0)
+                mention = text[offset:offset + length]
+                
+                # 检查是否是机器人的用户名
+                if self._bot_username and mention.lower() == f"@{self._bot_username.lower()}":
+                    return True
+        
+        return False
+    
+    def _extract_mentions(self, text: str, entities: list) -> list:
+        """提取所有 @提及"""
+        mentions = []
+        for entity in entities:
+            if entity.get('type') == 'mention':
+                offset = entity.get('offset', 0)
+                length = entity.get('length', 0)
+                mention = text[offset:offset + length]
+                if mention.startswith('@'):
+                    mentions.append(mention[1:])  # 去掉 @
+        return mentions
+    
+    def _extract_command(self, text: str, entities: list) -> str:
+        """
+        提取命令内容
+        
+        Telegram 的命令格式：/command@botname args
+        """
+        # 移除 @botname 部分
+        if self._bot_username:
+            text = text.replace(f"@{self._bot_username}", "")
+        
+        # 清理多余空格
+        return ' '.join(text.split())
+    
+    def format_response(
+        self, 
+        response: BotResponse, 
+        message: BotMessage
+    ) -> WebhookResponse:
+        """
+        格式化 Telegram 响应
+        
+        Telegram Webhook 可以在响应中直接发送消息。
+        响应格式：
+        {
+            "method": "sendMessage",
+            "chat_id": 123456,
+            "text": "回复内容",
+            "parse_mode": "Markdown",
+            "reply_to_message_id": 123
+        }
+        """
+        if not response.text:
+            return WebhookResponse.success()
+        
+        # 构建响应
+        body = {
+            "method": "sendMessage",
+            "chat_id": int(message.chat_id) if message.chat_id.lstrip('-').isdigit() else message.chat_id,
+            "text": response.text,
+        }
+        
+        # Markdown 格式
+        if response.markdown:
+            body["parse_mode"] = "Markdown"
+        
+        # 回复原消息
+        if response.reply_to_message and message.message_id:
+            body["reply_to_message_id"] = int(message.message_id)
+        
+        return WebhookResponse.success(body)
+    
+    def send_message(
+        self, 
+        chat_id: str, 
+        text: str,
+        parse_mode: Optional[str] = None,
+        reply_to_message_id: Optional[int] = None
+    ) -> bool:
+        """
+        直接通过 API 发送消息
+        
+        用于异步发送或发送到其他会话。
+        
+        Args:
+            chat_id: 目标会话 ID
+            text: 消息内容
+            parse_mode: 解析模式（Markdown, HTML, MarkdownV2）
+            reply_to_message_id: 回复的消息 ID
+            
+        Returns:
+            是否发送成功
+        """
+        if not self._bot_token:
+            logger.warning("[Telegram] 未配置 bot_token")
+            return False
+        
+        import requests
+        
+        try:
+            api_url = f"https://api.telegram.org/bot{self._bot_token}/sendMessage"
+            
+            payload = {
+                "chat_id": chat_id,
+                "text": text,
+            }
+            
+            if parse_mode:
+                payload["parse_mode"] = parse_mode
+            
+            if reply_to_message_id:
+                payload["reply_to_message_id"] = reply_to_message_id
+            
+            resp = requests.post(api_url, json=payload, timeout=10)
+            
+            if resp.status_code == 200:
+                result = resp.json()
+                if result.get('ok'):
+                    logger.info("[Telegram] 消息发送成功")
+                    return True
+                else:
+                    logger.error(f"[Telegram] API 返回错误: {result}")
+                    return False
+            else:
+                logger.error(f"[Telegram] 请求失败: {resp.status_code}")
+                return False
+                
+        except Exception as e:
+            logger.error(f"[Telegram] 发送消息异常: {e}")
+            return False
+    
+    def set_webhook(self, url: str, secret_token: Optional[str] = None) -> bool:
+        """
+        设置 Webhook URL
+        
+        Args:
+            url: Webhook URL
+            secret_token: 密钥（可选）
+            
+        Returns:
+            是否设置成功
+        """
+        if not self._bot_token:
+            logger.warning("[Telegram] 未配置 bot_token")
+            return False
+        
+        import requests
+        
+        try:
+            api_url = f"https://api.telegram.org/bot{self._bot_token}/setWebhook"
+            
+            payload = {"url": url}
+            
+            if secret_token:
+                payload["secret_token"] = secret_token
+            
+            resp = requests.post(api_url, json=payload, timeout=10)
+            
+            if resp.status_code == 200:
+                result = resp.json()
+                if result.get('ok'):
+                    logger.info(f"[Telegram] Webhook 设置成功: {url}")
+                    return True
+                else:
+                    logger.error(f"[Telegram] 设置 Webhook 失败: {result}")
+                    return False
+            else:
+                logger.error(f"[Telegram] 请求失败: {resp.status_code}")
+                return False
+                
+        except Exception as e:
+            logger.error(f"[Telegram] 设置 Webhook 异常: {e}")
+            return False
+    
+    def delete_webhook(self) -> bool:
+        """删除 Webhook"""
+        if not self._bot_token:
+            return False
+        
+        import requests
+        
+        try:
+            api_url = f"https://api.telegram.org/bot{self._bot_token}/deleteWebhook"
+            resp = requests.post(api_url, timeout=10)
+            return resp.status_code == 200 and resp.json().get('ok', False)
+        except Exception as e:
+            logger.error(f"[Telegram] 删除 Webhook 异常: {e}")
+            return False
+    
+    def get_bot_info(self) -> Optional[Dict]:
+        """获取 Bot 信息"""
+        if not self._bot_token:
+            return None
+        
+        import requests
+        
+        try:
+            api_url = f"https://api.telegram.org/bot{self._bot_token}/getMe"
+            resp = requests.get(api_url, timeout=10)
+            
+            if resp.status_code == 200:
+                result = resp.json()
+                if result.get('ok'):
+                    bot_info = result.get('result', {})
+                    self._bot_username = bot_info.get('username')
+                    return bot_info
+            
+            return None
+            
+        except Exception as e:
+            logger.error(f"[Telegram] 获取 Bot 信息异常: {e}")
+            return None

bot/platforms/wecom.py

@@ -0,0 +1,300 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+企业微信平台适配器
+===================================
+
+处理企业微信机器人的回调消息。
+
+企业微信机器人文档：
+https://developer.work.weixin.qq.com/document/path/91770
+"""
+
+import hashlib
+import logging
+import time
+import xml.etree.ElementTree as ET
+from datetime import datetime
+from typing import Dict, Any, Optional
+
+from bot.platforms.base import BotPlatform
+from bot.models import BotMessage, BotResponse, WebhookResponse, ChatType
+
+logger = logging.getLogger(__name__)
+
+
+class WecomPlatform(BotPlatform):
+    """
+    企业微信平台适配器
+    
+    支持：
+    - 应用消息回调
+    - URL 验证
+    - 消息加解密
+    
+    配置要求：
+    - WECOM_CORPID: 企业 ID
+    - WECOM_TOKEN: 回调 Token
+    - WECOM_ENCODING_AES_KEY: 消息加解密密钥
+    - WECOM_AGENT_ID: 应用 AgentId
+    
+    注意：企业微信消息回调需要在企业微信管理后台配置回调 URL
+    """
+    
+    def __init__(self):
+        from config import get_config
+        config = get_config()
+        
+        self._corpid = getattr(config, 'wecom_corpid', None)
+        self._token = getattr(config, 'wecom_token', None)
+        self._encoding_aes_key = getattr(config, 'wecom_encoding_aes_key', None)
+        self._agent_id = getattr(config, 'wecom_agent_id', None)
+        
+        # 初始化加解密器（如果配置了密钥）
+        self._crypto = None
+        if self._corpid and self._token and self._encoding_aes_key:
+            try:
+                self._init_crypto()
+            except Exception as e:
+                logger.warning(f"[WeCom] 加解密器初始化失败: {e}")
+    
+    def _init_crypto(self):
+        """初始化消息加解密器"""
+        # 企业微信消息加解密需要额外的库
+        # 这里提供一个简化的实现框架
+        pass
+    
+    @property
+    def platform_name(self) -> str:
+        return "wecom"
+    
+    def verify_request(self, headers: Dict[str, str], body: bytes) -> bool:
+        """
+        验证企业微信请求签名
+        
+        签名算法：
+        1. 将 token, timestamp, nonce, msg_encrypt 排序后拼接
+        2. SHA1 加密
+        3. 比对签名
+        """
+        if not self._token:
+            logger.warning("[WeCom] 未配置 token，跳过签名验证")
+            return True
+        
+        # 从 URL 参数获取（需要在路由处理中传递）
+        # 这里假设参数已经放在 headers 中（实际实现需要从 URL 获取）
+        msg_signature = headers.get('msg_signature', '')
+        timestamp = headers.get('timestamp', '')
+        nonce = headers.get('nonce', '')
+        
+        if not all([msg_signature, timestamp, nonce]):
+            logger.debug("[WeCom] 缺少签名参数")
+            return True  # 可能是其他类型的请求
+        
+        # 解析 XML 获取加密消息
+        try:
+            root = ET.fromstring(body.decode('utf-8'))
+            encrypt = root.find('Encrypt')
+            msg_encrypt = encrypt.text if encrypt is not None else ''
+        except Exception as e:
+            logger.warning(f"[WeCom] 解析 XML 失败: {e}")
+            return False
+        
+        # 计算签名
+        sort_list = sorted([self._token, timestamp, nonce, msg_encrypt])
+        sign_str = ''.join(sort_list)
+        expected_signature = hashlib.sha1(sign_str.encode('utf-8')).hexdigest()
+        
+        if msg_signature != expected_signature:
+            logger.warning("[WeCom] 签名验证失败")
+            return False
+        
+        return True
+    
+    def handle_challenge(self, data: Dict[str, Any]) -> Optional[WebhookResponse]:
+        """
+        处理企业微信 URL 验证请求
+        
+        企业微信验证时会发送 GET 请求，包含 echostr 参数。
+        需要解密 echostr 后返回。
+        
+        注意：这个方法处理的是已解析的数据，
+        实际的 URL 验证在路由层处理（因为是 GET 请求）。
+        """
+        # 企业微信的 URL 验证是 GET 请求，不会到达这里
+        # 这里只是占位实现
+        return None
+    
+    def parse_message(self, data: Dict[str, Any]) -> Optional[BotMessage]:
+        """
+        解析企业微信消息
+        
+        企业微信消息格式（解密后的 XML）：
+        <xml>
+            <ToUserName><![CDATA[xxx]]></ToUserName>
+            <FromUserName><![CDATA[userid]]></FromUserName>
+            <CreateTime>1234567890</CreateTime>
+            <MsgType><![CDATA[text]]></MsgType>
+            <Content><![CDATA[@机器人 /analyze 600519]]></Content>
+            <MsgId>xxx</MsgId>
+            <AgentID>1000002</AgentID>
+        </xml>
+        
+        这里的 data 是已经解析好的字典格式。
+        """
+        # 检查消息类型
+        msg_type = data.get('MsgType', '')
+        if msg_type != 'text':
+            logger.debug(f"[WeCom] 忽略非文本消息: {msg_type}")
+            return None
+        
+        # 获取消息内容
+        raw_content = data.get('Content', '')
+        
+        # 提取命令
+        content = self._extract_command(raw_content)
+        
+        # 企业微信的 @提及 格式
+        mentioned = '@' in raw_content
+        
+        # 创建时间
+        create_time = data.get('CreateTime', '')
+        try:
+            timestamp = datetime.fromtimestamp(int(create_time))
+        except (ValueError, TypeError):
+            timestamp = datetime.now()
+        
+        return BotMessage(
+            platform=self.platform_name,
+            message_id=data.get('MsgId', ''),
+            user_id=data.get('FromUserName', ''),
+            user_name=data.get('FromUserName', ''),  # 企业微信返回的是 userid
+            chat_id=data.get('ToUserName', ''),
+            chat_type=ChatType.UNKNOWN,  # 企业微信需要额外判断
+            content=content,
+            raw_content=raw_content,
+            mentioned=mentioned,
+            mentions=[],
+            timestamp=timestamp,
+            raw_data=data,
+        )
+    
+    def _extract_command(self, text: str) -> str:
+        """提取命令内容"""
+        import re
+        # 移除 @提及
+        text = re.sub(r'@[\S]+\s*', '', text.strip())
+        return text.strip()
+    
+    def format_response(
+        self, 
+        response: BotResponse, 
+        message: BotMessage
+    ) -> WebhookResponse:
+        """
+        格式化企业微信响应
+        
+        企业微信被动回复消息格式（XML）：
+        <xml>
+            <ToUserName><![CDATA[userid]]></ToUserName>
+            <FromUserName><![CDATA[corpid]]></FromUserName>
+            <CreateTime>1234567890</CreateTime>
+            <MsgType><![CDATA[text]]></MsgType>
+            <Content><![CDATA[回复内容]]></Content>
+        </xml>
+        
+        注意：由于需要返回 XML 格式，这里的处理与其他平台不同。
+        """
+        if not response.text:
+            return WebhookResponse.success()
+        
+        # 企业微信的被动回复需要返回 XML 格式
+        # 但我们的 WebhookResponse 是 JSON 格式的
+        # 这里我们通过通知服务主动发送消息
+        self._send_reply(response, message)
+        
+        return WebhookResponse.success()
+    
+    def _send_reply(self, response: BotResponse, message: BotMessage) -> None:
+        """
+        发送回复消息
+        
+        通过企业微信 Webhook 发送回复
+        """
+        import threading
+        
+        def _send():
+            try:
+                from notification import NotificationService
+                
+                notifier = NotificationService()
+                
+                # 发送到企业微信
+                notifier.send_to_wechat(response.text)
+                
+            except Exception as e:
+                logger.error(f"[WeCom] 发送回复失败: {e}")
+        
+        # 异步发送
+        thread = threading.Thread(target=_send, daemon=True)
+        thread.start()
+    
+    def decrypt_message(self, encrypt_msg: str) -> str:
+        """
+        解密企业微信消息
+        
+        这里提供简化的框架，实际实现需要：
+        1. Base64 解码
+        2. AES 解密
+        3. 去除填充
+        4. 校验 corpid
+        
+        Args:
+            encrypt_msg: 加密的消息内容
+            
+        Returns:
+            解密后的 XML 字符串
+        """
+        if not self._encoding_aes_key:
+            logger.warning("[WeCom] 未配置加密密钥，无法解密")
+            return encrypt_msg
+        
+        # 实际的解密实现需要引入加密库
+        # 这里返回原始内容作为占位
+        logger.warning("[WeCom] 消息解密功能需要完整实现")
+        return encrypt_msg
+    
+    def encrypt_message(self, reply_msg: str) -> str:
+        """
+        加密回复消息
+        
+        Args:
+            reply_msg: 回复的 XML 字符串
+            
+        Returns:
+            加密后的消息
+        """
+        if not self._encoding_aes_key:
+            return reply_msg
+        
+        # 实际的加密实现需要引入加密库
+        logger.warning("[WeCom] 消息加密功能需要完整实现")
+        return reply_msg
+    
+    @staticmethod
+    def parse_xml_to_dict(xml_str: str) -> Dict[str, str]:
+        """
+        将 XML 字符串解析为字典
+        
+        Args:
+            xml_str: XML 字符串
+            
+        Returns:
+            解析后的字典
+        """
+        try:
+            root = ET.fromstring(xml_str)
+            return {child.tag: child.text or '' for child in root}
+        except ET.ParseError as e:
+            logger.error(f"[WeCom] XML 解析失败: {e}")
+            return {}

config.py

@@ -126,6 +126,31 @@ class Config:
     webui_host: str = "127.0.0.1"
     webui_port: int = 8000
     
+    # === 机器人配置 ===
+    bot_enabled: bool = True              # 是否启用机器人功能
+    bot_command_prefix: str = "/"         # 命令前缀
+    bot_rate_limit_requests: int = 10     # 频率限制：窗口内最大请求数
+    bot_rate_limit_window: int = 60       # 频率限制：窗口时间（秒）
+    bot_admin_users: List[str] = field(default_factory=list)  # 管理员用户 ID 列表
+    
+    # 飞书机器人（事件订阅）- 已有 feishu_app_id, feishu_app_secret
+    feishu_verification_token: Optional[str] = None  # 事件订阅验证 Token
+    feishu_encrypt_key: Optional[str] = None         # 消息加密密钥（可选）
+    
+    # 钉钉机器人
+    dingtalk_app_key: Optional[str] = None      # 应用 AppKey
+    dingtalk_app_secret: Optional[str] = None   # 应用 AppSecret
+    dingtalk_stream_enabled: bool = False       # 是否启用 Stream 模式（无需公网IP）
+    
+    # 企业微信机器人（回调模式）
+    wecom_corpid: Optional[str] = None              # 企业 ID
+    wecom_token: Optional[str] = None               # 回调 Token
+    wecom_encoding_aes_key: Optional[str] = None    # 消息加解密密钥
+    wecom_agent_id: Optional[str] = None            # 应用 AgentId
+    
+    # Telegram 机器人 - 已有 telegram_bot_token, telegram_chat_id
+    telegram_webhook_secret: Optional[str] = None   # Webhook 密钥
+    
     # 单例实例存储
     _instance: Optional['Config'] = None
     
@@ -222,6 +247,26 @@ class Config:
             webui_enabled=os.getenv('WEBUI_ENABLED', 'false').lower() == 'true',
             webui_host=os.getenv('WEBUI_HOST', '127.0.0.1'),
             webui_port=int(os.getenv('WEBUI_PORT', '8000')),
+            # 机器人配置
+            bot_enabled=os.getenv('BOT_ENABLED', 'true').lower() == 'true',
+            bot_command_prefix=os.getenv('BOT_COMMAND_PREFIX', '/'),
+            bot_rate_limit_requests=int(os.getenv('BOT_RATE_LIMIT_REQUESTS', '10')),
+            bot_rate_limit_window=int(os.getenv('BOT_RATE_LIMIT_WINDOW', '60')),
+            bot_admin_users=[u.strip() for u in os.getenv('BOT_ADMIN_USERS', '').split(',') if u.strip()],
+            # 飞书机器人
+            feishu_verification_token=os.getenv('FEISHU_VERIFICATION_TOKEN'),
+            feishu_encrypt_key=os.getenv('FEISHU_ENCRYPT_KEY'),
+            # 钉钉机器人
+            dingtalk_app_key=os.getenv('DINGTALK_APP_KEY'),
+            dingtalk_app_secret=os.getenv('DINGTALK_APP_SECRET'),
+            dingtalk_stream_enabled=os.getenv('DINGTALK_STREAM_ENABLED', 'false').lower() == 'true',
+            # 企业微信机器人
+            wecom_corpid=os.getenv('WECOM_CORPID'),
+            wecom_token=os.getenv('WECOM_TOKEN'),
+            wecom_encoding_aes_key=os.getenv('WECOM_ENCODING_AES_KEY'),
+            wecom_agent_id=os.getenv('WECOM_AGENT_ID'),
+            # Telegram
+            telegram_webhook_secret=os.getenv('TELEGRAM_WEBHOOK_SECRET'),
         )
     
     @classmethod

docs/bot/bot-command.md

@@ -0,0 +1,265 @@
+
+
+## 一、整体设计
+
+```mermaid
+flowchart TB
+    subgraph Platforms [外部平台]
+        FS[飞书]
+        DT[钉钉]
+        WC[企业微信]
+        TG[Telegram]
+        More[更多平台...]
+    end
+
+    subgraph BotModule [bot/ 模块]
+        WH[Webhook Server]
+        Adapters[平台适配器]
+        Dispatcher[命令分发器]
+        Commands[命令处理器]
+    end
+
+    subgraph Core [现有核心模块]
+        AS[AnalysisService]
+        MA[MarketAnalyzer]
+        NS[NotificationService]
+    end
+
+    FS -->|POST /bot/feishu| WH
+    DT -->|POST /bot/dingtalk| WH
+    WC -->|POST /bot/wecom| WH
+    TG -->|POST /bot/telegram| WH
+
+    WH --> Adapters
+    Adapters -->|统一消息格式| Dispatcher
+    Dispatcher --> Commands
+    Commands --> AS
+    Commands --> MA
+    Commands --> NS
+```
+
+
+
+## 二、目录结构
+
+在项目根目录新建 `bot/` 目录：
+
+```
+bot/
+├── __init__.py             # 模块入口，导出主要类
+├── models.py               # 统一的消息/响应模型
+├── dispatcher.py           # 命令分发器（核心）
+├── commands/               # 命令处理器
+│   ├── __init__.py
+│   ├── base.py             # 命令抽象基类
+│   ├── analyze.py          # /analyze 股票分析
+│   ├── market.py           # /market 大盘复盘
+│   ├── help.py             # /help 帮助信息
+│   └── status.py           # /status 系统状态
+└── platforms/              # 平台适配器
+    ├── __init__.py
+    ├── base.py             # 平台抽象基类
+    ├── feishu.py           # 飞书机器人
+    ├── dingtalk.py         # 钉钉机器人
+    ├── dingtalk_stream.py  # 钉钉机器人Stream
+    ├── wecom.py            # 企业微信机器人
+    └── telegram.py         # Telegram 机器人
+```
+
+## 三、核心抽象设计
+
+### 3.1 统一消息模型 (`bot/models.py`)
+
+```python
+@dataclass
+class BotMessage:
+    """统一的机器人消息模型"""
+    platform: str           # 平台标识: feishu/dingtalk/wecom/telegram
+    user_id: str            # 发送者 ID
+    user_name: str          # 发送者名称
+    chat_id: str            # 会话 ID（群聊或私聊）
+    chat_type: str          # 会话类型: group/private
+    content: str            # 消息文本内容
+    raw_data: Dict          # 原始请求数据（平台特定）
+    timestamp: datetime     # 消息时间
+    mentioned: bool = False # 是否@了机器人
+
+@dataclass
+class BotResponse:
+    """统一的机器人响应模型"""
+    text: str               # 回复文本
+    markdown: bool = False  # 是否为 Markdown
+    at_user: bool = True    # 是否@发送者
+```
+
+### 3.2 平台适配器基类 (`bot/platforms/base.py`)
+
+```python
+class BotPlatform(ABC):
+    """平台适配器抽象基类"""
+    
+    @property
+    @abstractmethod
+    def platform_name(self) -> str:
+        """平台标识名称"""
+        pass
+    
+    @abstractmethod
+    def verify_request(self, headers: Dict, body: bytes) -> bool:
+        """验证请求签名（安全校验）"""
+        pass
+    
+    @abstractmethod
+    def parse_message(self, data: Dict) -> Optional[BotMessage]:
+        """解析平台消息为统一格式"""
+        pass
+    
+    @abstractmethod
+    def format_response(self, response: BotResponse) -> Dict:
+        """将统一响应转换为平台格式"""
+        pass
+```
+
+### 3.3 命令基类 (`bot/commands/base.py`)
+
+```python
+class BotCommand(ABC):
+    """命令处理器抽象基类"""
+    
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """命令名称 (如 'analyze')"""
+        pass
+    
+    @property
+    @abstractmethod
+    def aliases(self) -> List[str]:
+        """命令别名 (如 ['a', '分析'])"""
+        pass
+    
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """命令描述"""
+        pass
+    
+    @property
+    @abstractmethod
+    def usage(self) -> str:
+        """使用说明"""
+        pass
+    
+    @abstractmethod
+    async def execute(self, message: BotMessage, args: List[str]) -> BotResponse:
+        """执行命令"""
+        pass
+```
+
+### 3.4 命令分发器 (`bot/dispatcher.py`)
+
+```python
+class CommandDispatcher:
+    """命令分发器 - 单例模式"""
+    
+    def __init__(self):
+        self._commands: Dict[str, BotCommand] = {}
+        self._aliases: Dict[str, str] = {}
+    
+    def register(self, command: BotCommand) -> None:
+        """注册命令"""
+        self._commands[command.name] = command
+        for alias in command.aliases:
+            self._aliases[alias] = command.name
+    
+    def dispatch(self, message: BotMessage) -> BotResponse:
+        """分发消息到对应命令"""
+        # 1. 解析命令和参数
+        # 2. 查找命令处理器
+        # 3. 执行并返回响应
+```
+
+## 四、已支持的命令
+
+| 命令 | 别名 | 说明 | 示例 |
+
+|------|------|------|------|
+
+| /analyze | /a, 分析 | 分析指定股票 | `/analyze 600519` |
+
+| /market | /m, 大盘 | 大盘复盘 | `/market` |
+
+| /batch | /b, 批量 | 批量分析自选股 | `/batch` |
+
+| /help | /h, 帮助 | 显示帮助信息 | `/help` |
+
+| /status | /s, 状态 | 系统状态 | `/status` |
+
+## 五、Webhook 路由
+
+在 [web/router.py](../../web/router.py) 中注册新路由：
+
+```python
+# Webhook 路由
+/bot/feishu      # POST - 飞书事件回调
+/bot/dingtalk    # POST - 钉钉事件回调
+/bot/wecom       # POST - 企业微信事件回调
+/bot/telegram    # POST - Telegram 更新回调
+```
+
+## 配置
+
+在 [config.py](../../config.py) 中新增机器人配置：
+
+```python
+# === 机器人配置 ===
+bot_enabled: bool = False              # 是否启用机器人
+bot_command_prefix: str = "/"          # 命令前缀
+
+# 飞书机器人（事件订阅）
+feishu_app_id: str                     # 已有
+feishu_app_secret: str                 # 已有
+feishu_verification_token: str         # 新增：事件校验 Token
+feishu_encrypt_key: str                # 新增：加密密钥
+
+# 钉钉机器人（应用）
+dingtalk_app_key: str                  # 新增
+dingtalk_app_secret: str               # 新增
+
+# 企业微信机器人
+wecom_token: str                       # 新增：回调 Token
+wecom_encoding_aes_key: str            # 新增：EncodingAESKey
+
+# Telegram 机器人
+telegram_bot_token: str                # 已有
+telegram_webhook_secret: str           # 新增：Webhook 密钥
+```
+
+## 扩展说明
+### 怎样新增一个通知平台
+
+1. 在 `bot/platforms/` 创建新文件
+2. 继承 `BotPlatform` 基类
+3. 实现 `verify_request`, `parse_message`, `format_response`
+4. 在路由中注册 Webhook 端点
+
+### 怎样新增新增命令
+
+1. 在 `bot/commands/` 创建新文件
+2. 继承 `BotCommand` 基类
+3. 实现 `execute` 方法
+4. 在分发器中注册命令
+
+## 安全相关配置
+
+- 支持命令频率限制（防刷）
+- 敏感操作（如批量分析）可设置权限白名单
+
+在 [config.py](../../config.py) 中新增机器人安全配置：
+
+```python
+    bot_rate_limit_requests: int = 10     # 频率限制：窗口内最大请求数
+    bot_rate_limit_window: int = 60       # 频率限制：窗口时间（秒）
+    bot_admin_users: List[str] = field(default_factory=list)  # 管理员用户 ID 列表，限制敏感操作
+```
+

main.py

@@ -47,6 +47,7 @@ from data_provider import DataFetcherManager
 from data_provider.akshare_fetcher import AkshareFetcher, RealtimeQuote, ChipDistribution
 from analyzer import GeminiAnalyzer, AnalysisResult, STOCK_NAME_MAP
 from notification import NotificationService, NotificationChannel, send_daily_report
+from bot.models import BotMessage
 from search_service import SearchService, SearchResponse
 from enums import ReportType
 from stock_analyzer import StockTrendAnalyzer, TrendAnalysisResult
@@ -135,7 +136,8 @@ class StockAnalysisPipeline:
     def __init__(
         self,
         config: Optional[Config] = None,
-        max_workers: Optional[int] = None
+        max_workers: Optional[int] = None,
+        source_message: Optional[BotMessage] = None
     ):
         """
         初始化调度器
@@ -146,6 +148,7 @@ class StockAnalysisPipeline:
         """
         self.config = config or get_config()
         self.max_workers = max_workers or self.config.max_workers
+        self.source_message = source_message
         
         # 初始化各模块
         self.db = get_db()
@@ -153,7 +156,7 @@ class StockAnalysisPipeline:
         self.akshare_fetcher = AkshareFetcher()  # 用于获取增强数据（量比、筹码等）
         self.trend_analyzer = StockTrendAnalyzer()  # 趋势分析器
         self.analyzer = GeminiAnalyzer()
-        self.notifier = NotificationService()
+        self.notifier = NotificationService(source_message=source_message)
         
         # 初始化搜索服务
         self.search_service = SearchService(
@@ -622,6 +625,7 @@ class StockAnalysisPipeline:
             # 推送通知
             if self.notifier.is_available():
                 channels = self.notifier.get_available_channels()
+                context_success = self.notifier.send_to_context(report)
 
                 # 企业微信：只发精简版（平台限制）
                 wechat_success = False
@@ -647,7 +651,7 @@ class StockAnalysisPipeline:
                     else:
                         logger.warning(f"未知通知渠道: {channel}")
 
-                success = wechat_success or non_wechat_success
+                success = wechat_success or non_wechat_success or context_success
                 if success:
                     logger.info("决策仪表盘推送成功")
                 else:
@@ -889,6 +893,25 @@ def run_full_analysis(
         logger.exception(f"分析流程执行失败: {e}")
 
 
+def start_bot_stream_clients(config: Config) -> None:
+    """Start bot stream clients when enabled in config."""
+    if not config.dingtalk_stream_enabled:
+        return
+
+    try:
+        from bot.platforms import start_dingtalk_stream_background, DINGTALK_STREAM_AVAILABLE
+        if DINGTALK_STREAM_AVAILABLE:
+            if start_dingtalk_stream_background():
+                logger.info("[Main] Dingtalk Stream client started in background.")
+            else:
+                logger.warning("[Main] Dingtalk Stream client failed to start.")
+        else:
+            logger.warning("[Main] Dingtalk Stream enabled but SDK is missing.")
+            logger.warning("[Main] Run: pip install dingtalk-stream")
+    except Exception as exc:
+        logger.error(f"[Main] Failed to start Dingtalk Stream client: {exc}")
+
+
 def main() -> int:
     """
     主入口函数
@@ -929,6 +952,7 @@ def main() -> int:
         try:
             from webui import run_server_in_thread
             run_server_in_thread(host=config.webui_host, port=config.webui_port)
+            start_bot_stream_clients(config)
         except Exception as e:
             logger.error(f"启动 WebUI 失败: {e}")
     

notification.py

@@ -30,6 +30,7 @@ import requests
 
 from config import get_config
 from analyzer import AnalysisResult
+from bot.models import BotMessage
 
 logger = logging.getLogger(__name__)
 
@@ -111,13 +112,15 @@ class NotificationService:
     注意：所有已配置的渠道都会收到推送
     """
     
-    def __init__(self):
+    def __init__(self, source_message: Optional[BotMessage] = None):
         """
         初始化通知服务
         
         检测所有已配置的渠道，推送时会向所有渠道发送
         """
         config = get_config()
+        self._source_message = source_message
+        self._context_channels: List[str] = []
         
         # 各渠道的 Webhook URL
         self._wechat_url = config.wechat_webhook_url
@@ -152,12 +155,15 @@ class NotificationService:
         
         # 检测所有已配置的渠道
         self._available_channels = self._detect_all_channels()
+        if self._has_context_channel():
+            self._context_channels.append("钉钉会话")
         
-        if not self._available_channels:
+        if not self._available_channels and not self._context_channels:
             logger.warning("未配置有效的通知渠道，将不发送推送通知")
         else:
             channel_names = [ChannelDetector.get_channel_name(ch) for ch in self._available_channels]
-            logger.info(f"已配置 {len(self._available_channels)} 个通知渠道：{', '.join(channel_names)}")
+            channel_names.extend(self._context_channels)
+            logger.info(f"已配置 {len(channel_names)} 个通知渠道：{', '.join(channel_names)}")
     
     def _detect_all_channels(self) -> List[NotificationChannel]:
         """
@@ -207,8 +213,8 @@ class NotificationService:
         return bool(self._pushover_config['user_key'] and self._pushover_config['api_token'])
     
     def is_available(self) -> bool:
-        """检查通知服务是否可用（至少有一个渠道）"""
-        return len(self._available_channels) > 0
+        """检查通知服务是否可用（至少有一个渠道或上下文渠道）"""
+        return len(self._available_channels) > 0 or self._has_context_channel()
     
     def get_available_channels(self) -> List[NotificationChannel]:
         """获取所有已配置的渠道"""
@@ -216,7 +222,40 @@ class NotificationService:
     
     def get_channel_names(self) -> str:
         """获取所有已配置渠道的名称"""
-        return ', '.join([ChannelDetector.get_channel_name(ch) for ch in self._available_channels])
+        names = [ChannelDetector.get_channel_name(ch) for ch in self._available_channels]
+        if self._has_context_channel():
+            names.append("钉钉会话")
+        return ', '.join(names)
+
+    def _has_context_channel(self) -> bool:
+        """判断是否存在基于消息上下文的临时渠道（如钉钉会话）"""
+        return self._extract_dingtalk_session_webhook() is not None
+
+    def _extract_dingtalk_session_webhook(self) -> Optional[str]:
+        """从来源消息中提取钉钉会话 Webhook（用于 Stream 模式回复）"""
+        if not isinstance(self._source_message, BotMessage):
+            return None
+        raw_data = getattr(self._source_message, "raw_data", {}) or {}
+        if not isinstance(raw_data, dict):
+            return None
+        session_webhook = (
+            raw_data.get("_session_webhook")
+            or raw_data.get("sessionWebhook")
+            or raw_data.get("session_webhook")
+            or raw_data.get("session_webhook_url")
+        )
+        if not session_webhook and isinstance(raw_data.get("headers"), dict):
+            session_webhook = raw_data["headers"].get("sessionWebhook")
+        return session_webhook
+
+    def send_to_context(self, content: str) -> bool:
+        """
+        向基于消息上下文的渠道发送消息（例如钉钉 Stream 会话）
+        
+        Args:
+            content: Markdown 格式内容
+        """
+        return self._send_via_source_context(content)
     
     def generate_daily_report(
         self, 
@@ -2301,6 +2340,26 @@ class NotificationService:
             "message": content,
             "body": content
         }
+
+    def _send_via_source_context(self, content: str) -> bool:
+        """
+        使用消息上下文（如钉钉会话 webhook）发送一份报告
+        
+        主要用于从钉钉 Stream 触发的任务，确保结果能回到触发的会话。
+        """
+        session_webhook = self._extract_dingtalk_session_webhook()
+        if not session_webhook:
+            return False
+
+        try:
+            if self._send_dingtalk_chunked(session_webhook, content, max_bytes=20000):
+                logger.info("已通过钉钉会话（Stream）推送报告")
+                return True
+            logger.error("钉钉会话（Stream）推送失败")
+            return False
+        except Exception as e:
+            logger.error(f"钉钉会话（Stream）推送异常: {e}")
+            return False
     
     def send(self, content: str) -> bool:
         """
@@ -2314,7 +2373,12 @@ class NotificationService:
         Returns:
             是否至少有一个渠道发送成功
         """
-        if not self.is_available():
+        context_success = self.send_to_context(content)
+
+        if not self._available_channels:
+            if context_success:
+                logger.info("已通过消息上下文渠道完成推送（无其他通知渠道）")
+                return True
             logger.warning("通知服务不可用，跳过推送")
             return False
         
@@ -2353,7 +2417,7 @@ class NotificationService:
                 fail_count += 1
         
         logger.info(f"通知发送完成：成功 {success_count} 个，失败 {fail_count} 个")
-        return success_count > 0
+        return success_count > 0 or context_success
     
     def _send_chunked_messages(self, content: str, max_length: int) -> bool:
         """

requirements.txt

@@ -34,6 +34,6 @@ google-search-results>=2.4.0  # SerpAPI（每月 100 次免费）
 requests>=2.31.0            # HTTP 请求
 fake-useragent>=1.4.0       # 随机 User-Agent 防封禁
 httpx[socks]                # HTTP 客户端 + SOCKS 代理支持（OpenAI 可选依赖）
-
+dingtalk-stream >= 0.24.3    # 钉钉 Stream SDK
 # 数据库
 # SQLite 是 Python 内置，无需额外安装

web/handlers.py

@@ -246,12 +246,64 @@ class ApiHandler:
         return JsonResponse({"success": True, "task": task})
 
 
+# ============================================================
+# Bot Webhook 处理器
+# ============================================================
+
+class BotHandler:
+    """
+    机器人 Webhook 处理器
+    
+    处理各平台的机器人回调请求。
+    """
+    
+    def handle_webhook(self, platform: str, form_data: Dict[str, list], headers: Dict[str, str], body: bytes) -> Response:
+        """
+        处理 Webhook 请求
+        
+        Args:
+            platform: 平台名称 (feishu, dingtalk, wecom, telegram)
+            form_data: POST 数据（已解析）
+            headers: HTTP 请求头
+            body: 原始请求体
+            
+        Returns:
+            Response 对象
+        """
+        try:
+            from bot.handler import handle_webhook
+            from bot.models import WebhookResponse
+            
+            # 调用 bot 模块处理
+            webhook_response = handle_webhook(platform, headers, body)
+            
+            # 转换为 web 响应
+            return JsonResponse(
+                webhook_response.body,
+                status=HTTPStatus(webhook_response.status_code)
+            )
+            
+        except ImportError as e:
+            logger.error(f"[BotHandler] Bot 模块未正确安装: {e}")
+            return JsonResponse(
+                {"error": "Bot module not available"},
+                status=HTTPStatus.INTERNAL_SERVER_ERROR
+            )
+        except Exception as e:
+            logger.error(f"[BotHandler] 处理 {platform} Webhook 失败: {e}")
+            return JsonResponse(
+                {"error": str(e)},
+                status=HTTPStatus.INTERNAL_SERVER_ERROR
+            )
+
+
 # ============================================================
 # 处理器工厂
 # ============================================================
 
 _page_handler: PageHandler | None = None
 _api_handler: ApiHandler | None = None
+_bot_handler: BotHandler | None = None
 
 
 def get_page_handler() -> PageHandler:
@@ -268,3 +320,11 @@ def get_api_handler() -> ApiHandler:
     if _api_handler is None:
         _api_handler = ApiHandler()
     return _api_handler
+
+
+def get_bot_handler() -> BotHandler:
+    """获取 Bot 处理器实例"""
+    global _bot_handler
+    if _bot_handler is None:
+        _bot_handler = BotHandler()
+    return _bot_handler

web/router.py

@@ -18,8 +18,8 @@ from typing import Callable, Dict, List, Optional, TYPE_CHECKING, Tuple
 from urllib.parse import parse_qs, urlparse
 
 from web.handlers import (
-    Response, HtmlResponse,
-    get_page_handler, get_api_handler
+    Response, HtmlResponse, JsonResponse,
+    get_page_handler, get_api_handler, get_bot_handler
 )
 from web.templates import render_error_page
 
@@ -173,9 +173,17 @@ class Router:
         parsed = urlparse(request_handler.path)
         path = parsed.path
         
-        # 读取 POST body
+        # 读取 POST body（保留原始字节用于 Bot Webhook）
         content_length = int(request_handler.headers.get("Content-Length", "0") or "0")
-        raw_body = request_handler.rfile.read(content_length).decode("utf-8", errors="replace")
+        raw_body_bytes = request_handler.rfile.read(content_length)
+        raw_body = raw_body_bytes.decode("utf-8", errors="replace")
+        
+        # 检查是否是 Bot Webhook 路由
+        if path.startswith("/bot/"):
+            self._dispatch_bot_webhook(request_handler, path, raw_body_bytes)
+            return
+        
+        # 普通 POST 请求
         form_data = parse_qs(raw_body)
         
         # 匹配路由
@@ -194,6 +202,42 @@ class Router:
             logger.error(f"[Router] 处理 POST 请求失败: {path} - {e}")
             self._send_error(request_handler, str(e))
     
+    def _dispatch_bot_webhook(
+        self,
+        request_handler: 'BaseHTTPRequestHandler',
+        path: str,
+        body: bytes
+    ) -> None:
+        """
+        分发 Bot Webhook 请求
+        
+        Bot Webhook 需要原始 body 和 headers，与普通路由处理不同。
+        
+        Args:
+            request_handler: HTTP 请求处理器
+            path: 请求路径
+            body: 原始请求体字节
+        """
+        # 提取平台名称：/bot/feishu -> feishu
+        parts = path.strip('/').split('/')
+        if len(parts) < 2:
+            self._send_not_found(request_handler, path)
+            return
+        
+        platform = parts[1]
+        
+        # 获取请求头
+        headers = {key: value for key, value in request_handler.headers.items()}
+        
+        try:
+            bot_handler = get_bot_handler()
+            response = bot_handler.handle_webhook(platform, {}, headers, body)
+            response.send(request_handler)
+            
+        except Exception as e:
+            logger.error(f"[Router] 处理 Bot Webhook 失败: {path} - {e}")
+            self._send_error(request_handler, str(e))
+    
     def list_routes(self) -> List[Tuple[str, str, str]]:
         """
         列出所有路由
@@ -278,6 +322,39 @@ def create_default_router() -> Router:
         "查询任务状态"
     )
     
+    # === Bot Webhook 路由 ===
+    # 注意：Bot Webhook 路由在 dispatch_post 中特殊处理
+    # 这里只是为了在路由列表中显示
+    # 实际请求会被 _dispatch_bot_webhook 方法处理
+    
+    # 飞书机器人 Webhook
+    router.register(
+        "/bot/feishu", "POST",
+        lambda form: JsonResponse({"error": "Use POST with JSON body"}),
+        "飞书机器人 Webhook"
+    )
+    
+    # 钉钉机器人 Webhook
+    router.register(
+        "/bot/dingtalk", "POST",
+        lambda form: JsonResponse({"error": "Use POST with JSON body"}),
+        "钉钉机器人 Webhook"
+    )
+    
+    # 企业微信机器人 Webhook
+    router.register(
+        "/bot/wecom", "POST",
+        lambda form: JsonResponse({"error": "Use POST with JSON body"}),
+        "企业微信机器人 Webhook"
+    )
+    
+    # Telegram 机器人 Webhook
+    router.register(
+        "/bot/telegram", "POST",
+        lambda form: JsonResponse({"error": "Use POST with JSON body"}),
+        "Telegram 机器人 Webhook"
+    )
+    
     return router
 
 

web/services.py

@@ -20,6 +20,7 @@ from datetime import datetime
 from typing import Optional, Dict, Any, List, Union
 
 from enums import ReportType
+from bot.models import BotMessage
 
 logger = logging.getLogger(__name__)
 
@@ -171,7 +172,8 @@ class AnalysisService:
     def submit_analysis(
         self, 
         code: str, 
-        report_type: Union[ReportType, str] = ReportType.SIMPLE
+        report_type: Union[ReportType, str] = ReportType.SIMPLE,
+        source_message: Optional[BotMessage] = None
     ) -> Dict[str, Any]:
         """
         提交异步分析任务
@@ -190,7 +192,7 @@ class AnalysisService:
         task_id = f"{code}_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
         
         # 提交到线程池
-        self.executor.submit(self._run_analysis, code, task_id, report_type)
+        self.executor.submit(self._run_analysis, code, task_id, report_type, source_message)
         
         logger.info(f"[AnalysisService] 已提交股票 {code} 的分析任务, task_id={task_id}, report_type={report_type.value}")
         
@@ -219,7 +221,8 @@ class AnalysisService:
         self, 
         code: str, 
         task_id: str, 
-        report_type: ReportType = ReportType.SIMPLE
+        report_type: ReportType = ReportType.SIMPLE,
+        source_message: Optional[BotMessage] = None
     ) -> Dict[str, Any]:
         """
         执行单只股票分析
@@ -252,7 +255,11 @@ class AnalysisService:
             
             # 创建分析管道
             config = get_config()
-            pipeline = StockAnalysisPipeline(config=config, max_workers=1)
+            pipeline = StockAnalysisPipeline(
+                config=config,
+                max_workers=1,
+                source_message=source_message
+            )
             
             # 执行单只股票分析（启用单股推送）
             result = pipeline.process_single_stock(

webui.py

@@ -65,6 +65,27 @@ __all__ = [
 ]
 
 
+def _start_bot_stream_clients() -> None:
+    """启动 Bot Stream 模式客户端（如果已配置）"""
+    from config import get_config
+    config = get_config()
+    
+    # 钉钉 Stream 模式
+    if config.dingtalk_stream_enabled:
+        try:
+            from bot.platforms import start_dingtalk_stream_background, DINGTALK_STREAM_AVAILABLE
+            if DINGTALK_STREAM_AVAILABLE:
+                if start_dingtalk_stream_background():
+                    logger.info("[WebUI] 钉钉 Stream 客户端已在后台启动")
+                else:
+                    logger.warning("[WebUI] 钉钉 Stream 客户端启动失败")
+            else:
+                logger.warning("[WebUI] 钉钉 Stream 模式已启用但 SDK 未安装")
+                logger.warning("[WebUI] 请运行: pip install dingtalk-stream")
+        except Exception as e:
+            logger.error(f"[WebUI] 启动钉钉 Stream 客户端失败: {e}")
+
+
 def main() -> int:
     """
     主入口函数
@@ -85,6 +106,15 @@ def main() -> int:
     print("  GET  /task?id=xxx   - 任务状态")
     print("  POST /update        - 更新配置")
     print()
+    print("Bot Webhooks:")
+    print("  POST /bot/feishu    - 飞书机器人")
+    print("  POST /bot/dingtalk  - 钉钉机器人")
+    print("  POST /bot/wecom     - 企业微信机器人")
+    print("  POST /bot/telegram  - Telegram 机器人")
+    print()
+    
+    # 启动 Bot Stream 客户端（如果配置了）
+    _start_bot_stream_clients()
     
     try:
         run_server(host=host, port=port)