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.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 zipfile from pyzipper import AESZipFile import pandas as pd import requests from urllib.parse import urlparse import numpy as np import sqlite3 @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 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 self.data_path = None self.sqllitaa = None async def initialize(self): """可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。""" # 读取配置信息 self.email_usr = self.config.email_usr self.emcil_paswd = self.config.emcil_paswd 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") 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") # 创建游标对象 cursor = self.sqllitaa.cursor() # 创建表 # id:流水号,payid:账单号,paytype:交易类型,paydate:交易时间,payname:交易名称,payto:收入/支出,paysize:交易金额 # paystu:交易状态,paybox:收支方式,payqd:记录渠道 try: cursor.execute( ''' create table if not exists zhangdan( id int primary key , payid text not null, paytype text not null, paydate text not null, payname text not null, payto text not null, paysize text not null, paystu text not null, paybox text not null, payqd text not null ) ''' ) self.sqllitaa.commit() cursor.close() self.sqllitaa.close() except Exception as e: logger.info(f"{e}") self.sqllitaa.commit() cursor.close() self.sqllitaa.close() else: self.sqllitaa = sqlite3.connect(f"{self.data_path}/zhangdan.db") # 帮助 @filter.command("zdhelp") async def mchelp(self, event: AstrMessageEvent): logger.info(" 开始执行 zdhelp 命令 ") helptxt = """ """ yield event.plain_result(f"{helptxt}") # 获取解压密码冰进行下载解压 @filter.command("mm") async def wxmm(self, event: AstrMessageEvent, wxmm: str,zfbmm: str): # # 登录邮箱 下载文件 yield event.plain_result(f"{self.emaillog()}") logger.info("开始解压文件") # # 解压文件 self.data_path file_list = os.listdir(self.data_path ) 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: 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: 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}") yield event.plain_result("保存完成__使用/chazhang 查询最后10条记录") except Exception as e: yield event.plain_result(f"保存失败{e}") logger.info(f"{e}") cur.close() self.sqllitaa.close() # 查账 @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() 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}") self.sqllitaa.commit() 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 def exceltt(self,filename): yuju = [] try: if ".csv" in filename: ata = pd.read_csv(filename, encoding='GB2312', on_bad_lines='skip',header=22) # 读取表格数据 for biao in range(10): # 暂存当行数据 shuju = [] for aac in ata.loc[biao]: shuju.append(str(aac)) # id:流水号,payid:账单号,paytype:交易类型,paydate:交易时间,payname:交易名称,payto:收入/支出,paysize:交易金额 # paystu:交易状态,paybox:收支方式,payqd:记录渠道 # 插入数据 yuju.append(f"replace into zhangdan(id,payid,paytype,paydate,payname,payto,paysize,paystu,paybox,payqd) VALUES('{np.float64(shuju[9])}','{shuju[9]}','{shuju[1]}','{shuju[0]}','{shuju[2]+shuju[4]}','{shuju[5]}','{shuju[6]}','{shuju[8]}','{shuju[7]}',' 支付宝')") elif ".xls" in filename: ata = pd.read_excel(filename,header=17) for biao in range(10): # 暂存当行数据 shuju = [] for aac in ata.loc[biao]: shuju.append(str(aac)) # id:流水号,payid:账单号,paytype:交易类型,paydate:交易时间,payname:交易名称,payto:收入/支出,paysize:交易金额 # paystu:交易状态,paybox:收支方式,payqd:记录渠道 # 插入数据 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}") 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()}") zf.extractall(path=output_path, pwd=mima.encode()) logger.info(f"解压成功，文件保存至：{output_path}") except (RuntimeError, zipfile.BadZipFile) as e: logger.error(f"解压失败，原因：{e}") # RuntimeError 常因密码错误触发 except Exception as e: logger.error(f"解压过程中出现未知错误：{e}") 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 if "." not in file_name: file_name += ".zip" if file_name not in self.no_file: logger.info(f"下载文件名是{file_name}") file_path = os.path.join(download_dir, file_name) logger.info(f"下载文件路径是{file_path}") with open(file_path, 'wb') as file: for chunk in response.iter_content(chunk_size=8192): if chunk: file.write(chunk) return file_name 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 async def terminate(self): """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。"""