new file: .idea/.gitignore

new file: .idea/astrbot_plugin_bd_mc.iml new file: .idea/inspectionProfiles/Project_Default.xml new file: .idea/misc.xml new file: .idea/modules.xml new file: .idea/vcs.xml new file: LICENSE new file: __pycache__/main.cpython-311.pyc new file: __pycache__/main.cpython-312.pyc new file: __pycache__/main.cpython-313.pyc new file: _conf_schema.json new file: main.py new file: metadata.yaml new file: static/files/ceshi.txt new file: static/img/server-icon.png
2026-05-13 23:37:42 +08:00
parent 11f31fadd3
commit 7f05c25847
15 changed files with 978 additions and 0 deletions
@@ -0,0 +1,230 @@
+
+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 requests
+from urllib.parse import urlparse
+@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.emcil_white = None
+        self.file_path = "data/plugins/astrbot_bdzxb_bill"
+        self.config = config
+        self.chatid = 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
+    # 登录邮箱
+    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")
+            # 设定查询条件 
+            from_clause = " OR ".join([f'FROM "{sender}"' for sender in self.emcil_white])
+             # 获取今天的日期，格式化为IMAP要求的格式(如"21-Jul-2025")
+            today = datetime.now().strftime("%d-%b-%Y")
+            # 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, from_clause)
+            if status != "OK":  # 检查搜索是否成功
+                raise Exception("邮件搜索失败")
+
+             # 获取邮件ID列表并只取最后20个（最新的20封）
+            mail_ids = messages[0].split()[-20:]
+            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]  # 解析发件人邮箱
+                if from_header not in self.emcil_white:  # 严格检查白名单
+                    logger.debug(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容器部分
+                        continue
+
+                    # 检查是否为.zip附件
+                    # ==================
+                    filename = part.get_filename()  # 获取附件文件名
+                    # 获取正文
+                    maildata = part.get_payload(decode=True)
+                    # 微信导出文件需要点击链接下载 检测正文中是否包含链接
+                    etree = html.etree
+                    tree = etree.HTML(maildata)
+                    a_tags = tree.xpath('//a')
+                    for a_tag in a_tags:
+                        # 获取到url后下载url
+                        urlt = a_tag.get("href")
+                        if urlt:
+                            filenames += self.download_file(urlt=urlt)
+                    content_type = part.get_content_type()  # 获取内容类型
+                    # 判断条件：
+                    # 1. 文件名以.zip结尾，或
+                    # 2. 内容类型是Excel文件
+                    # 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"
+
+                    # 文件名处理
+                    # =========
+                    # 解码邮件头中的文件名
+                    decoded_name = email.header.decode_header(filename)[0][0]
+                    if isinstance(decoded_name, bytes):
+                        decoded_name = decoded_name.decode('utf-8')
+
+                    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 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}")  
+                        
+
+               
+            return "获取成功:"+filenames
+            # 查看白名单邮件
+            # if from_header not in self.emcil_white:  # 严格检查白名单
+            #     logger.debug(f"跳过非白名单发件人: {from_header}")
+        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}")
+
+
+
+
+    def download_file(self,urlt):
+        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
+
+#保存文件
+    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):
+        """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。"""