modified: __pycache__/main.cpython-312.pyc

modified:   _conf_schema.json
	modified:   main.py

_conf_schema.json

@@ -25,5 +25,11 @@
     "type": "list",
     "hint": "只接受指定发件人的邮件",
     "obvious_hint": true
+  },
+    "no_file": {
+    "description": "屏蔽附件名称",
+    "type": "list",
+    "hint": "屏蔽干扰附件",
+    "obvious_hint": true
   }
 }

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:
+        # try:
-            mail.login(email_user, email_password)
+        mail.login(email_user, email_password)
-            # 查询收件箱
+        # 查询收件箱 只读模式
-            mail.select("inbox")
+        mail.select("inbox",readonly=True)
-            # 设定查询条件 
+        # 设定查询条件 
-            from_clause = " OR ".join([f'FROM "{sender}"' for sender in self.emcil_white])
+        for emlname in self.emcil_white:
-             # 获取今天的日期，格式化为IMAP要求的格式(如"21-Jul-2025")
+            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]
+                    try:
-                    if isinstance(decoded_name, bytes):
+                        decoded_name = email.header.decode_header(filename)[0][0]
-                        decoded_name = decoded_name.decode('utf-8')
+                        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)  # 拆分文件名和扩展名
@@ -154,61 +207,181 @@ class MyPlugin(Star):
                     # 拼接完整的文件保存路径
                     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):  # 检查文件是否已存在
+                        # ===========
-                        # 以二进制模式保存附件
+                        if not os.path.exists(filepath):  # 检查文件是否已存在
-                        with open(filepath, "wb") as f:  
+                            # 以二进制模式保存附件
-                            f.write(part.get_payload(decode=True))  # 解码并写入附件内容
+                            with open(filepath, "wb") as f:  
-                        logger.info(f"下载附件: {filename} -> {new_filename}")  
+                                f.write(part.get_payload(decode=True))  # 解码并写入附件内容
-                    else:  # 文件已存在则跳过
+                            logger.info(f"下载附件: {filename} -> {new_filename}")  
-                        logger.info(f"附件已存在，跳过: {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:
+        # except Exception as e:
-            logger.error(f" 邮箱登陆失败 抛出异常：{e}")
+        #     logger.error(f" 邮箱登陆失败 抛出异常：{e}")
-            return "邮件搜索失败"
+        #     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()
+        self.emaillog()
-        yield event.plain_result(f"{emaillogtxt}")
+        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):
+    def download_file(self,urlt,filenamett):
-        response = requests.get(urlt, stream=True)
+        try:
-        response.raise_for_status()
+            response = requests.get(urlt, stream=True)
+            response.raise_for_status()
 
-        download_dir = StarTools.get_data_dir("astrbot_bdzxb_bill")
+            download_dir = StarTools.get_data_dir("astrbot_bdzxb_bill")
-        os.makedirs(download_dir, exist_ok=True)
+            os.makedirs(download_dir, exist_ok=True)
-        file_name = os.path.basename(urlparse(urlt).path)
+            file_name = os.path.basename(urlparse(urlt).path)+filenamett
-        if "." not in file_name:
+            if "." not in file_name:
-            file_name += ".zip"
+                file_name += ".zip"
-        logger.info(f"下载文件名是{file_name}")
+            if file_name not in self.no_file:
-        file_path = os.path.join(download_dir, file_name)
+                logger.info(f"下载文件名是{file_name}")
-        logger.info(f"下载文件路径是{file_path}")
+                file_path = os.path.join(download_dir, file_name)
-        with open(file_path, 'wb') as file:
+                logger.info(f"下载文件路径是{file_path}")
-            for chunk in response.iter_content(chunk_size=8192):
+                with open(file_path, 'wb') as file:
-                if chunk:
+                    for chunk in response.iter_content(chunk_size=8192):
-                    file.write(chunk)
+                        if chunk:
+                            file.write(chunk)
+                return file_name
+            else:
+                return
+        except Exception as e:
+             return 
    
-        return file_name
 
 #保存文件
     def get_json_path(self, filename: str) -> Path: