diff --git a/__pycache__/main.cpython-312.pyc b/__pycache__/main.cpython-312.pyc index 382d3f5..bec6fdf 100644 Binary files a/__pycache__/main.cpython-312.pyc and b/__pycache__/main.cpython-312.pyc differ diff --git a/_conf_schema.json b/_conf_schema.json index b406af6..f3805f8 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -25,5 +25,11 @@ "type": "list", "hint": "只接受指定发件人的邮件", "obvious_hint": true + }, + "no_file": { + "description": "屏蔽附件名称", + "type": "list", + "hint": "屏蔽干扰附件", + "obvious_hint": true } } \ No newline at end of file diff --git a/main.py b/main.py index 43e4347..9acbee2 100644 --- a/main.py +++ b/main.py @@ -24,9 +24,14 @@ 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): @@ -36,11 +41,13 @@ class MyPlugin(Star): 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): """可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。""" # 读取配置信息 @@ -48,7 +55,46 @@ class MyPlugin(Star): 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") + # 登录邮箱 下载文件 def emaillog(self): logger.info(f" 开始执行 登录邮箱 {self.email_usr} 命令 ") # 邮箱登录用户名 @@ -59,53 +105,45 @@ class MyPlugin(Star): imap_server = self.emcil_imap mail = imaplib.IMAP4_SSL(imap_server) # 使用用户名和授权码登录邮箱 - try: - mail.login(email_user, email_password) - # 查询收件箱 - mail.select("inbox") - # 设定查询条件 - from_clause = " OR ".join([f'FROM "{sender}"' for sender in self.emcil_white]) - # 获取今天的日期，格式化为IMAP要求的格式(如"21-Jul-2025") + # 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}" HAS attachment)' + 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}") + logger.info(f"搜索条件: {search_criteria}") # 执行IMAP搜索命令 - status, messages = mail.search(None, from_clause) + status, messages = mail.search(None, search_criteria) if status != "OK": # 检查搜索是否成功 raise Exception("邮件搜索失败") - # 获取邮件ID列表并只取最后20个（最新的20封） + # 获取邮件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对象 + # 解析邮件内容为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.debug(f"跳过非白名单发件人: {from_header}") + logger.info(f"跳过非白名单发件人: {from_header}") continue - # # 邮件日期验证 - # # =========== - # try: - # # 解析邮件日期头 - # mail_date = email.utils.parsedate_to_datetime(email_message['Date']) if email_message['Date'] else None - # if mail_date and mail_date.date() != datetime.now().date(): # 如果不是今天则跳过 - # logger.debug(f"跳过非今天({mail_date.date()})收到的邮件") - # continue - # elif not mail_date: # 如果邮件没有日期信息则默认处理 - # logger.debug("邮件无日期信息，默认处理") - # except Exception as e: # 日期解析错误处理 - # logger.warning(f"日期解析错误: {str(e)}，默认处理") - # 遍历邮件各部分 + + # 遍历邮件各部分 # ============= for part in email_message.walk(): # 递归遍历邮件所有部分 if part.get_content_maintype() == "multipart": # 跳过multipart容器部分 @@ -119,12 +157,18 @@ class MyPlugin(Star): # 微信导出文件需要点击链接下载 检测正文中是否包含链接 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: - filenames += self.download_file(urlt=urlt) + try: + filenames += self.download_file(urlt=urlt,filenamett=f"{from_header}{mail_date}") + except Exception as e: + logger.debug(f"URl 文件下载失败：{e}") content_type = part.get_content_type() # 获取内容类型 # 判断条件： # 1. 文件名以.zip结尾，或 @@ -132,18 +176,27 @@ class MyPlugin(Star): # if (filename and filename.lower().endswith(".zip")) or \ # (content_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"): + + # 处理无文件名的情况 # ================ if not filename: # 如果附件没有文件名 - # 使用邮件ID生成唯一文件名 - filename = f"attachment_{mail_id.decode()}.zip" + # 按日期发件人生成附件名 + + filename = f"{from_header}{mail_date}.zip" # 文件名处理 # ========= # 解码邮件头中的文件名 - decoded_name = email.header.decode_header(filename)[0][0] - if isinstance(decoded_name, bytes): - decoded_name = decoded_name.decode('utf-8') + 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) # 拆分文件名和扩展名 @@ -153,62 +206,182 @@ class MyPlugin(Star): new_filename = filename # 拼接完整的文件保存路径 filepath = os.path.join(StarTools.get_data_dir("astrbot_bdzxb_bill"), new_filename) - - # 附件保存处理 - # =========== - 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}") - else: # 文件已存在则跳过 - logger.info(f"附件已存在，跳过: {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 + + return "获取成功:"+filenames # 查看白名单邮件 # if from_header not in self.emcil_white: # 严格检查白名单 # logger.debug(f"跳过非白名单发件人: {from_header}") - except Exception as e: - logger.error(f" 邮箱登陆失败 抛出异常：{e}") - return "邮件搜索失败" + # except Exception as e: + # logger.error(f" 邮箱登陆失败 抛出异常：{e}") + # return "邮件搜索失败" - #帮助 + # 帮助 @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): - # 登录邮箱 - emaillogtxt = self.emaillog() - yield event.plain_result(f"{emaillogtxt}") + # # 登录邮箱 下载文件 + 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") + 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): + # logger.info(f"执行语句：{icc}") + cur.execute(icc) + self.sqllitaa.commit() + cur.close() + self.sqllitaa.close() + yield event.plain_result("保存完成__使用/chazhang 查询最后10条记录") + except Exception as 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: + logger.info(f"查询失败{e}") + self.sqllitaa.commit() + cursor.close() + self.sqllitaa.close() + + 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): - response = requests.get(urlt, stream=True) - response.raise_for_status() + 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) - if "." not in file_name: - file_name += ".zip" - 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 + 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: