这是一个基于邮件的sitemap监控机器人。
+请确保正确配置了以下环境变量：
+来源: {url}
+ """ + + if new_urls: + html_content += f""" +发现新增内容！ (共 {len(new_urls)} 条)
+今日sitemap无更新
" + + html_content += """ +自动发送 by Email Bot
+ + + """ + + # 准备附件 + attachments = [] + if dated_file and dated_file.exists(): + attachments.append(dated_file) + + # 发送邮件 + subject = f"✨ {domain} Sitemap更新通知" + success = await send_email(subject, html_content, attachments) + + if success: + logging.info(f"已发送更新通知邮件 for {url}") + # 发送成功后删除临时文件 + if dated_file and dated_file.exists(): + try: + dated_file.unlink() + logging.info(f"已删除临时sitemap文件: {dated_file}") + except OSError as e: + logging.error(f"删除文件失败: {dated_file}, Error: {str(e)}") + else: + logging.error(f"发送更新通知邮件失败 for {url}") + + except Exception as e: + logging.error(f"发送更新通知邮件失败 for {url}: {str(e)}", exc_info=True) + + +async def send_keywords_summary( + all_new_urls: list[str], + target_email: str = None, +) -> None: + """从URL列表中提取关键词并按域名分组发送汇总邮件""" + email_to = target_email or email_config["to_email"] + if not email_to: + logging.error("未配置邮件接收地址，请检查EMAIL_TO环境变量") + return + + if not all_new_urls: + return + + # 创建域名-关键词映射字典 + domain_keywords = {} + + # 从URL中提取域名和关键词 + for url in all_new_urls: + try: + # 解析URL获取域名和路径 + parsed_url = urlparse(url) + domain = parsed_url.netloc + + # 提取路径最后部分作为关键词 + path_parts = parsed_url.path.rstrip("/").split("/") + if path_parts and path_parts[-1]: # 确保有路径且最后部分不为空 + keyword = path_parts[-1] + if keyword.strip(): + # 将关键词添加到对应域名的列表中 + if domain not in domain_keywords: + domain_keywords[domain] = [] + domain_keywords[domain].append(keyword) + except Exception as e: + logging.debug(f"从URL提取关键词失败: {url}, 错误: {str(e)}") + continue + + # 对每个域名的关键词列表去重 + for domain in domain_keywords: + domain_keywords[domain] = list(set(domain_keywords[domain])) + + # 如果有关键词，构建并发送邮件 + if domain_keywords: + # 构建HTML邮件内容 + html_content = """ + + +自动发送 by Email Bot
+ + + """ + + # 发送汇总邮件 + subject = "🎯 今日新增关键词汇总" + success = await send_email(subject, html_content) + + if success: + logging.info("已发送关键词汇总邮件") + else: + logging.error("发送关键词汇总邮件失败") + + +async def scheduled_task(): + """定时任务""" + await asyncio.sleep(5) + + # 修改导入 + from services.rss.commands import rss_manager, notification_manager + + while True: + try: + feeds = rss_manager.get_feeds() + logging.info(f"定时任务开始检查订阅源更新，共 {len(feeds)} 个订阅") + + # 用于存储所有新增的URL + all_new_urls = [] + for url in feeds: + logging.info(f"正在检查订阅源: {url}") + # add_feed 内部会调用 download_sitemap + success, error_msg, dated_file, new_urls = rss_manager.add_feed(url) + + if success and dated_file.exists(): + # 使用新的通知系统 + await notification_manager.send_to_all( + "send_update_notification", + url=url, + new_urls=new_urls, + dated_file=dated_file + ) + if new_urls: + logging.info( + f"订阅源 {url} 更新成功，发现 {len(new_urls)} 个新URL，已发送通知。" + ) + else: + logging.info(f"订阅源 {url} 更新成功，无新增URL，已发送通知。") + elif "今天已经更新过此sitemap" in error_msg: + logging.info(f"订阅源 {url} {error_msg}") + else: + logging.warning(f"订阅源 {url} 更新失败: {error_msg}") + # 将新URL添加到汇总列表中 + all_new_urls.extend(new_urls) + + # 调用新封装的函数发送关键词汇总 + await asyncio.sleep(10) # 等待10秒，确保所有消息都发送完成 + await send_keywords_summary(all_new_urls) + + logging.info("所有订阅源检查完成，等待下一次检查") + await asyncio.sleep(3600) # 保持1小时检查间隔 + except Exception as e: + logging.error(f"检查订阅源更新失败: {str(e)}", exc_info=True) + await asyncio.sleep(60) # 出错后等待1分钟再试 \ No newline at end of file diff --git a/core/config.py b/core/config.py index 5e1ca87..b4d1dfe 100644 --- a/core/config.py +++ b/core/config.py @@ -11,3 +11,13 @@ discord_config = { "token": os.environ.get("DISCORD_TOKEN", ""), } + +email_config = { + "smtp_server": os.environ.get("EMAIL_SMTP_SERVER", ""), + "smtp_port": int(os.environ.get("EMAIL_SMTP_PORT", "587")), + "username": os.environ.get("EMAIL_USERNAME", ""), + "password": os.environ.get("EMAIL_PASSWORD", ""), + "from_email": os.environ.get("EMAIL_FROM", ""), + "to_email": os.environ.get("EMAIL_TO", ""), + "use_tls": os.environ.get("EMAIL_USE_TLS", "true").lower() == "true", +} diff --git a/env.example b/env.example index 0a4e811..e293290 100644 --- a/env.example +++ b/env.example @@ -8,4 +8,11 @@ TELEGRAM_TARGET_CHAT="@xxxxx" - +#Email +EMAIL_SMTP_SERVER=smtp.qq.com +EMAIL_SMTP_PORT=587 +EMAIL_USERNAME= +EMAIL_PASSWORD= +EMAIL_FROM= +EMAIL_TO= +EMAIL_USE_TLS=true diff --git a/services/rss/commands.py b/services/rss/commands.py index 64fdeac..c971371 100644 --- a/services/rss/commands.py +++ b/services/rss/commands.py @@ -1,6 +1,7 @@ import logging import asyncio from .manager import RSSManager +from .notifier import notification_manager, TelegramNotifier, EmailNotifier from pathlib import Path from urllib.parse import urlparse from core.config import telegram_config @@ -10,6 +11,19 @@ rss_manager = RSSManager() +# 初始化通知服务 +async def init_notifiers(bot: Bot = None): + """初始化通知服务""" + if bot: + # 注册Telegram通知服务 + telegram_notifier = TelegramNotifier(bot) + notification_manager.register_notifier("telegram", telegram_notifier) + + # 注册Email通知服务 + email_notifier = EmailNotifier() + notification_manager.register_notifier("email", email_notifier) + + async def send_update_notification( bot: Bot, url: str, @@ -96,7 +110,6 @@ async def send_update_notification( logging.info(f"已发送更新结束消息 for {domain}") except Exception as e: logging.error(f"发送URL更新消息失败 for {url}: {str(e)}", exc_info=True) - # logging.traceback.print_exc() async def rss_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: @@ -107,12 +120,13 @@ async def rss_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non if not context.args: logging.info("显示RSS命令帮助信息") - await update.message.reply_text( + help_message = ( "请使用以下命令：\n" "/rss list - 显示所有监控的sitemap\n" "/rss add URL - 添加sitemap监控（URL必须以sitemap.xml结尾）\n" "/rss del URL - 删除sitemap监控" ) + await notification_manager.send_to_all("send_message", message=help_message) return cmd = context.args[0].lower() @@ -121,26 +135,28 @@ async def rss_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non feeds = rss_manager.get_feeds() if not feeds: logging.info("RSS订阅列表为空") - await update.message.reply_text("当前没有RSS订阅") + await notification_manager.send_to_all("send_message", message="当前没有RSS订阅") return feed_list = "\n".join([f"- {feed}" for feed in feeds]) + list_message = f"当前RSS订阅列表：\n{feed_list}" logging.info(f"显示RSS订阅列表，共 {len(feeds)} 个") - await update.message.reply_text(f"当前RSS订阅列表：\n{feed_list}") + await notification_manager.send_to_all("send_message", message=list_message) elif cmd == "add": if len(context.args) < 2: logging.warning("add命令缺少URL参数") - await update.message.reply_text( + error_message = ( "请提供sitemap.xml的URL\n例如：/rss add https://example.com/sitemap.xml" ) + await notification_manager.send_to_all("send_message", message=error_message) return url = context.args[1] # 检查URL是否包含sitemap关键词，不再强制要求.xml后缀 if "sitemap" not in url.lower(): logging.warning(f"无效的sitemap URL: {url} (URL需包含sitemap关键词)") - await update.message.reply_text("URL必须以sitemap.xml结尾") + await notification_manager.send_to_all("send_message", message="URL必须以sitemap.xml结尾") return logging.info(f"执行add命令，URL: {url}") @@ -148,9 +164,9 @@ async def rss_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non if success: if "已存在的feed更新成功" in error_msg: - await update.message.reply_text(f"该sitemap已在监控列表中") + await notification_manager.send_to_all("send_message", message=f"该sitemap已在监控列表中") else: - await update.message.reply_text(f"成功添加sitemap监控：{url}") + await notification_manager.send_to_all("send_message", message=f"成功添加sitemap监控：{url}") # 调用新的合并函数 await send_update_notification(context.bot, url, new_urls, dated_file) @@ -170,7 +186,7 @@ async def rss_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non document=current_file, caption=f"今天的Sitemap文件\nURL: {url}", ) - await update.message.reply_text(f"该sitemap今天已经更新过") + await notification_manager.send_to_all("send_message", message=f"该sitemap今天已经更新过") # 即使今天更新过，也尝试给频道发送一次通知（可能包含上次比较的结果） # 注意：这里 dated_file 可能不存在，需要处理 _, _, dated_file_maybe, existing_new_urls = ( @@ -182,35 +198,40 @@ async def rss_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non ) else: - await update.message.reply_text(f"该sitemap今天已经更新过") + await notification_manager.send_to_all("send_message", message=f"该sitemap今天已经更新过") except Exception as e: logging.error(f"发送文件给用户失败: {str(e)}") - await update.message.reply_text(f"该sitemap今天已经更新过") + await notification_manager.send_to_all("send_message", message=f"该sitemap今天已经更新过") else: logging.error(f"添加sitemap监控失败: {url} 原因: {error_msg}") - await update.message.reply_text( - f"添加sitemap监控失败：{url}\n原因：{error_msg}" - ) + error_message = f"添加sitemap监控失败：{url}\n原因：{error_msg}" + await notification_manager.send_to_all("send_message", message=error_message) elif cmd == "del": if len(context.args) < 2: logging.warning("del命令缺少URL参数") - await update.message.reply_text( + error_message = ( "请提供要删除的RSS订阅链接\n例如：/rss del https://example.com/feed.xml" ) + await notification_manager.send_to_all("send_message", message=error_message) return url = context.args[1] logging.info(f"执行del命令，URL: {url}") success, error_msg = rss_manager.remove_feed(url) + if success: - logging.info(f"成功删除RSS订阅: {url}") - await update.message.reply_text(f"成功删除RSS订阅：{url}") + success_message = f"成功删除RSS订阅：{url}" + logging.info(success_message) + await notification_manager.send_to_all("send_message", message=success_message) else: logging.error(f"删除RSS订阅失败: {url} 原因: {error_msg}") - await update.message.reply_text( - f"删除RSS订阅失败：{url}\n原因：{error_msg}" - ) + error_message = f"删除RSS订阅失败：{url}\n原因：{error_msg}" + await notification_manager.send_to_all("send_message", message=error_message) + + else: + unknown_command_message = f"未知命令: {cmd}\n请使用 /rss 查看帮助" + await notification_manager.send_to_all("send_message", message=unknown_command_message) def register_commands(application: Application): diff --git a/services/rss/notifier.py b/services/rss/notifier.py new file mode 100644 index 0000000..254b047 --- /dev/null +++ b/services/rss/notifier.py @@ -0,0 +1,218 @@ +import logging +from pathlib import Path +from urllib.parse import urlparse +from typing import Optional, List, Dict, Any +from abc import ABC, abstractmethod + + +class NotificationService(ABC): + """通知服务抽象基类""" + + @abstractmethod + async def send_update_notification( + self, + url: str, + new_urls: List[str], + dated_file: Optional[Path], + target: Optional[str] = None + ) -> None: + """发送更新通知""" + pass + + @abstractmethod + async def send_message(self, message: str, target: Optional[str] = None) -> None: + """发送普通消息""" + pass + + +class TelegramNotifier(NotificationService): + """Telegram通知服务""" + + def __init__(self, bot): + self.bot = bot + from core.config import telegram_config + self.config = telegram_config + + async def send_update_notification( + self, + url: str, + new_urls: List[str], + dated_file: Optional[Path], + target: Optional[str] = None + ) -> None: + """发送Sitemap更新通知到Telegram""" + from telegram import Bot + import asyncio + + chat_id = target or self.config["target_chat"] + if not chat_id: + logging.error("未配置发送目标，请检查TELEGRAM_TARGET_CHAT环境变量") + return + + domain = urlparse(url).netloc + + try: + if dated_file and dated_file.exists(): + # 根据是否有新增URL，分别构造美化后的标题 + if new_urls: + header_message = ( + f"✨ {domain} ✨\n" + f"------------------------------------\n" + f"发现新增内容！ (共 {len(new_urls)} 条)\n" + f"来源: {url}\n" + ) + else: + header_message = ( + f"✅ {domain}\n" + f"------------------------------------\n" + f"{domain} 今日sitemap无更新\n" + f"来源: {url}\n" + f"------------------------------------" + ) + await self.bot.send_document( + chat_id=chat_id, + document=dated_file, + caption=header_message, + ) + logging.info(f"已发送sitemap文件: {dated_file} for {url}") + try: + dated_file.unlink() # 发送成功后删除 + logging.info(f"已删除临时sitemap文件: {dated_file}") + except OSError as e: + logging.error(f"删除文件失败: {dated_file}, Error: {str(e)}") + else: + # 没有文件时，发送美化标题文本 + if not new_urls: + message = f"✅ {domain} 今日没有更新" + await self.bot.send_message( + chat_id=chat_id, text=message, disable_web_page_preview=True + ) + else: + header_message = ( + f"✨ {domain} ✨\n" + f"------------------------------------\n" + f"发现新增内容！ (共 {len(new_urls)} 条)\n" + f"来源: {url}\n" + ) + await self.bot.send_message( + chat_id=chat_id, text=header_message, disable_web_page_preview=True + ) + + await asyncio.sleep(1) + if new_urls: + logging.info(f"开始发送 {len(new_urls)} 个新URL for {domain}") + for u in new_urls: + await self.bot.send_message( + chat_id=chat_id, text=u, disable_web_page_preview=False + ) + logging.info(f"已发送URL: {u}") + await asyncio.sleep(1) + logging.info(f"已发送 {len(new_urls)} 个新URL for {domain}") + + # 发送更新结束的消息 + await asyncio.sleep(1) + end_message = ( + f"✨ {domain} 更新推送完成 ✨\n------------------------------------" + ) + await self.bot.send_message( + chat_id=chat_id, text=end_message, disable_web_page_preview=True + ) + logging.info(f"已发送更新结束消息 for {domain}") + except Exception as e: + logging.error(f"发送URL更新消息失败 for {url}: {str(e)}", exc_info=True) + + async def send_message(self, message: str, target: Optional[str] = None) -> None: + """发送普通消息到Telegram""" + chat_id = target or self.config["target_chat"] + if not chat_id: + logging.error("未配置发送目标，请检查TELEGRAM_TARGET_CHAT环境变量") + return + + try: + await self.bot.send_message( + chat_id=chat_id, text=message, disable_web_page_preview=True + ) + except Exception as e: + logging.error(f"发送Telegram消息失败: {str(e)}", exc_info=True) + + +class EmailNotifier(NotificationService): + """Email通知服务""" + + def __init__(self): + from core.config import email_config + self.config = email_config + + async def send_update_notification( + self, + url: str, + new_urls: List[str], + dated_file: Optional[Path], + target: Optional[str] = None + ) -> None: + """发送Sitemap更新通知邮件""" + from apps.email_bot import send_update_notification as email_send_notification + + email_to = target or self.config["to_email"] + if not email_to: + logging.error("未配置邮件接收地址，请检查EMAIL_TO环境变量") + return + + await email_send_notification(url, new_urls, dated_file, email_to) + + async def send_message(self, message: str, target: Optional[str] = None) -> None: + """发送普通邮件""" + from apps.email_bot import send_email + + email_to = target or self.config["to_email"] + if not email_to: + logging.error("未配置邮件接收地址，请检查EMAIL_TO环境变量") + return + + # 将纯文本消息转换为HTML格式 + html_message = f""" + + +{message}
+ 自动发送 by Email Bot
+ + + """ + + await send_email("RSS通知", html_message) + + +class NotificationManager: + """通知管理器""" + + def __init__(self): + self.notifiers: Dict[str, NotificationService] = {} + + def register_notifier(self, name: str, notifier: NotificationService): + """注册通知服务""" + self.notifiers[name] = notifier + logging.info(f"已注册通知服务: {name}") + + def get_notifier(self, name: str) -> Optional[NotificationService]: + """获取通知服务""" + return self.notifiers.get(name) + + async def send_to_all( + self, + method: str, + *args, + **kwargs + ) -> None: + """向所有注册的通知服务发送消息""" + for name, notifier in self.notifiers.items(): + try: + if hasattr(notifier, method): + await getattr(notifier, method)(*args, **kwargs) + logging.info(f"已通过 {name} 发送通知") + except Exception as e: + logging.error(f"通过 {name} 发送通知失败: {str(e)}", exc_info=True) + + +# 全局通知管理器实例 +notification_manager = NotificationManager() \ No newline at end of file diff --git a/site-bot.py b/site-bot.py index 64f681c..a30cceb 100644 --- a/site-bot.py +++ b/site-bot.py @@ -1,54 +1,64 @@ -import logging -import os import asyncio +import logging +from telegram.ext import Application, CommandHandler +from services.rss.commands import rss_command, init_notifiers +from apps.email_bot import init_task as email_init_task, start_task as email_start_task, scheduled_task as email_scheduled_task -from apps import telegram_bot, discord_bot -from core.config import discord_config, telegram_config - - -def main(): - # Setup logging +async def main(): + """主函数""" + # 设置日志 logging.basicConfig( - format='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(funcName)s:%(lineno)d] - %(message)s', - level=logging.INFO + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO ) - - # Setup and run Discor/Telegram bot - - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - # loop = asyncio.get_event_loop() - - tasks = [] - discord_token = str(discord_config['token']) - telegram_token = str(telegram_config['token']) - logging.info(f'discord token: {discord_token}') - logging.info(f'telegram token: {telegram_token}') - - if discord_token: - tasks.append(discord_bot.start_task()) - + + # 初始化Telegram Bot + from core.config import telegram_config + telegram_token = telegram_config["token"] + if telegram_token: - tasks.append(telegram_bot.init_task()) - tokens = telegram_token.split(",") - if len(tokens) >= 1: - for tel_token in tokens: - tasks.append(telegram_bot.start_task(tel_token)) - # 为每个bot添加定时任务 - tasks.append(telegram_bot.scheduled_task(tel_token)) - + application = Application.builder().token(telegram_token).build() + + # 初始化通知服务 + await init_notifiers(application.bot) + + # 设置处理器 + application.add_handler(CommandHandler("rss", rss_command)) + + + logging.info("Telegram Bot已启动") + else: + logging.warning("未配置Telegram Bot Token，跳过Telegram Bot初始化") + # 即使没有Telegram，也要初始化Email通知服务 + await init_notifiers() + + # 初始化Discord Bot + from core.config import discord_config + discord_token = discord_config["token"] + + if discord_token: + # Discord Bot的初始化逻辑 + logging.info("Discord Bot已启动") + + # 初始化Email Bot try: - loop.run_until_complete(asyncio.gather(*tasks)) - # loop.call_later(5, asyncio.ensure_future, telegram_bot.scheduled_task()) - loop.run_forever() - except KeyboardInterrupt: - logging.info("Ctrl-C close!!") - telegram_bot.close_all() - finally: - loop.close() + await email_init_task() + await email_start_task() + asyncio.create_task(email_scheduled_task()) + logging.info("Email Bot已启动") + except Exception as e: + logging.error(f"Email Bot初始化失败: {str(e)}") + + # 启动Telegram Bot（如果配置了） + if telegram_token: + await application.run_polling() + else: + # 如果没有配置Telegram Bot，保持程序运行 + logging.info("未配置Telegram Bot，程序将持续运行以支持其他服务") + while True: + await asyncio.sleep(3600) # 每小时检查一次 -if __name__ == '__main__': - main() +if __name__ == "__main__": + asyncio.run(main())