modified: .idea/misc.xml
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
This commit is contained in:
Generated
+2
-1
@@ -1,8 +1,9 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="Black">
|
||||
<option name="sdkName" value="Python 3.13 (venv)" />
|
||||
</component>
|
||||
<component name="ProjectRootManager" version="2" languageLevel="JDK_X" default="true" project-jdk-name="Python 3.13 (venv)" project-jdk-type="Python SDK">
|
||||
<component name="ProjectRootManager" version="2" languageLevel="JDK_22" project-jdk-name="Python 3.13" project-jdk-type="Python SDK">
|
||||
<output url="file://$PROJECT_DIR$/out" />
|
||||
</component>
|
||||
</project>
|
||||
@@ -1,3 +1,47 @@
|
||||
# astrbot_bdzxb_bill
|
||||
|
||||
astrbot插件，通过微信支付宝导出账单到邮箱，然后进行解析统计
|
||||
astrbot插件，通过微信支付宝导出账单到邮箱，然后进行解析统计
|
||||
|
||||
## 需要导入的包
|
||||
|
||||
* sqlite3
|
||||
* numpy
|
||||
* urllib.parse
|
||||
* requests
|
||||
* pandas
|
||||
* zipfile
|
||||
* os
|
||||
* pyzipper
|
||||
* re
|
||||
* lxml
|
||||
* datetime
|
||||
* imaplib
|
||||
* secrets
|
||||
* threading
|
||||
* flask
|
||||
* flask_cors
|
||||
|
||||
---
|
||||
|
||||
## 配置文件
|
||||
|
||||
* 
|
||||
|
||||
## 使用方法
|
||||
|
||||
1. 补充完整配置文件
|
||||
2. 导出账单到邮箱
|
||||
1. 微信： 点击微信支付》我的账单》账单明细》点击右上角的三个点》下载账单》用于个人对账》接受方式 选择邮箱》下一步》发送
|
||||
2. 支付宝：我的》账单》点击右上角三个点》开具交易流水证明》用于个人对账》接收方式选邮箱》发送
|
||||
3. 账单导出到邮箱后 引用内消息会推送解压密码给你
|
||||
4. 在bot中输入指令 /mm 解压密码 解压密码 此处不区分前后顺序 最终结果会放入 插件data文件夹的zhangdan.db中，/chazhang 指令可查询最后10条数据
|
||||
5. 插件提供了一个web页面展示数据（AI生成），使用/rwebui 启动服务 服务端口可根据插件配置文件修改，查看后可使用/swebui 指令关闭web服务
|
||||
6. 服务器指令均需astrbot管理员权限执行
|
||||
|
||||
## 指令列表
|
||||
|
||||
* /zdhelp 获取帮助
|
||||
* /mm 密码1 密码2 # 解压并记录数据
|
||||
* /chazhang # 查看最后10条数据
|
||||
* /rwebui 启动webui
|
||||
* /swebui 关闭webui
|
||||
|
||||
Binary file not shown.
@@ -31,5 +31,27 @@
|
||||
"type": "list",
|
||||
"hint": "屏蔽干扰附件",
|
||||
"obvious_hint": true
|
||||
},
|
||||
"rwebui": {
|
||||
"description": "是否开启webui",
|
||||
"type": "bool",
|
||||
"hint": "是否开启webui",
|
||||
"obvious_hint": true,
|
||||
"default":false
|
||||
},
|
||||
"webui_port": {
|
||||
"description": "webui端口",
|
||||
"type": "int",
|
||||
"hint": "webui端口 2~65534 输入无效端口会默认使用5001",
|
||||
"obvious_hint": true,
|
||||
"default":5001
|
||||
},
|
||||
"webui_passwd": {
|
||||
"description": "webui访问密码",
|
||||
"type": "string",
|
||||
"hint": "webui访问密码",
|
||||
"obvious_hint": true,
|
||||
"default":"admin"
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,30 +1,7 @@
|
||||
|
||||
from pathlib import Path
|
||||
import astrbot.core.message.components as Comp
|
||||
import datetime
|
||||
import json
|
||||
from astrbot.api.event import filter, AstrMessageEvent,MessageChain
|
||||
from astrbot.api.event import filter, AstrMessageEvent
|
||||
from astrbot.api.star import Context, Star, register, StarTools
|
||||
from astrbot.api import logger, AstrBotConfig
|
||||
from astrbot.api.message_components import Node, Plain,Video,Image,Nodes,File
|
||||
from astrbot.core.utils.session_waiter import (
|
||||
session_waiter,
|
||||
SessionController,
|
||||
)
|
||||
from typing import List
|
||||
import imaplib
|
||||
# 导入email库用于解析邮件内容
|
||||
import email
|
||||
# 导入os库用于文件和目录操作
|
||||
import os
|
||||
# 从datetime导入datetime用于日期时间处理
|
||||
from datetime import datetime
|
||||
# 导入email.utils用于邮件日期解析
|
||||
import email.utils
|
||||
# 导入email.header用于解码邮件头
|
||||
import email.header
|
||||
from lxml import html
|
||||
import re
|
||||
import os
|
||||
import zipfile
|
||||
from pyzipper import AESZipFile
|
||||
import pandas as pd
|
||||
@@ -32,22 +9,38 @@ import requests
|
||||
from urllib.parse import urlparse
|
||||
import numpy as np
|
||||
import sqlite3
|
||||
from .tools.mail_tool import emaillog
|
||||
from .tools.zhangdan_flask import runwebui,stop_server
|
||||
|
||||
@register("astrbot_bdzxb_bill", "bd_bill", "账单统计", "1.0.0")
|
||||
class MyPlugin(Star):
|
||||
def __init__(self, context: Context, config: AstrBotConfig):
|
||||
super().__init__(context)
|
||||
self.ceshi = None
|
||||
# 邮箱
|
||||
self.email_usr = None
|
||||
# 授权码
|
||||
self.emcil_paswd = None
|
||||
# imap服务器地址
|
||||
self.emcil_imap = None
|
||||
self.message_target = None
|
||||
# 略过附件列表
|
||||
self.no_file = None
|
||||
# 发件人白名单
|
||||
self.emcil_white = None
|
||||
# 插件文件位置
|
||||
self.file_path = "data/plugins/astrbot_bdzxb_bill"
|
||||
# 配置文件
|
||||
self.config = config
|
||||
self.chatid = None
|
||||
# 插件data文件保存位置
|
||||
self.data_path = None
|
||||
# sqllite Connection
|
||||
self.sqllitaa = None
|
||||
# webui 开关
|
||||
self.rwebui = None
|
||||
# webui 端口
|
||||
self.webui_port = None
|
||||
# webui 密码
|
||||
self.webui_passwd = None
|
||||
# 初始化插件
|
||||
async def initialize(self):
|
||||
"""可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。"""
|
||||
# 读取配置信息
|
||||
@@ -56,9 +49,17 @@ class MyPlugin(Star):
|
||||
self.emcil_imap = self.config.emcil_imap
|
||||
self.emcil_white = self.config.emcil_white
|
||||
self.no_file = self.config.no_file
|
||||
|
||||
self.data_path = StarTools.get_data_dir("astrbot_bdzxb_bill")
|
||||
self.rwebui = self.config.rwebui
|
||||
# 检查端口号是否合规 不合规更新为5001
|
||||
if self.rwebui >35565 or self.rwebui<2:
|
||||
self.rwebui = 5001
|
||||
self.config.save_config() # 保存配置
|
||||
self.webui_port = self.config.webui_port
|
||||
self.webui_passwd = self.config.webui_passwd
|
||||
# 如果data文件夹不包含db文件 则创建 并且初始化数据库
|
||||
file_list = os.listdir(f"{self.data_path}")
|
||||
|
||||
if "zhangdan.db" not in file_list:
|
||||
self.sqllitaa = sqlite3.connect(f"{self.data_path}/zhangdan.db")
|
||||
# 创建游标对象
|
||||
@@ -93,193 +94,119 @@ class MyPlugin(Star):
|
||||
self.sqllitaa.close()
|
||||
else:
|
||||
self.sqllitaa = sqlite3.connect(f"{self.data_path}/zhangdan.db")
|
||||
|
||||
|
||||
# 读取配置文件 查看石头需要直接启动webui
|
||||
if self.rwebui:
|
||||
runwebui(int(self.webui_port), self.webui_passwd, f"{self.data_path}/zhangdan.db")
|
||||
|
||||
def guanbiweb():
|
||||
MyPlugin.rwebui = False
|
||||
MyPlugin.config.save_config() # 保存配置
|
||||
|
||||
# 帮助
|
||||
@filter.permission_type(filter.PermissionType.ADMIN)
|
||||
@filter.command("zdhelp")
|
||||
async def mchelp(self, event: AstrMessageEvent):
|
||||
logger.info(" 开始执行 zdhelp 命令 ")
|
||||
helptxt = """ """
|
||||
helptxt = "/zdhelp 获取本帮助文档\n/mm 密码1 密码2 # 解压并记录数据 \n/chazhang # 查看最后10条数据 \n/rwebui 启动webui \n/swebui 关闭webui"
|
||||
yield event.plain_result(f"{helptxt}")
|
||||
|
||||
# 获取解压密码冰进行下载解压
|
||||
# 启动webui
|
||||
@filter.permission_type(filter.PermissionType.ADMIN)
|
||||
@filter.command("rwebui")
|
||||
async def rwebuicmd(self, event: AstrMessageEvent):
|
||||
logger.info(" 开始执行 rwebui 命令 ")
|
||||
|
||||
try:
|
||||
self.rwebui = False
|
||||
self.config.save_config() # 保存配置
|
||||
stop_server()
|
||||
except Exception as e:
|
||||
logger.info("准备开启服务")
|
||||
|
||||
# 查看是否开启
|
||||
self.rwebui = True
|
||||
self.config.save_config() # 保存配置
|
||||
runwebui(int(self.webui_port), self.webui_passwd, f"{self.data_path}/zhangdan.db")
|
||||
|
||||
yield event.plain_result(f"webUI已启动 0.0.0.0：{self.webui_port}")
|
||||
# 停止webui
|
||||
@filter.permission_type(filter.PermissionType.ADMIN)
|
||||
@filter.command("swebui")
|
||||
async def swebuicmd(self, event: AstrMessageEvent):
|
||||
logger.info(" 开始执行 swebui 命令 ")
|
||||
# 查看是否开启
|
||||
if self.rwebui:
|
||||
self.rwebui = False
|
||||
self.config.save_config() # 保存配置
|
||||
stop_server()
|
||||
yield event.plain_result("webdui已关闭")
|
||||
|
||||
# 获取解压密码并进行下载解压
|
||||
@filter.permission_type(filter.PermissionType.ADMIN)
|
||||
@filter.command("mm")
|
||||
async def wxmm(self, event: AstrMessageEvent, wxmm: str,zfbmm: str):
|
||||
# # 登录邮箱 下载文件
|
||||
yield event.plain_result(f"{self.emaillog()}")
|
||||
# 登录邮箱 下载文件
|
||||
yield event.plain_result(emaillog())
|
||||
logger.info("开始解压文件")
|
||||
# # 解压文件 self.data_path
|
||||
# 解压文件 self.data_path
|
||||
file_list = os.listdir(self.data_path )
|
||||
logger.info(f"{file_list}")
|
||||
# logger.info(f"{file_list}")
|
||||
# 解压 两个密码都试一次
|
||||
for aac in file_list:
|
||||
self.jzip(filename=aac,mima=wxmm)
|
||||
self.jzip(filename=aac,mima=zfbmm)
|
||||
# 解析文件
|
||||
file_list = os.listdir(f"{self.data_path}/zhangdan")
|
||||
yield event.plain_result(f"解压后文件列表：{file_list}")
|
||||
self.sqllitaa = sqlite3.connect(f"{self.data_path}/zhangdan.db")
|
||||
cur = self.sqllitaa.cursor()
|
||||
try:
|
||||
for aaa in file_list:
|
||||
# 获取sql语句列表并执行
|
||||
for icc in self.exceltt(filename=str(self.data_path)+"/zhangdan/"+aaa):
|
||||
cur.execute(icc)
|
||||
self.sqllitaa.commit()
|
||||
cur.close()
|
||||
self.sqllitaa.close()
|
||||
# 删除源文件
|
||||
file_list = os.listdir(f"{self.data_path}")
|
||||
for nam in file_list:
|
||||
if "zhangdan.db" not in nam:
|
||||
try:
|
||||
try:
|
||||
# 删除源文件
|
||||
file_list = os.listdir(f"{self.data_path}")
|
||||
for nam in file_list:
|
||||
if "zhangdan" not in nam:
|
||||
os.remove(str(self.data_path)+"\\"+nam)
|
||||
except IOError as e:
|
||||
file_listaa = os.listdir(f"{self.data_path}/zhangdan/")
|
||||
for namaa in file_listaa:
|
||||
os.remove(str(self.data_path)+"\\zhangdan\\"+namaa)
|
||||
# logger.info(f"{e}")
|
||||
|
||||
file_listaa = os.listdir(f"{self.data_path}/zhangdan/")
|
||||
for namaa in file_listaa:
|
||||
os.remove(str(self.data_path)+"\\zhangdan\\"+namaa)
|
||||
except IOError as e:
|
||||
logger.error(f"文件删除失败 原因{e}")
|
||||
yield event.plain_result("保存完成__使用/chazhang 查询最后10条记录")
|
||||
|
||||
except Exception as e:
|
||||
yield event.plain_result(f"保存失败{e}")
|
||||
logger.info(f"{e}")
|
||||
logger.error(f"保存失败 原因：{e}")
|
||||
cur.close()
|
||||
self.sqllitaa.close()
|
||||
|
||||
# 查账
|
||||
@filter.permission_type(filter.PermissionType.ADMIN)
|
||||
@filter.command("chazhang")
|
||||
async def chazhang(self, event: AstrMessageEvent):
|
||||
logger.info(" 开始执行 chazhang 命令 ")
|
||||
self.sqllitaa = sqlite3.connect(f"{self.data_path}/zhangdan.db")
|
||||
# 创建游标对象
|
||||
cursor = self.sqllitaa.cursor()
|
||||
# 返回最后10条
|
||||
try:
|
||||
cursor.execute('SELECT * FROM zhangdan ORDER BY id DESC LIMIT 10;')
|
||||
jieg =cursor.fetchall()
|
||||
self.sqllitaa.commit()
|
||||
cursor.close()
|
||||
self.sqllitaa.close()
|
||||
yield event.plain_result(f"{jieg}")
|
||||
except Exception as e:
|
||||
yield event.plain_result(f"查询失败：{e}")
|
||||
logger.info(f"查询失败{e}")
|
||||
logger.error(f"查询失败{e}")
|
||||
self.sqllitaa.commit()
|
||||
cursor.close()
|
||||
self.sqllitaa.close()
|
||||
cursor.close()
|
||||
self.sqllitaa.close()
|
||||
|
||||
# 登录邮箱 下载文件
|
||||
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")
|
||||
search_criteria = f'({from_clause} SINCE "{today.encode('utf-8')}" 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)
|
||||
# 微信导出文件需要点击链接下载 检测正文中是否包含链接
|
||||
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:
|
||||
logger.info(f"{e} \n解析文件名称失败{email.header.decode_header(filename)[0][0]}")
|
||||
decoded_name = f"{from_header}{datetime.now().strftime("%d-%b-%Y")}.zip"
|
||||
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:
|
||||
# 附件保存处理
|
||||
# ===========
|
||||
if not os.path.exists(filepath): # 检查文件是否已存在
|
||||
# 以二进制模式保存附件
|
||||
with open(filepath, "wb") as f:
|
||||
f.write(part.get_payload(decode=True)) # 解码并写入附件内容
|
||||
logger.info(f"下载附件: {filename} -> {new_filename}")
|
||||
filenames += new_filename+"\n"
|
||||
else: # 文件已存在则跳过
|
||||
with open(filepath, "wb") as f:
|
||||
f.write(part.get_payload(decode=True)) # 解码并写入附件内容
|
||||
logger.info(f"附件已存在，覆盖: {new_filename}")
|
||||
return "获取成功:"+filenames
|
||||
|
||||
# 分析文件并返回SQL语句
|
||||
def exceltt(self,filename):
|
||||
yuju = []
|
||||
try:
|
||||
@@ -309,34 +236,21 @@ class MyPlugin(Star):
|
||||
yuju.append(f"replace into zhangdan(id,payid,paytype,paydate,payname,payto,paysize,paystu,paybox,payqd) VALUES('{np.float64(shuju[8])}','{shuju[8]}','{shuju[1]}','{shuju[0]}','{shuju[2]+shuju[3]}','{shuju[4]}','{shuju[5]}','{shuju[7]}','{shuju[6]}','微信')")
|
||||
# 读取xlsx文件
|
||||
return yuju
|
||||
|
||||
except Exception as e:
|
||||
logger.info(f"{e}")
|
||||
logger.error(f"解析文件失败 原因{e}")
|
||||
return
|
||||
|
||||
|
||||
# if "支付宝" in filename:
|
||||
|
||||
# elif "微信" in filename:
|
||||
|
||||
# 解压文件
|
||||
def jzip(self,filename,mima):
|
||||
filename = self.data_path / f'{filename}'
|
||||
# 输出目录
|
||||
output_path = f'{self.data_path}/zhangdan'
|
||||
# 检查目录是否存在
|
||||
try:
|
||||
os.mkdir(output_path)
|
||||
except Exception as e:
|
||||
logger.info(f"文件夹已存在开始解压{filename}")
|
||||
# try:
|
||||
# # 使用AESZipFile打开加密的ZIP文件
|
||||
# with AESZipFile(filename, 'r') as zf:
|
||||
# # 列出ZIP文件中的所有文件和文件夹
|
||||
# logger.info(f"本次解压文件{zf.namelist()}")
|
||||
# # 解压所有文件到指定目录
|
||||
# zf.extractall(path=output_path, pwd=mima.encode("bytes"))
|
||||
# except Exception as e:
|
||||
# logger.debug(f"解压失败：{e}")
|
||||
|
||||
# 解压文件
|
||||
try:
|
||||
with AESZipFile(filename, 'r') as zf:
|
||||
logger.info(f"本次解压文件{zf.namelist()}")
|
||||
@@ -346,15 +260,11 @@ class MyPlugin(Star):
|
||||
logger.error(f"解压失败，原因：{e}") # RuntimeError 常因密码错误触发
|
||||
except Exception as e:
|
||||
logger.error(f"解压过程中出现未知错误：{e}")
|
||||
|
||||
|
||||
|
||||
|
||||
# 通过url下载文件
|
||||
def download_file(self,urlt,filenamett):
|
||||
try:
|
||||
response = requests.get(urlt, stream=True)
|
||||
response.raise_for_status()
|
||||
|
||||
download_dir = StarTools.get_data_dir("astrbot_bdzxb_bill")
|
||||
os.makedirs(download_dir, exist_ok=True)
|
||||
file_name = os.path.basename(urlparse(urlt).path)+filenamett
|
||||
@@ -372,24 +282,8 @@ class MyPlugin(Star):
|
||||
else:
|
||||
return
|
||||
except Exception as e:
|
||||
return
|
||||
|
||||
|
||||
#保存文件
|
||||
def get_json_path(self, filename: str) -> Path:
|
||||
"""
|
||||
读取Jason文件
|
||||
|
||||
Args:
|
||||
filename: 文件名称
|
||||
|
||||
Returns:
|
||||
JSON文件的Path对象
|
||||
"""
|
||||
data_path = StarTools.get_data_dir("astrbot_bdzxb_bill")
|
||||
json_path = data_path / f'{filename}.json'
|
||||
json_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return json_path
|
||||
logger.error(f"文件下载失败 原因：{e}")
|
||||
return
|
||||
|
||||
async def terminate(self):
|
||||
"""可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。"""
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
这是一个测试文件
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 32 KiB |
@@ -0,0 +1 @@
|
||||
24764
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,117 @@
|
||||
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
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user