From 3cbf4b370bcd9554b327338e5305305d359a6930 Mon Sep 17 00:00:00 2001 From: bdzxb <1443103348@qq.com> Date: Sat, 16 May 2026 15:54:35 +0800 Subject: [PATCH] modified: main.py deleted: tools/mail_tool.py --- main.py | 114 +++++++++++++++++++++++++++++++++++++++++++ tools/mail_tool.py | 117 --------------------------------------------- 2 files changed, 114 insertions(+), 117 deletions(-) delete mode 100644 tools/mail_tool.py diff --git a/main.py b/main.py index fc8d6dd..ba45cad 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,13 @@ import numpy as np import sqlite3 from .tools.mail_tool import emaillog from .tools.zhangdan_flask import runwebui,stop_server +import imaplib +from datetime import datetime +import email.utils +import email.header +import email +from lxml import html +import re @register("astrbot_bdzxb_bill", "bd_bill", "账单统计", "1.0.0") class MyPlugin(Star): @@ -284,6 +291,113 @@ class MyPlugin(Star): except Exception as e: logger.error(f"文件下载失败 原因：{e}") return + + # 登录邮箱 下载文件 + def emaillog(self): + logger.info(f" 开始执行 登录邮箱 {self.email_usr} 命令 ") + # 邮箱登录用户名 + email_user = self.email_usr + # 邮箱授权码(非登录密码)，用于IMAP协议认证 + email_password = self.emcil_paswd + # IMAP服务器地址，Foxmail使用QQ邮箱的IMAP服务器 + imap_server = self.emcil_imap + mail = imaplib.IMAP4_SSL(imap_server) + # 使用用户名和授权码登录邮箱 + # try: + mail.login(email_user, email_password) + # 查询收件箱 只读模式 + mail.select("inbox",readonly=True) + # 设定查询条件 + for emlname in self.emcil_white: + from_clause = " OR ".join([f'FROM "{sender}"' for sender in emlname]) + # 获取今天的日期，格式化为IMAP要求的格式(如"21-Jul-2025") + today = datetime.now().strftime("%d-%b-%Y") + today = today.encode('utf-8') + search_criteria = f'({from_clause} SINCE "{today}" HAS attachment)' + logger.info(f"严格模式: 只处理今天({today})收到的邮件") + logger.info(f"发件人白名单: {self.emcil_white}") + logger.info(f"搜索条件: {search_criteria}") + + # 执行IMAP搜索命令 + status, messages = mail.search(None, search_criteria) + if status != "OK": # 检查搜索是否成功 + raise Exception("邮件搜索失败") + + # 获取邮件ID列表并只取最后20个（最新的20封） + mail_ids = messages[0].split()[-20:] + logger.info(f"获取到邮件数量: {len(messages)}") + filenames = "" + for mail_id in reversed(mail_ids): + # 获取邮件完整内容(RFC822格式) + status, msg_data = mail.fetch(mail_id, "(RFC822)") + if status != "OK": # 如果获取失败则跳过 + continue + # 解析邮件内容为Message对象 + email_message = email.message_from_bytes(msg_data[0][1]) + # 验证发件人是否在白名单中 + # ======================== + from_header = email.utils.parseaddr(email_message['From'])[1] # 解析发件人邮箱 + # logger.info(f"邮件发件人: {from_header}") + if from_header not in self.emcil_white: # 严格检查白名单 + # logger.info(f"跳过非白名单发件人: {from_header}") + continue + + # 遍历邮件各部分 + # ============= + for part in email_message.walk(): # 递归遍历邮件所有部分 + if part.get_content_maintype() == "multipart": # 跳过multipart容器部分 + continue + filename = part.get_filename() # 获取附件文件名 + # 获取正文 + maildata = part.get_payload(decode=True) + # 微信导出文件需要点击链接下载 检测正文中是否包含链接 解析正文内容 筛选a标签获取下载链接 + etree = html.etree + tree = etree.HTML(maildata) + # logger.info(f"邮件正文{maildata}") + a_tags = tree.xpath('//a') + mail_date = email.utils.parsedate_to_datetime(email_message['Date']) if email_message['Date'] else None + mail_date = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9]", "", str(mail_date)) + for a_tag in a_tags: + # 获取到url后下载url + urlt = a_tag.get("href") + if urlt: + try: + filenames += self.download_file(urlt=urlt,filenamett=f"{from_header}{mail_date}") + except Exception as e: + logger.debug(f"URl 文件下载失败：{e}") + # 处理无文件名的情况 + # ================ + if not filename: # 如果附件没有文件名 + # 按日期发件人生成附件名 + filename = f"{from_header}{mail_date}.zip" + # 文件名处理 + # ========= + # 解码邮件头中的文件名 + try: + decoded_name = email.header.decode_header(filename)[0][0] + if isinstance(decoded_name, bytes): + decoded_name = decoded_name.decode() + except Exception as e: + # 文件名解析失败 自动生成新名字 + decoded_name,filename = f"{from_header}{mail_date}.zip" + today_str = datetime.now().strftime("%Y%m%d") # 获取当前日期字符串 + base_name, ext = os.path.splitext(decoded_name) # 拆分文件名和扩展名 + # 生成新文件名格式：YYYYMMDD-原文件名.zip + new_filename = f"{today_str}-{base_name}{ext}" + if filename: + new_filename = filename + # 拼接完整的文件保存路径 + filepath = os.path.join(StarTools.get_data_dir("astrbot_bdzxb_bill"), new_filename) + if new_filename not in self.no_file: + # 附件保存处理 + # 以二进制模式保存附件 + with open(filepath, "wb") as f: + f.write(part.get_payload(decode=True)) # 解码并写入附件内容 + # logger.info(f"下载附件: {filename} -> {new_filename}") + filenames += new_filename+"\n" + return "获取成功:"+filenames + + async def terminate(self): """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。""" diff --git a/tools/mail_tool.py b/tools/mail_tool.py deleted file mode 100644 index 45a7f01..0000000 --- a/tools/mail_tool.py +++ /dev/null @@ -1,117 +0,0 @@ -import imaplib -from datetime import datetime -import email.utils -import email.header -from astrbot.api import logger -import email -from lxml import html -import re -import os -from astrbot.api.star import StarTools - - -# 登录邮箱 下载文件 -def emaillog(self): - logger.info(f" 开始执行 登录邮箱 {self.email_usr} 命令 ") - # 邮箱登录用户名 - email_user = self.email_usr - # 邮箱授权码(非登录密码)，用于IMAP协议认证 - email_password = self.emcil_paswd - # IMAP服务器地址，Foxmail使用QQ邮箱的IMAP服务器 - imap_server = self.emcil_imap - mail = imaplib.IMAP4_SSL(imap_server) - # 使用用户名和授权码登录邮箱 - # try: - mail.login(email_user, email_password) - # 查询收件箱 只读模式 - mail.select("inbox",readonly=True) - # 设定查询条件 - for emlname in self.emcil_white: - from_clause = " OR ".join([f'FROM "{sender}"' for sender in emlname]) - # 获取今天的日期，格式化为IMAP要求的格式(如"21-Jul-2025") - today = datetime.now().strftime("%d-%b-%Y") - today = today.encode('utf-8') - search_criteria = f'({from_clause} SINCE "{today}" HAS attachment)' - logger.info(f"严格模式: 只处理今天({today})收到的邮件") - logger.info(f"发件人白名单: {self.emcil_white}") - logger.info(f"搜索条件: {search_criteria}") - - # 执行IMAP搜索命令 - status, messages = mail.search(None, search_criteria) - if status != "OK": # 检查搜索是否成功 - raise Exception("邮件搜索失败") - - # 获取邮件ID列表并只取最后20个（最新的20封） - mail_ids = messages[0].split()[-20:] - logger.info(f"获取到邮件数量: {len(messages)}") - filenames = "" - for mail_id in reversed(mail_ids): - # 获取邮件完整内容(RFC822格式) - status, msg_data = mail.fetch(mail_id, "(RFC822)") - if status != "OK": # 如果获取失败则跳过 - continue - # 解析邮件内容为Message对象 - email_message = email.message_from_bytes(msg_data[0][1]) - # 验证发件人是否在白名单中 - # ======================== - from_header = email.utils.parseaddr(email_message['From'])[1] # 解析发件人邮箱 - # logger.info(f"邮件发件人: {from_header}") - if from_header not in self.emcil_white: # 严格检查白名单 - # logger.info(f"跳过非白名单发件人: {from_header}") - continue - - # 遍历邮件各部分 - # ============= - for part in email_message.walk(): # 递归遍历邮件所有部分 - if part.get_content_maintype() == "multipart": # 跳过multipart容器部分 - continue - filename = part.get_filename() # 获取附件文件名 - # 获取正文 - maildata = part.get_payload(decode=True) - # 微信导出文件需要点击链接下载 检测正文中是否包含链接 解析正文内容 筛选a标签获取下载链接 - etree = html.etree - tree = etree.HTML(maildata) - # logger.info(f"邮件正文{maildata}") - a_tags = tree.xpath('//a') - mail_date = email.utils.parsedate_to_datetime(email_message['Date']) if email_message['Date'] else None - mail_date = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9]", "", str(mail_date)) - for a_tag in a_tags: - # 获取到url后下载url - urlt = a_tag.get("href") - if urlt: - try: - filenames += self.download_file(urlt=urlt,filenamett=f"{from_header}{mail_date}") - except Exception as e: - logger.debug(f"URl 文件下载失败：{e}") - # 处理无文件名的情况 - # ================ - if not filename: # 如果附件没有文件名 - # 按日期发件人生成附件名 - filename = f"{from_header}{mail_date}.zip" - # 文件名处理 - # ========= - # 解码邮件头中的文件名 - try: - decoded_name = email.header.decode_header(filename)[0][0] - if isinstance(decoded_name, bytes): - decoded_name = decoded_name.decode() - except Exception as e: - # 文件名解析失败 自动生成新名字 - decoded_name,filename = f"{from_header}{mail_date}.zip" - today_str = datetime.now().strftime("%Y%m%d") # 获取当前日期字符串 - base_name, ext = os.path.splitext(decoded_name) # 拆分文件名和扩展名 - # 生成新文件名格式：YYYYMMDD-原文件名.zip - new_filename = f"{today_str}-{base_name}{ext}" - if filename: - new_filename = filename - # 拼接完整的文件保存路径 - filepath = os.path.join(StarTools.get_data_dir("astrbot_bdzxb_bill"), new_filename) - if new_filename not in self.no_file: - # 附件保存处理 - # 以二进制模式保存附件 - with open(filepath, "wb") as f: - f.write(part.get_payload(decode=True)) # 解码并写入附件内容 - # logger.info(f"下载附件: {filename} -> {new_filename}") - filenames += new_filename+"\n" - return "获取成功:"+filenames - \ No newline at end of file