Conversation
- 新增告警管理器模块 (alert_manager.py) - 支持邮件、飞书、企业微信、钉钉、Telegram多种通知渠道 - 添加告警规则配置系统 - 集成到主程序，自动检查告警 - 添加独立的告警检查脚本 - 支持冷却时间和每日告警限制 - 保存告警历史记录，避免重复通知 Closes #8
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! 此拉取请求引入了一个全面的告警通知功能，使用户能够根据基金或指数的涨跌幅设置自定义阈值。当市场数据达到预设条件时，系统将通过多种配置的渠道（如邮件、飞书、企业微信、钉钉、Telegram）自动发送通知。这极大地增强了对市场波动的实时监控能力，帮助用户及时获取关键信息并作出响应。 Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
| def send_daily_summary(self, fund_data: Dict[str, Dict], triggered_alerts: List[Dict[str, Any]]) -> bool: | ||
| """ | ||
| 发送每日汇总报告 | ||
|
|
||
| Args: | ||
| fund_data: 所有基金数据 | ||
| triggered_alerts: 今日触发的告警列表 | ||
|
|
||
| Returns: | ||
| bool: 是否发送成功 | ||
| """ | ||
| if not self.global_settings.get('enable_daily_summary', False): | ||
| return False | ||
|
|
||
| # 检查是否应该发送每日汇总 | ||
| summary_time = self.global_settings.get('summary_time', '18:00') | ||
| current_time = datetime.now().strftime('%H:%M') | ||
|
|
||
| if current_time != summary_time: | ||
| return False | ||
|
|
||
| # 创建汇总消息 | ||
| total_funds = len(fund_data) | ||
| total_alerts = len(triggered_alerts) | ||
|
|
||
| summary_message = { | ||
| "type": "daily_summary", | ||
| "date": datetime.now().strftime('%Y-%m-%d'), | ||
| "total_funds": total_funds, | ||
| "total_alerts": total_alerts, | ||
| "triggered_alerts": triggered_alerts, | ||
| "timestamp": datetime.now().isoformat() | ||
| } | ||
|
|
||
| # 发送到所有启用的渠道 | ||
| success = False | ||
| for channel_name, channel in self.channels.items(): | ||
| if channel.enabled: | ||
| # 这里需要为每个渠道实现汇总消息的发送 | ||
| # 暂时只记录日志 | ||
| logger.info(f"发送每日汇总到 {channel_name}") | ||
|
|
||
| return success |
There was a problem hiding this comment.
这个每日汇总功能目前存在一些问题：
- 时间检查逻辑非常脆弱：
current_time != summary_time这个判断要求脚本在summary_time指定的那个确切分钟内执行，否则永远不会发送汇总。这在实际部署中很容易失败。建议修改为检查当前时间是否已超过汇总时间，并且今天尚未发送过汇总（这需要一个机制来记录每日汇总的发送状态）。 - 功能未完成：目前该函数只是打印日志信息，并没有真正实现发送逻辑。
- 返回值不正确：函数总是返回
False，因为success变量从未被修改。
由于该功能尚不完整且存在逻辑问题，建议在完善前暂时禁用或移除该功能。
| from email.mime.text import MIMEText | ||
| from email.mime.multipart import MIMEMultipart | ||
| from datetime import datetime, timedelta | ||
| from typing import Dict, List, Optional, Any, Tuple |
There was a problem hiding this comment.
此行导入的 Optional 和 Tuple 在文件中并未使用，建议移除以保持代码整洁。
| from typing import Dict, List, Optional, Any, Tuple | |
| from typing import Dict, List, Any |
| from datetime import datetime, timedelta | ||
| from typing import Dict, List, Optional, Any, Tuple | ||
| import requests | ||
| from pathlib import Path |
There was a problem hiding this comment.
此行导入的 Path 在文件中并未使用，建议移除以保持代码整洁。
| def __init__(self, config: Dict[str, Any]): | ||
| super().__init__(config) | ||
| self.webhook_url = config.get('webhook_url', '') | ||
| self.key = config.get('key', '') |
There was a problem hiding this comment.
在 WechatWorkNotification 中，从配置中获取的 key 属性在 send 方法中并未使用。根据企业微信的文档，机器人的 key 是作为 webhook_url 的一部分。这使得配置文件中的 key 字段变得多余且令人困惑。建议移除此行，并指导用户在 webhook_url 中提供完整的 URL。
| import hmac | ||
| import hashlib | ||
| import base64 | ||
| import urllib.parse |
There was a problem hiding this comment.
为了保持代码风格的统一和清晰，建议将 hmac, hashlib, base64, urllib.parse 这些标准库的 import 语句移到文件的顶部。将所有导入都放在文件开头是 Python 的通用惯例 (PEP 8)，这样可以更容易地查看模块的依赖关系。在方法内部导入通常只在为了解决循环导入或可选依赖等特殊情况下使用，这里似乎没有这个必要。
| def _load_channels(self) -> Dict[str, NotificationChannel]: | ||
| """加载通知渠道""" | ||
| channels_config = self.config.get('notification_channels', {}) | ||
| channels = {} | ||
|
|
||
| # 邮件渠道 | ||
| if 'email' in channels_config: | ||
| channels['email'] = EmailNotification(channels_config['email']) | ||
|
|
||
| # 飞书渠道 | ||
| if 'feishu' in channels_config: | ||
| channels['feishu'] = FeishuNotification(channels_config['feishu']) | ||
|
|
||
| # 企业微信渠道 | ||
| if 'wechat_work' in channels_config: | ||
| channels['wechat_work'] = WechatWorkNotification(channels_config['wechat_work']) | ||
|
|
||
| # 钉钉渠道 | ||
| if 'dingtalk' in channels_config: | ||
| channels['dingtalk'] = DingTalkNotification(channels_config['dingtalk']) | ||
|
|
||
| # Telegram渠道 | ||
| if 'telegram' in channels_config: | ||
| channels['telegram'] = TelegramNotification(channels_config['telegram']) | ||
|
|
||
| logger.info(f"加载了 {len(channels)} 个通知渠道") | ||
| return channels |
There was a problem hiding this comment.
这个方法中的一系列 if 判断来加载不同的通知渠道，当未来需要支持更多渠道时，会导致这个方法越来越长，不易维护。建议使用工厂模式或字典映射来进行重构，这样可以使代码更具扩展性和可读性。
例如，你可以创建一个从渠道名称到渠道类的映射字典：
CHANNEL_MAP = {
'email': EmailNotification,
'feishu': FeishuNotification,
'wechat_work': WechatWorkNotification,
'dingtalk': DingTalkNotification,
'telegram': TelegramNotification
}
class AlertManager:
# ...
def _load_channels(self) -> Dict[str, NotificationChannel]:
"""加载通知渠道"""
channels_config = self.config.get('notification_channels', {})
channels = {}
for name, config in channels_config.items():
if name in CHANNEL_MAP and config.get('enabled', False):
try:
channels[name] = CHANNEL_MAP[name](config)
except Exception as e:
logger.error(f"加载通知渠道 '{name}' 失败: {e}")
logger.info(f"加载了 {len(channels)} 个启用的通知渠道")
return channels1 participant
Closes #8