From 4df277e109c38604e31e2184fcff36f561f8ffa5 Mon Sep 17 00:00:00 2001 From: zhulinsen Date: Sun, 11 Jan 2026 22:37:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A4=9A=E6=B8=A0=E9=81=93=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E6=94=AF=E6=8C=81=20(=E4=BC=81=E4=B8=9A=E5=BE=AE?= =?UTF-8?q?=E4=BF=A1/=E9=A3=9E=E4=B9=A6/Telegram/=E9=82=AE=E4=BB=B6?= =?UTF-8?q?=E5=90=8C=E6=97=B6=E6=8E=A8=E9=80=81)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 31 +- .github/workflows/daily_analysis.yml | 17 +- CHANGELOG.md | 14 +- README.md | 58 ++- config.py | 33 +- main.py | 24 +- notification.py | 604 +++++++++++++++++++++++++-- 7 files changed, 718 insertions(+), 63 deletions(-) diff --git a/.env.example b/.env.example index bb4923d52..781894f6c 100644 --- a/.env.example +++ b/.env.example @@ -50,8 +50,35 @@ TAVILY_API_KEYS=your_tavily_key_here # SerpAPI Keys（支持多个，逗号分隔） SERPAPI_KEYS=your_serpapi_key_here -# 企业微信机器人 Webhook URL（用于推送日报） -WECHAT_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key_here +# =================================== +# 通知渠道配置（可同时配置多个，全部推送） +# =================================== +# +# 【方式一】企业微信机器人 +# 在企业微信群 -> 设置 -> 群机器人 -> 添加 -> 复制 Webhook 地址 +# +# WECHAT_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key_here +# +# 【方式二】飞书机器人 +# 在飞书群 -> 设置 -> 群机器人 -> 添加机器人 -> 自定义机器人 -> 复制 Webhook 地址 +# +# FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/your_key_here +# +# 【方式三】Telegram 机器人（需同时配置两项） +# 1. 在 Telegram 找 @BotFather -> /newbot -> 获取 Bot Token +# 2. 获取 Chat ID：发消息给 @userinfobot 或访问 https://api.telegram.org/bot/getUpdates +# +# TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyz +# TELEGRAM_CHAT_ID=123456789 +# +# 【方式四】邮件推送（只需 2 项配置，SMTP 自动识别） +# 支持 QQ邮箱、163邮箱、Gmail、Outlook 等主流邮箱 +# 1. 获取授权码（以QQ邮箱为例）：设置 -> 账户 -> POP3/SMTP服务 -> 开启 -> 获取授权码 +# 2. 填写下面两项即可： +# +# EMAIL_SENDER=your_email@qq.com +# EMAIL_PASSWORD=your_email_auth_code +# EMAIL_RECEIVERS=receiver@example.com # 可选，留空则发给自己 # 数据库路径 DATABASE_PATH=./data/stock_analysis.db diff --git a/.github/workflows/daily_analysis.yml b/.github/workflows/daily_analysis.yml index 1888be2cd..6391f16d4 100644 --- a/.github/workflows/daily_analysis.yml +++ b/.github/workflows/daily_analysis.yml @@ -56,8 +56,23 @@ jobs: TAVILY_API_KEYS: ${{ secrets.TAVILY_API_KEYS }} SERPAPI_API_KEYS: ${{ secrets.SERPAPI_API_KEYS }} - # 企业微信通知 + # OpenAI 兼容 API（备选） + OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} + OPENAI_BASE_URL: ${{ secrets.OPENAI_BASE_URL }} + OPENAI_MODEL: ${{ secrets.OPENAI_MODEL }} + + # 通知渠道（可同时配置多个，全部推送） + # 方式一：企业微信 WECHAT_WEBHOOK_URL: ${{ secrets.WECHAT_WEBHOOK_URL }} + # 方式二：飞书 + FEISHU_WEBHOOK_URL: ${{ secrets.FEISHU_WEBHOOK_URL }} + # 方式三：Telegram（需同时配置 Bot Token 和 Chat ID） + TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }} + TELEGRAM_CHAT_ID: ${{ secrets.TELEGRAM_CHAT_ID }} + # 方式四：邮件（只需邮箱和授权码，SMTP自动识别） + EMAIL_SENDER: ${{ secrets.EMAIL_SENDER }} + EMAIL_PASSWORD: ${{ secrets.EMAIL_PASSWORD }} + EMAIL_RECEIVERS: ${{ secrets.EMAIL_RECEIVERS }} # 自选股列表 (从 secrets 或使用默认值) STOCK_LIST: ${{ secrets.STOCK_LIST || '600519' }} diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e7c52420..72dd5b10d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,9 +9,21 @@ ### 计划中 - 钉钉机器人支持 -- 飞书机器人支持 - Web 管理界面 +## [1.2.0] - 2026-01-11 + +### 新增 +- 📢 多渠道推送支持 + - 企业微信 Webhook + - 飞书 Webhook（新增） + - 邮件 SMTP（新增） + - 自动识别渠道类型，配置更简单 + +### 改进 +- 统一使用 `NOTIFICATION_URL` 配置，兼容旧的 `WECHAT_WEBHOOK_URL` +- 邮件支持 Markdown 转 HTML 渲染 + ## [1.1.0] - 2026-01-11 ### 新增 diff --git a/README.md b/README.md index eb8299d35..426d09712 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ [![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/) [![GitHub Actions](https://img.shields.io/badge/GitHub%20Actions-Ready-2088FF?logo=github-actions&logoColor=white)](https://github.com/features/actions) -> 🤖 基于 AI 大模型的 A 股自选股智能分析系统，每日自动分析并推送「决策仪表盘」到企业微信/钉钉/飞书 +> 🤖 基于 AI 大模型的 A 股自选股智能分析系统，每日自动分析并推送「决策仪表盘」到企业微信/飞书/Telegram/邮箱 ![运行效果演示](./sources/2026-01-10_155341_daily_analysis.gif) @@ -15,7 +15,7 @@ - **AI 决策仪表盘** - 一句话核心结论 + 精确买卖点位 + 检查清单 - **多维度分析** - 技术面 + 筹码分布 + 舆情情报 + 实时行情 - **大盘复盘** - 每日市场概览、板块涨跌、北向资金 -- **定时推送** - 支持企业微信机器人自动推送 +- **多渠道推送** - 支持企业微信、飞书、Telegram、邮件（自动识别） - **零成本部署** - GitHub Actions 免费运行，无需服务器 - **💰 白嫖 Gemini API** - Google AI Studio 提供免费额度，个人使用完全够用 - **🔄 多模型支持** - 支持 OpenAI 兼容 API（DeepSeek、通义千问等）作为备选 @@ -47,19 +47,38 @@ 进入你 Fork 的仓库 → `Settings` → `Secrets and variables` → `Actions` → `New repository secret` +**AI 模型配置（二选一）** + +| Secret 名称 | 说明 | 必填 | +|------------|------|:----:| +| `GEMINI_API_KEY` | [Google AI Studio](https://aistudio.google.com/) 获取免费 Key | ✅* | +| `OPENAI_API_KEY` | OpenAI 兼容 API Key（支持 DeepSeek、通义千问等） | 可选 | +| `OPENAI_BASE_URL` | OpenAI 兼容 API 地址（如 `https://api.deepseek.com/v1`） | 可选 | +| `OPENAI_MODEL` | 模型名称（如 `deepseek-chat`） | 可选 | + +> *注：`GEMINI_API_KEY` 和 `OPENAI_API_KEY` 至少配置一个 + +**通知渠道配置（可同时配置多个，全部推送）** + +| Secret 名称 | 说明 | 必填 | +|------------|------|:----:| +| `WECHAT_WEBHOOK_URL` | 企业微信 Webhook URL | 可选 | +| `FEISHU_WEBHOOK_URL` | 飞书 Webhook URL | 可选 | +| `TELEGRAM_BOT_TOKEN` | Telegram Bot Token（@BotFather 获取） | 可选 | +| `TELEGRAM_CHAT_ID` | Telegram Chat ID | 可选 | +| `EMAIL_SENDER` | 发件人邮箱（如 `xxx@qq.com`） | 可选 | +| `EMAIL_PASSWORD` | 邮箱授权码（非登录密码） | 可选 | +| `EMAIL_RECEIVERS` | 收件人邮箱（留空则发给自己） | 可选 | + +> *注：至少配置一个渠道，配置多个则同时推送到所有渠道 + +**其他配置** + | Secret 名称 | 说明 | 必填 | |------------|------|:----:| -| `GEMINI_API_KEY` | [Google AI Studio](https://aistudio.google.com/) 获取 | ✅* | -| `WECHAT_WEBHOOK_URL` | 企业微信机器人 Webhook | ✅ | | `STOCK_LIST` | 自选股代码，如 `600519,300750,002594` | ✅ | | `TAVILY_API_KEYS` | [Tavily](https://tavily.com/) 搜索 API（新闻搜索） | 推荐 | -| `SERPAPI_API_KEYS` | [SerpAPI](https://serpapi.com/) Key | 可选 | | `TUSHARE_TOKEN` | [Tushare Pro](https://tushare.pro/) Token | 可选 | -| `OPENAI_API_KEY` | OpenAI 兼容 API Key（备选） | 可选 | -| `OPENAI_BASE_URL` | OpenAI 兼容 API 地址 | 可选 | -| `OPENAI_MODEL` | OpenAI 兼容模型名称 | 可选 | - -> *注：`GEMINI_API_KEY` 和 `OPENAI_API_KEY` 至少配置一个，系统优先使用 Gemini #### 3. 启用 Actions @@ -208,22 +227,25 @@ daily_stock_analysis/ ### 🔔 通知渠道扩展 - [x] 企业微信机器人 +- [x] 飞书机器人 +- [x] Telegram Bot +- [x] 邮件通知（SMTP） - [ ] 钉钉机器人 -- [ ] 飞书机器人 -- [ ] Telegram Bot - [ ] Discord Webhook - [ ] Slack Webhook -- [ ] 邮件通知 - [ ] iOS/Android 推送（Bark/Pushover） ### 🤖 AI 模型支持 -- [x] Google Gemini -- [ ] OpenAI GPT-4 +- [x] Google Gemini（主力，免费额度） +- [x] OpenAI 兼容 API（支持以下模型） + - [x] OpenAI GPT-4/4o + - [x] DeepSeek + - [x] 通义千问 + - [x] Moonshot（月之暗面） + - [x] 智谱 GLM - [ ] Claude -- [ ] 通义千问 - [ ] 文心一言 -- [ ] DeepSeek -- [ ] 本地模型（Ollama） +- [x] 本地模型（Ollama） ### 📊 数据源扩展 - [x] AkShare（免费） diff --git a/config.py b/config.py index bf47713d2..412236355 100644 --- a/config.py +++ b/config.py @@ -53,9 +53,23 @@ class Config: tavily_api_keys: List[str] = field(default_factory=list) # Tavily API Keys serpapi_keys: List[str] = field(default_factory=list) # SerpAPI Keys - # === 通知配置 === + # === 通知配置（可同时配置多个，全部推送）=== + + # 企业微信 Webhook wechat_webhook_url: Optional[str] = None + # 飞书 Webhook + feishu_webhook_url: Optional[str] = None + + # Telegram 配置（需要同时配置 Bot Token 和 Chat ID） + telegram_bot_token: Optional[str] = None # Bot Token（@BotFather 获取） + telegram_chat_id: Optional[str] = None # Chat ID + + # 邮件配置（只需邮箱和授权码，SMTP 自动识别） + email_sender: Optional[str] = None # 发件人邮箱 + email_password: Optional[str] = None # 邮箱密码/授权码 + email_receivers: List[str] = field(default_factory=list) # 收件人列表（留空则发给自己） + # === 数据库配置 === database_path: str = "./data/stock_analysis.db" @@ -150,6 +164,12 @@ class Config: tavily_api_keys=tavily_api_keys, serpapi_keys=serpapi_keys, wechat_webhook_url=os.getenv('WECHAT_WEBHOOK_URL'), + feishu_webhook_url=os.getenv('FEISHU_WEBHOOK_URL'), + telegram_bot_token=os.getenv('TELEGRAM_BOT_TOKEN'), + telegram_chat_id=os.getenv('TELEGRAM_CHAT_ID'), + email_sender=os.getenv('EMAIL_SENDER'), + email_password=os.getenv('EMAIL_PASSWORD'), + email_receivers=[r.strip() for r in os.getenv('EMAIL_RECEIVERS', '').split(',') if r.strip()], database_path=os.getenv('DATABASE_PATH', './data/stock_analysis.db'), log_dir=os.getenv('LOG_DIR', './logs'), log_level=os.getenv('LOG_LEVEL', 'INFO'), @@ -188,8 +208,15 @@ class Config: if not self.tavily_api_keys and not self.serpapi_keys: warnings.append("提示：未配置搜索引擎 API Key (Tavily/SerpAPI)，新闻搜索功能将不可用") - if not self.wechat_webhook_url: - warnings.append("提示：未配置企业微信 Webhook，将不发送推送通知") + # 检查通知配置 + has_notification = ( + self.wechat_webhook_url or + self.feishu_webhook_url or + (self.telegram_bot_token and self.telegram_chat_id) or + (self.email_sender and self.email_password) + ) + if not has_notification: + warnings.append("提示：未配置通知渠道，将不发送推送通知") return warnings diff --git a/main.py b/main.py index 37b70b3b6..773e577d0 100644 --- a/main.py +++ b/main.py @@ -574,20 +574,20 @@ class StockAnalysisPipeline: filepath = self.notifier.save_report_to_file(report) logger.info(f"决策仪表盘日报已保存: {filepath}") - # 推送到企业微信（使用精简版决策仪表盘） + # 推送通知 if self.notifier.is_available(): - # 生成精简版决策仪表盘用于微信推送 - wechat_dashboard = self.notifier.generate_wechat_dashboard(results) - logger.info(f"微信决策仪表盘长度: {len(wechat_dashboard)} 字符") - logger.debug(f"微信推送内容:\n{wechat_dashboard}") + # 生成精简版决策仪表盘用于推送 + dashboard_content = self.notifier.generate_wechat_dashboard(results) + logger.info(f"决策仪表盘长度: {len(dashboard_content)} 字符") + logger.debug(f"推送内容:\n{dashboard_content}") - success = self.notifier.send_to_wechat(wechat_dashboard) + success = self.notifier.send(dashboard_content) if success: logger.info("决策仪表盘推送成功") else: logger.warning("决策仪表盘推送失败") else: - logger.info("企业微信未配置，跳过推送") + logger.info("通知渠道未配置，跳过推送") except Exception as e: logger.error(f"发送通知失败: {e}") @@ -686,14 +686,12 @@ def run_market_review(notifier: NotificationService, analyzer=None, search_servi review_report = market_analyzer.run_daily_review() if review_report: - # 推送到微信 + # 推送通知 if notifier.is_available(): # 添加标题 - wechat_report = f"## 🎯 大盘复盘\n\n{review_report}" - if len(wechat_report) > 3800: - wechat_report = wechat_report[:3800] + "\n...(已截断)" + report_content = f"🎯 大盘复盘\n\n{review_report}" - success = notifier.send_to_wechat(wechat_report) + success = notifier.send(report_content) if success: logger.info("大盘复盘推送成功") else: @@ -791,7 +789,7 @@ def main() -> int: # 模式1: 仅大盘复盘 if args.market_review: logger.info("模式: 仅大盘复盘") - notifier = NotificationService(config.wechat_webhook_url) + notifier = NotificationService() # 初始化搜索服务和分析器（如果有配置） search_service = None diff --git a/notification.py b/notification.py index 49c453294..d04e0adf1 100644 --- a/notification.py +++ b/notification.py @@ -7,12 +7,22 @@ A股自选股智能分析系统 - 通知层 职责： 1. 汇总分析结果生成日报 2. 支持 Markdown 格式输出 -3. 推送到企业微信 Webhook +3. 多渠道推送（自动识别）： + - 企业微信 Webhook + - 飞书 Webhook + - Telegram Bot + - 邮件 SMTP """ import logging +import smtplib +import re from datetime import datetime from typing import List, Dict, Any, Optional +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from email.header import Header +from enum import Enum import requests @@ -22,31 +32,157 @@ from analyzer import AnalysisResult logger = logging.getLogger(__name__) +class NotificationChannel(Enum): + """通知渠道类型""" + WECHAT = "wechat" # 企业微信 + FEISHU = "feishu" # 飞书 + TELEGRAM = "telegram" # Telegram + EMAIL = "email" # 邮件 + UNKNOWN = "unknown" # 未知 + + +# SMTP 服务器配置（自动识别） +SMTP_CONFIGS = { + # QQ邮箱 + "qq.com": {"server": "smtp.qq.com", "port": 465, "ssl": True}, + # 网易邮箱 + "163.com": {"server": "smtp.163.com", "port": 465, "ssl": True}, + "126.com": {"server": "smtp.126.com", "port": 465, "ssl": True}, + # Gmail + "gmail.com": {"server": "smtp.gmail.com", "port": 587, "ssl": False}, + # Outlook + "outlook.com": {"server": "smtp-mail.outlook.com", "port": 587, "ssl": False}, + "hotmail.com": {"server": "smtp-mail.outlook.com", "port": 587, "ssl": False}, + "live.com": {"server": "smtp-mail.outlook.com", "port": 587, "ssl": False}, + # 新浪 + "sina.com": {"server": "smtp.sina.com", "port": 465, "ssl": True}, + # 搜狐 + "sohu.com": {"server": "smtp.sohu.com", "port": 465, "ssl": True}, + # 阿里云 + "aliyun.com": {"server": "smtp.aliyun.com", "port": 465, "ssl": True}, + # 139邮箱 + "139.com": {"server": "smtp.139.com", "port": 465, "ssl": True}, +} + + +class ChannelDetector: + """ + 渠道检测器 - 简化版 + + 根据配置直接判断渠道类型（不再需要 URL 解析） + """ + + @staticmethod + def get_channel_name(channel: NotificationChannel) -> str: + """获取渠道中文名称""" + names = { + NotificationChannel.WECHAT: "企业微信", + NotificationChannel.FEISHU: "飞书", + NotificationChannel.TELEGRAM: "Telegram", + NotificationChannel.EMAIL: "邮件", + NotificationChannel.UNKNOWN: "未知渠道", + } + return names.get(channel, "未知渠道") + + class NotificationService: """ 通知服务 职责： 1. 生成 Markdown 格式的分析日报 - 2. 推送消息到企业微信机器人 + 2. 向所有已配置的渠道推送消息（多渠道并发） 3. 支持本地保存日报 + + 支持的渠道： + - 企业微信 Webhook + - 飞书 Webhook + - Telegram Bot + - 邮件 SMTP + + 注意：所有已配置的渠道都会收到推送 """ - def __init__(self, webhook_url: Optional[str] = None): + def __init__(self): """ 初始化通知服务 - Args: - webhook_url: 企业微信 Webhook URL（可选，默认从配置读取） + 检测所有已配置的渠道，推送时会向所有渠道发送 """ - self._webhook_url = webhook_url or get_config().wechat_webhook_url + config = get_config() - if not self._webhook_url: - logger.warning("企业微信 Webhook URL 未配置，将不发送推送通知") + # 各渠道的 Webhook URL + self._wechat_url = config.wechat_webhook_url + self._feishu_url = getattr(config, 'feishu_webhook_url', None) + + # Telegram 配置 + self._telegram_config = { + 'bot_token': getattr(config, 'telegram_bot_token', None), + 'chat_id': getattr(config, 'telegram_chat_id', None), + } + + # 邮件配置 + self._email_config = { + 'sender': config.email_sender, + 'password': config.email_password, + 'receivers': config.email_receivers or ([config.email_sender] if config.email_sender else []), + } + + # 检测所有已配置的渠道 + self._available_channels = self._detect_all_channels() + + if not self._available_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)}") + + def _detect_all_channels(self) -> List[NotificationChannel]: + """ + 检测所有已配置的渠道 + + Returns: + 已配置的渠道列表 + """ + channels = [] + + # 企业微信 + if self._wechat_url: + channels.append(NotificationChannel.WECHAT) + + # 飞书 + if self._feishu_url: + channels.append(NotificationChannel.FEISHU) + + # Telegram + if self._is_telegram_configured(): + channels.append(NotificationChannel.TELEGRAM) + + # 邮件 + if self._is_email_configured(): + channels.append(NotificationChannel.EMAIL) + + return channels + + def _is_telegram_configured(self) -> bool: + """检查 Telegram 配置是否完整""" + return bool(self._telegram_config['bot_token'] and self._telegram_config['chat_id']) + + def _is_email_configured(self) -> bool: + """检查邮件配置是否完整（只需邮箱和授权码）""" + return bool(self._email_config['sender'] and self._email_config['password']) def is_available(self) -> bool: - """检查通知服务是否可用""" - return bool(self._webhook_url) + """检查通知服务是否可用（至少有一个渠道）""" + return len(self._available_channels) > 0 + + def get_available_channels(self) -> List[NotificationChannel]: + """获取所有已配置的渠道""" + return self._available_channels + + def get_channel_names(self) -> str: + """获取所有已配置渠道的名称""" + return ', '.join([ChannelDetector.get_channel_name(ch) for ch in self._available_channels]) def generate_daily_report( self, @@ -761,7 +897,7 @@ class NotificationService: Returns: 是否发送成功 """ - if not self.is_available(): + if not self._wechat_url: logger.warning("企业微信 Webhook 未配置，跳过推送") return False @@ -771,13 +907,13 @@ class NotificationService: content = content[:3950] + "\n\n...(内容过长已截断，详见完整报告)" try: - return self._send_single_message(content) + return self._send_wechat_message(content) except Exception as e: logger.error(f"发送企业微信消息失败: {e}") return False - def _send_single_message(self, content: str) -> bool: - """发送单条消息""" + def _send_wechat_message(self, content: str) -> bool: + """发送企业微信消息""" payload = { "msgtype": "markdown", "markdown": { @@ -786,7 +922,7 @@ class NotificationService: } response = requests.post( - self._webhook_url, + self._wechat_url, json=payload, timeout=10 ) @@ -803,6 +939,418 @@ class NotificationService: logger.error(f"企业微信请求失败: {response.status_code}") return False + def send_to_feishu(self, content: str) -> bool: + """ + 推送消息到飞书机器人 + + 飞书自定义机器人 Webhook 消息格式： + { + "msg_type": "text", + "content": { + "text": "文本内容" + } + } + + 注意：飞书文本消息无严格长度限制，但建议控制在合理范围 + + Args: + content: 消息内容（Markdown 会转为纯文本） + + Returns: + 是否发送成功 + """ + if not self._feishu_url: + logger.warning("飞书 Webhook 未配置，跳过推送") + return False + + try: + # 飞书自定义机器人的消息格式 + # 支持 text 和 post（富文本）两种类型 + # 使用 post 富文本可以支持更好的格式显示 + + # 将 Markdown 转换为飞书 post 格式 + # 简化处理：使用 text 类型，保持原有格式 + payload = { + "msg_type": "text", + "content": { + "text": content + } + } + + logger.debug(f"飞书请求 URL: {self._feishu_url}") + logger.debug(f"飞书请求 payload: {payload}") + + response = requests.post( + self._feishu_url, + json=payload, + timeout=10 + ) + + logger.debug(f"飞书响应状态码: {response.status_code}") + logger.debug(f"飞书响应内容: {response.text}") + + if response.status_code == 200: + result = response.json() + # 飞书成功响应： + # - 新版: {"code": 0, "msg": "success"} + # - 旧版: {"StatusCode": 0, "StatusMessage": "success"} + code = result.get('code') if 'code' in result else result.get('StatusCode') + if code == 0: + logger.info("飞书消息发送成功") + return True + else: + error_msg = result.get('msg') or result.get('StatusMessage', '未知错误') + error_code = result.get('code') or result.get('StatusCode', 'N/A') + logger.error(f"飞书返回错误 [code={error_code}]: {error_msg}") + logger.error(f"完整响应: {result}") + return False + else: + logger.error(f"飞书请求失败: HTTP {response.status_code}") + logger.error(f"响应内容: {response.text}") + return False + + except Exception as e: + logger.error(f"发送飞书消息失败: {e}") + import traceback + logger.debug(traceback.format_exc()) + return False + + def send_to_email(self, content: str, subject: Optional[str] = None) -> bool: + """ + 通过 SMTP 发送邮件（自动识别 SMTP 服务器） + + Args: + content: 邮件内容（支持 Markdown，会转换为 HTML） + subject: 邮件主题（可选，默认自动生成） + + Returns: + 是否发送成功 + """ + if not self._is_email_configured(): + logger.warning("邮件配置不完整，跳过推送") + return False + + sender = self._email_config['sender'] + password = self._email_config['password'] + receivers = self._email_config['receivers'] + + try: + # 生成主题 + if subject is None: + date_str = datetime.now().strftime('%Y-%m-%d') + subject = f"📈 A股智能分析报告 - {date_str}" + + # 将 Markdown 转换为简单 HTML + html_content = self._markdown_to_html(content) + + # 构建邮件 + msg = MIMEMultipart('alternative') + msg['Subject'] = Header(subject, 'utf-8') + msg['From'] = sender + msg['To'] = ', '.join(receivers) + + # 添加纯文本和 HTML 两个版本 + text_part = MIMEText(content, 'plain', 'utf-8') + html_part = MIMEText(html_content, 'html', 'utf-8') + msg.attach(text_part) + msg.attach(html_part) + + # 自动识别 SMTP 配置 + domain = sender.split('@')[-1].lower() + smtp_config = SMTP_CONFIGS.get(domain) + + if smtp_config: + smtp_server = smtp_config['server'] + smtp_port = smtp_config['port'] + use_ssl = smtp_config['ssl'] + logger.info(f"自动识别邮箱类型: {domain} -> {smtp_server}:{smtp_port}") + else: + # 未知邮箱，尝试通用配置 + smtp_server = f"smtp.{domain}" + smtp_port = 465 + use_ssl = True + logger.warning(f"未知邮箱类型 {domain}，尝试通用配置: {smtp_server}:{smtp_port}") + + # 根据配置选择连接方式 + if use_ssl: + # SSL 连接（端口 465） + server = smtplib.SMTP_SSL(smtp_server, smtp_port, timeout=30) + else: + # TLS 连接（端口 587） + server = smtplib.SMTP(smtp_server, smtp_port, timeout=30) + server.starttls() + + server.login(sender, password) + server.send_message(msg) + server.quit() + + logger.info(f"邮件发送成功，收件人: {receivers}") + return True + + except smtplib.SMTPAuthenticationError: + logger.error("邮件发送失败：认证错误，请检查邮箱和授权码是否正确") + return False + except smtplib.SMTPConnectError as e: + logger.error(f"邮件发送失败：无法连接 SMTP 服务器 - {e}") + return False + except Exception as e: + logger.error(f"发送邮件失败: {e}") + return False + + def _markdown_to_html(self, markdown_text: str) -> str: + """ + 将 Markdown 转换为简单的 HTML + + 支持：标题、加粗、列表、分隔线 + """ + html = markdown_text + + # 转义 HTML 特殊字符 + html = html.replace('&', '&') + html = html.replace('<', '<') + html = html.replace('>', '>') + + # 标题 (# ## ###) + html = re.sub(r'^### (.+)$', r'

\1

', html, flags=re.MULTILINE) + html = re.sub(r'^## (.+)$', r'

\1

', html, flags=re.MULTILINE) + html = re.sub(r'^# (.+)$', r'

\1

', html, flags=re.MULTILINE) + + # 加粗 **text** + html = re.sub(r'\*\*(.+?)\*\*', r'\1', html) + + # 斜体 *text* + html = re.sub(r'\*(.+?)\*', r'\1', html) + + # 分隔线 --- + html = re.sub(r'^---$', r'
', html, flags=re.MULTILINE) + + # 列表项 - item + html = re.sub(r'^- (.+)$', r'
  • \1
    • ', html, flags=re.MULTILINE) + + # 引用 > text + html = re.sub(r'^> (.+)$', r'
    \1
    ', html, flags=re.MULTILINE) + + # 换行 + html = html.replace('\n', '
    \n') + + # 包装 HTML + return f""" + + + + + + + + {html} + + + """ + + def send_to_telegram(self, content: str) -> bool: + """ + 推送消息到 Telegram 机器人 + + Telegram Bot API 格式： + POST https://api.telegram.org/bot/sendMessage + { + "chat_id": "xxx", + "text": "消息内容", + "parse_mode": "Markdown" + } + + Args: + content: 消息内容（Markdown 格式） + + Returns: + 是否发送成功 + """ + if not self._is_telegram_configured(): + logger.warning("Telegram 配置不完整，跳过推送") + return False + + bot_token = self._telegram_config['bot_token'] + chat_id = self._telegram_config['chat_id'] + + try: + # Telegram API 端点 + api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage" + + # Telegram 消息最大长度 4096 字符 + max_length = 4096 + + if len(content) <= max_length: + # 单条消息发送 + return self._send_telegram_message(api_url, chat_id, content) + else: + # 分段发送长消息 + return self._send_telegram_chunked(api_url, chat_id, content, max_length) + + except Exception as e: + logger.error(f"发送 Telegram 消息失败: {e}") + import traceback + logger.debug(traceback.format_exc()) + return False + + def _send_telegram_message(self, api_url: str, chat_id: str, text: str) -> bool: + """发送单条 Telegram 消息""" + # 转换 Markdown 为 Telegram 支持的格式 + # Telegram 的 Markdown 格式稍有不同，做简单处理 + telegram_text = self._convert_to_telegram_markdown(text) + + payload = { + "chat_id": chat_id, + "text": telegram_text, + "parse_mode": "Markdown", + "disable_web_page_preview": True + } + + response = requests.post(api_url, json=payload, timeout=10) + + if response.status_code == 200: + result = response.json() + if result.get('ok'): + logger.info("Telegram 消息发送成功") + return True + else: + error_desc = result.get('description', '未知错误') + logger.error(f"Telegram 返回错误: {error_desc}") + + # 如果 Markdown 解析失败，尝试纯文本发送 + if 'parse' in error_desc.lower() or 'markdown' in error_desc.lower(): + logger.info("尝试使用纯文本格式重新发送...") + payload['parse_mode'] = None + payload['text'] = text # 使用原始文本 + del payload['parse_mode'] + + response = requests.post(api_url, json=payload, timeout=10) + if response.status_code == 200 and response.json().get('ok'): + logger.info("Telegram 消息发送成功（纯文本）") + return True + + return False + else: + logger.error(f"Telegram 请求失败: HTTP {response.status_code}") + logger.error(f"响应内容: {response.text}") + return False + + def _send_telegram_chunked(self, api_url: str, chat_id: str, content: str, max_length: int) -> bool: + """分段发送长 Telegram 消息""" + # 按段落分割 + sections = content.split("\n---\n") + + current_chunk = [] + current_length = 0 + all_success = True + chunk_index = 1 + + for section in sections: + section_length = len(section) + 5 # +5 for "\n---\n" + + if current_length + section_length > max_length: + # 发送当前块 + if current_chunk: + chunk_content = "\n---\n".join(current_chunk) + logger.info(f"发送 Telegram 消息块 {chunk_index}...") + if not self._send_telegram_message(api_url, chat_id, chunk_content): + all_success = False + chunk_index += 1 + + # 重置 + current_chunk = [section] + current_length = section_length + else: + current_chunk.append(section) + current_length += section_length + + # 发送最后一块 + if current_chunk: + chunk_content = "\n---\n".join(current_chunk) + logger.info(f"发送 Telegram 消息块 {chunk_index}（最后）...") + if not self._send_telegram_message(api_url, chat_id, chunk_content): + all_success = False + + return all_success + + def _convert_to_telegram_markdown(self, text: str) -> str: + """ + 将标准 Markdown 转换为 Telegram 支持的格式 + + Telegram Markdown 限制： + - 不支持 # 标题 + - 使用 *bold* 而非 **bold** + - 使用 _italic_ + """ + result = text + + # 移除 # 标题标记（Telegram 不支持） + result = re.sub(r'^#{1,6}\s+', '', result, flags=re.MULTILINE) + + # 转换 **bold** 为 *bold* + result = re.sub(r'\*\*(.+?)\*\*', r'*\1*', result) + + # 转义特殊字符（Telegram Markdown 需要） + # 注意：不转义已经用于格式的 * _ ` + for char in ['[', ']', '(', ')']: + result = result.replace(char, f'\\{char}') + + return result + + def send(self, content: str) -> bool: + """ + 统一发送接口 - 向所有已配置的渠道发送 + + 遍历所有已配置的渠道，逐一发送消息 + + Args: + content: 消息内容（Markdown 格式） + + Returns: + 是否至少有一个渠道发送成功 + """ + if not self.is_available(): + logger.warning("通知服务不可用，跳过推送") + return False + + channel_names = self.get_channel_names() + logger.info(f"正在向 {len(self._available_channels)} 个渠道发送通知：{channel_names}") + + success_count = 0 + fail_count = 0 + + for channel in self._available_channels: + channel_name = ChannelDetector.get_channel_name(channel) + try: + if channel == NotificationChannel.WECHAT: + result = self.send_to_wechat(content) + elif channel == NotificationChannel.FEISHU: + result = self.send_to_feishu(content) + elif channel == NotificationChannel.TELEGRAM: + result = self.send_to_telegram(content) + elif channel == NotificationChannel.EMAIL: + result = self.send_to_email(content) + else: + logger.warning(f"不支持的通知渠道: {channel}") + result = False + + if result: + success_count += 1 + else: + fail_count += 1 + + except Exception as e: + logger.error(f"{channel_name} 发送失败: {e}") + fail_count += 1 + + logger.info(f"通知发送完成：成功 {success_count} 个，失败 {fail_count} 个") + return success_count > 0 + def _send_chunked_messages(self, content: str, max_length: int) -> bool: """ 分段发送长消息 @@ -826,7 +1374,7 @@ class NotificationService: if current_chunk: chunk_content = "\n---\n".join(current_chunk) logger.info(f"发送消息块 {chunk_index}...") - if not self._send_single_message(chunk_content): + if not self.send(chunk_content): all_success = False chunk_index += 1 @@ -841,7 +1389,7 @@ class NotificationService: if current_chunk: chunk_content = "\n---\n".join(current_chunk) logger.info(f"发送消息块 {chunk_index}（最后）...") - if not self._send_single_message(chunk_content): + if not self.send(chunk_content): all_success = False return all_success @@ -937,7 +1485,7 @@ def send_daily_report(results: List[AnalysisResult]) -> bool: """ 发送每日报告的快捷方式 - 自动生成报告并推送到企业微信 + 自动识别渠道并推送 """ service = get_notification_service() @@ -947,8 +1495,8 @@ def send_daily_report(results: List[AnalysisResult]) -> bool: # 保存到本地 service.save_report_to_file(report) - # 推送到企业微信 - return service.send_to_wechat(report) + # 推送到配置的渠道（自动识别） + return service.send(report) if __name__ == "__main__": @@ -991,8 +1539,14 @@ if __name__ == "__main__": service = NotificationService() + # 显示检测到的渠道 + print(f"=== 通知渠道检测 ===") + print(f"当前渠道: {service.get_channel_name()}") + print(f"渠道类型: {service.get_channel()}") + print(f"服务可用: {service.is_available()}") + # 生成日报 - print("=== 生成日报测试 ===") + print("\n=== 生成日报测试 ===") report = service.generate_daily_report(test_results) print(report) @@ -1001,10 +1555,10 @@ if __name__ == "__main__": filepath = service.save_report_to_file(report) print(f"保存成功: {filepath}") - # 推送测试（仅当配置了 Webhook 时） + # 推送测试 if service.is_available(): - print("\n=== 推送测试 ===") - success = service.send_to_wechat(report) + print(f"\n=== 推送测试（{service.get_channel_name()}）===") + success = service.send(report) print(f"推送结果: {'成功' if success else '失败'}") else: - print("\n企业微信 Webhook 未配置，跳过推送测试") + print("\n通知渠道未配置，跳过推送测试")