bdzxb
3e277afddc
modified: README.md modified: __pycache__/main.cpython-312.pyc modified: _conf_schema.json new file: image.png modified: main.py deleted: static/files/ceshi.txt deleted: static/img/server-icon.png new file: tools/.flask.pid new file: tools/__pycache__/mail_tool.cpython-312.pyc new file: tools/__pycache__/zhangdan_flask.cpython-312.pyc new file: tools/mail_tool.py new file: tools/zhangdan_flask.py
117 lines
5.6 KiB
Python
117 lines
5.6 KiB
Python
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
|
|
|