fanhuadesenlinnn/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis synced 2026-09-20 10:53:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: Support iOS/Android push notification（Pushover）

Browse Source
This commit is contained in:
Krane
2026-01-14 18:21:37 +08:00
parent b1bfea5dad
commit 253acb296f
3 changed files with 244 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.env.example
View File

@@ -85,7 +85,12 @@ SERPAPI_API_KEYS=your_serpapi_key_here
# 系统会自动识别常见服务并使用对应格式
#
# CUSTOM_WEBHOOK_URLS=https://oapi.dingtalk.com/robot/send?access_token=xxx,https://hooks.slack.com/services/xxx
#
# 【方式六】Pushover 配置
# 注册Pushover账号，创建应用Token https://pushover.net/apps/build
# PUSHOVER_USER_KEY=your_user_key
# PUSHOVER_API_TOKEN=your_api_token
#
# 【高级配置】消息长度限制（字节）
# 超过限制会自动分批发送，一般无需修改
# FEISHU_MAX_BYTES=20000 # 飞书限制约 20KB，默认 20000 字节

9
config.py
View File

@@ -75,6 +75,10 @@ class Config:
email_password: Optional[str] = None # 邮箱密码/授权码
email_receivers: List[str] = field(default_factory=list) # 收件人列表（留空则发给自己）
# Pushover 配置（手机/桌面推送通知）
pushover_user_key: Optional[str] = None # 用户 Key（https://pushover.net 获取）
pushover_api_token: Optional[str] = None # 应用 API Token
# 自定义 Webhook（支持多个，逗号分隔）
# 适用于：钉钉、Discord、Slack、自建服务等任意支持 POST JSON 的 Webhook
custom_webhook_urls: List[str] = field(default_factory=list)
@@ -186,6 +190,8 @@ class Config:
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()],
pushover_user_key=os.getenv('PUSHOVER_USER_KEY'),
pushover_api_token=os.getenv('PUSHOVER_API_TOKEN'),
custom_webhook_urls=[u.strip() for u in os.getenv('CUSTOM_WEBHOOK_URLS', '').split(',') if u.strip()],
feishu_max_bytes=int(os.getenv('FEISHU_MAX_BYTES', '20000')),
wechat_max_bytes=int(os.getenv('WECHAT_MAX_BYTES', '4000')),
@@ -232,7 +238,8 @@ class Config:
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)
(self.email_sender and self.email_password) or
(self.pushover_user_key and self.pushover_api_token)
)
if not has_notification:
warnings.append("提示：未配置通知渠道，将不发送推送通知")

230
notification.py
View File

@@ -12,6 +12,7 @@ A股自选股智能分析系统 - 通知层
- 飞书 Webhook
- Telegram Bot
- 邮件 SMTP
- Pushover（手机/桌面推送）
"""
import logging
@@ -38,6 +39,7 @@ class NotificationChannel(Enum):
FEISHU = "feishu" # 飞书
TELEGRAM = "telegram" # Telegram
EMAIL = "email" # 邮件
PUSHOVER = "pushover" # Pushover（手机/桌面推送）
CUSTOM = "custom" # 自定义 Webhook
UNKNOWN = "unknown" # 未知
@@ -81,6 +83,7 @@ class ChannelDetector:
NotificationChannel.FEISHU: "飞书",
NotificationChannel.TELEGRAM: "Telegram",
NotificationChannel.EMAIL: "邮件",
NotificationChannel.PUSHOVER: "Pushover",
NotificationChannel.CUSTOM: "自定义Webhook",
NotificationChannel.UNKNOWN: "未知渠道",
}
@@ -101,6 +104,7 @@ class NotificationService:
- 飞书 Webhook
- Telegram Bot
- 邮件 SMTP
- Pushover（手机/桌面推送）
注意：所有已配置的渠道都会收到推送
"""
@@ -130,6 +134,12 @@ class NotificationService:
'receivers': config.email_receivers or ([config.email_sender] if config.email_sender else []),
}
# Pushover 配置
self._pushover_config = {
'user_key': getattr(config, 'pushover_user_key', None),
'api_token': getattr(config, 'pushover_api_token', None),
}
# 自定义 Webhook 配置
self._custom_webhook_urls = getattr(config, 'custom_webhook_urls', []) or []
@@ -171,6 +181,10 @@ class NotificationService:
if self._is_email_configured():
channels.append(NotificationChannel.EMAIL)
# Pushover
if self._is_pushover_configured():
channels.append(NotificationChannel.PUSHOVER)
# 自定义 Webhook
if self._custom_webhook_urls:
channels.append(NotificationChannel.CUSTOM)
@@ -185,6 +199,10 @@ class NotificationService:
"""检查邮件配置是否完整（只需邮箱和授权码）"""
return bool(self._email_config['sender'] and self._email_config['password'])
def _is_pushover_configured(self) -> bool:
"""检查 Pushover 配置是否完整"""
return bool(self._pushover_config['user_key'] and self._pushover_config['api_token'])
def is_available(self) -> bool:
"""检查通知服务是否可用（至少有一个渠道）"""
return len(self._available_channels) > 0
@@ -1637,6 +1655,216 @@ class NotificationService:
return result
def send_to_pushover(self, content: str, title: Optional[str] = None) -> bool:
"""
推送消息到 Pushover
Pushover API 格式：
POST https://api.pushover.net/1/messages.json
{
"token": "应用 API Token",
"user": "用户 Key",
"message": "消息内容",
"title": "标题（可选）"
}
Pushover 特点：
- 支持 iOS/Android/桌面多平台推送
- 消息限制 1024 字符
- 支持优先级设置
- 支持 HTML 格式
Args:
content: 消息内容（Markdown 格式，会转为纯文本）
title: 消息标题（可选，默认为"股票分析报告"）
Returns:
是否发送成功
"""
if not self._is_pushover_configured():
logger.warning("Pushover 配置不完整，跳过推送")
return False
user_key = self._pushover_config['user_key']
api_token = self._pushover_config['api_token']
# Pushover API 端点
api_url = "https://api.pushover.net/1/messages.json"
# 处理消息标题
if title is None:
date_str = datetime.now().strftime('%Y-%m-%d')
title = f"📈 股票分析报告 - {date_str}"
# Pushover 消息限制 1024 字符
max_length = 1024
# 转换 Markdown 为纯文本（Pushover 支持 HTML，但纯文本更通用）
plain_content = self._markdown_to_plain_text(content)
if len(plain_content) <= max_length:
# 单条消息发送
return self._send_pushover_message(api_url, user_key, api_token, plain_content, title)
else:
# 分段发送长消息
return self._send_pushover_chunked(api_url, user_key, api_token, plain_content, title, max_length)
def _markdown_to_plain_text(self, markdown_text: str) -> str:
"""
将 Markdown 转换为纯文本
移除 Markdown 格式标记，保留可读性
"""
text = markdown_text
# 移除标题标记 # ## ###
text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
# 移除加粗 **text** -> text
text = re.sub(r'\*\*(.+?)\*\*', r'\1', text)
# 移除斜体 *text* -> text
text = re.sub(r'\*(.+?)\*', r'\1', text)
# 移除引用 > text -> text
text = re.sub(r'^>\s+', '', text, flags=re.MULTILINE)
# 移除列表标记 - item -> item
text = re.sub(r'^[-*]\s+', '• ', text, flags=re.MULTILINE)
# 移除分隔线 ---
text = re.sub(r'^---+$', '────────', text, flags=re.MULTILINE)
# 移除表格语法 |---|---|
text = re.sub(r'\|[-:]+\|[-:|\s]+\|', '', text)
text = re.sub(r'^\|(.+)\|$', r'\1', text, flags=re.MULTILINE)
# 清理多余空行
text = re.sub(r'\n{3,}', '\n\n', text)
return text.strip()
def _send_pushover_message(
self,
api_url: str,
user_key: str,
api_token: str,
message: str,
title: str,
priority: int = 0
) -> bool:
"""
发送单条 Pushover 消息
Args:
api_url: Pushover API 端点
user_key: 用户 Key
api_token: 应用 API Token
message: 消息内容
title: 消息标题
priority: 优先级 (-2 ~ 2，默认 0)
"""
try:
payload = {
"token": api_token,
"user": user_key,
"message": message,
"title": title,
"priority": priority,
}
response = requests.post(api_url, data=payload, timeout=30)
if response.status_code == 200:
result = response.json()
if result.get('status') == 1:
logger.info("Pushover 消息发送成功")
return True
else:
errors = result.get('errors', ['未知错误'])
logger.error(f"Pushover 返回错误: {errors}")
return False
else:
logger.error(f"Pushover 请求失败: HTTP {response.status_code}")
logger.debug(f"响应内容: {response.text}")
return False
except Exception as e:
logger.error(f"发送 Pushover 消息失败: {e}")
return False
def _send_pushover_chunked(
self,
api_url: str,
user_key: str,
api_token: str,
content: str,
title: str,
max_length: int
) -> bool:
"""
分段发送长 Pushover 消息
按段落分割，确保每段不超过最大长度
"""
import time
# 按段落（分隔线或双换行）分割
if "────────" in content:
sections = content.split("────────")
separator = "────────"
else:
sections = content.split("\n\n")
separator = "\n\n"
chunks = []
current_chunk = []
current_length = 0
for section in sections:
# 计算添加这个 section 后的实际长度
# join() 只在元素之间放置分隔符，不是每个元素后面
# 所以：第一个元素不需要分隔符，后续元素需要一个分隔符连接
if current_chunk:
# 已有元素，添加新元素需要：当前长度 + 分隔符 + 新 section
new_length = current_length + len(separator) + len(section)
else:
# 第一个元素，不需要分隔符
new_length = len(section)
if new_length > max_length:
if current_chunk:
chunks.append(separator.join(current_chunk))
current_chunk = [section]
current_length = len(section)
else:
current_chunk.append(section)
current_length = new_length
if current_chunk:
chunks.append(separator.join(current_chunk))
total_chunks = len(chunks)
success_count = 0
logger.info(f"Pushover 分批发送：共 {total_chunks} 批")
for i, chunk in enumerate(chunks):
# 添加分页标记到标题
chunk_title = f"{title} ({i+1}/{total_chunks})" if total_chunks > 1 else title
if self._send_pushover_message(api_url, user_key, api_token, chunk, chunk_title):
success_count += 1
logger.info(f"Pushover 第 {i+1}/{total_chunks} 批发送成功")
else:
logger.error(f"Pushover 第 {i+1}/{total_chunks} 批发送失败")
# 批次间隔，避免触发频率限制
if i < total_chunks - 1:
time.sleep(1)
return success_count == total_chunks
def send_to_custom(self, content: str) -> bool:
"""
推送消息到自定义 Webhook
@@ -1780,6 +2008,8 @@ class NotificationService:
result = self.send_to_telegram(content)
elif channel == NotificationChannel.EMAIL:
result = self.send_to_email(content)
elif channel == NotificationChannel.PUSHOVER:
result = self.send_to_pushover(content)
elif channel == NotificationChannel.CUSTOM:
result = self.send_to_custom(content)
else: