diff --git a/.idea/misc.xml b/.idea/misc.xml index d50a49f..f141e52 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -1,8 +1,9 @@ + - + \ No newline at end of file diff --git a/README.md b/README.md index 21e7af2..83ee549 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,47 @@ # astrbot_bdzxb_bill -astrbot插件，通过微信支付宝导出账单到邮箱，然后进行解析统计 \ No newline at end of file +astrbot插件，通过微信支付宝导出账单到邮箱，然后进行解析统计 + +## 需要导入的包 + +* sqlite3 +* numpy +* urllib.parse +* requests +* pandas +* zipfile +* os +* pyzipper +* re +* lxml +* datetime +* imaplib +* secrets +* threading +* flask +* flask_cors + +--- + +## 配置文件 + +* ![alt text](image.png) + +## 使用方法 + +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 diff --git a/__pycache__/main.cpython-312.pyc b/__pycache__/main.cpython-312.pyc index c167f47..0a94aa9 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 f3805f8..bdd036b 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -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" } + } \ No newline at end of file diff --git a/image.png b/image.png new file mode 100644 index 0000000..c653271 Binary files /dev/null and b/image.png differ diff --git a/main.py b/main.py index 55bba06..fc8d6dd 100644 --- a/main.py +++ b/main.py @@ -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): """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。""" diff --git a/static/files/ceshi.txt b/static/files/ceshi.txt deleted file mode 100644 index 1a237b6..0000000 --- a/static/files/ceshi.txt +++ /dev/null @@ -1 +0,0 @@ -这是一个测试文件 \ No newline at end of file diff --git a/static/img/server-icon.png b/static/img/server-icon.png deleted file mode 100644 index df32239..0000000 Binary files a/static/img/server-icon.png and /dev/null differ diff --git a/tools/.flask.pid b/tools/.flask.pid new file mode 100644 index 0000000..dda085b --- /dev/null +++ b/tools/.flask.pid @@ -0,0 +1 @@ +24764 \ No newline at end of file diff --git a/tools/__pycache__/mail_tool.cpython-312.pyc b/tools/__pycache__/mail_tool.cpython-312.pyc new file mode 100644 index 0000000..49ba9dc Binary files /dev/null and b/tools/__pycache__/mail_tool.cpython-312.pyc differ diff --git a/tools/__pycache__/zhangdan_flask.cpython-312.pyc b/tools/__pycache__/zhangdan_flask.cpython-312.pyc new file mode 100644 index 0000000..d9e0e46 Binary files /dev/null and b/tools/__pycache__/zhangdan_flask.cpython-312.pyc differ diff --git a/tools/mail_tool.py b/tools/mail_tool.py new file mode 100644 index 0000000..45a7f01 --- /dev/null +++ b/tools/mail_tool.py @@ -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 + \ No newline at end of file diff --git a/tools/zhangdan_flask.py b/tools/zhangdan_flask.py new file mode 100644 index 0000000..0c34f39 --- /dev/null +++ b/tools/zhangdan_flask.py @@ -0,0 +1,1062 @@ +#!/usr/bin/env python3 +""" +账单查询服务 - Flask版 +从 zhangdan.db 读取数据，提供网页展示和API接口 +""" + +import sqlite3 +import os +import secrets +import threading +from flask import Flask, render_template_string, jsonify, request, session, redirect, url_for +from flask_cors import CORS + + + +# ===== 🔧 常量 ===== +PORT = 5001 +PID_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".flask.pid") + +# ===== 🔧 全局状态 ===== +_server = None # werkzeug server 实例 +_server_thread = None # 服务线程 +_server_lock = threading.Event() + +# 启动 +def runwebui(web_port,web_passwd,datadb): + global AUTH_PASSWORD,DB_PATH,AUTH_COOKIE_MAX_AGE + try: + AUTH_PASSWORD = web_passwd + AUTH_COOKIE_MAX_AGE = 86400 * 1 # 1天免登录 + DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), datadb) + start_server(host="0.0.0.0", port=web_port, debug=False ) + except EOFError as e: + print(e) + + +def start_server(host, port, debug): + """在子线程中启动 Flask 服务，主线程不受影响""" + global _server, _server_thread + + if _server_thread and _server_thread.is_alive(): + print("⚠️ 服务已在运行中") + return _server + + from werkzeug.serving import make_server + + _server = make_server(host, port, app, threaded=True) + _server_thread = threading.Thread(target=_server.serve_forever, daemon=True) + _server_thread.start() + _server_lock.set() + + # 写 PID + with open(PID_FILE, "w") as f: + f.write(str(os.getpid())) + + print(f"🚀 服务已启动 (子线程) → http://{host}:{port}") + return _server + + +def stop_server(): + """只关闭 Flask 子线程，主线程继续运行""" + + global _server, _server_thread + + if _server: + try: + _server.shutdown() # 优雅关闭 werkzeug server + except Exception: + pass + _server = None + + if _server_thread and _server_thread.is_alive(): + _server_thread.join(timeout=3) + _server_thread = None + + _server_lock.clear() + + # 清理 PID 文件 + if os.path.exists(PID_FILE): + try: + os.remove(PID_FILE) + except OSError: + pass + + print("🛑 Flask 服务已关闭（主线程仍在运行）") + + +def server_status(): + """检查 Flask 子线程是否在运行""" + alive = _server_thread is not None and _server_thread.is_alive() + return {"running": alive, "port": PORT} + +app = Flask(__name__) +app.secret_key = secrets.token_hex(32) +CORS(app) +def check_auth(): + """检查是否已登录""" + return session.get("authenticated") is True + + +def login_required(f): + """登录装饰器""" + from functools import wraps + @wraps(f) + def decorated(*args, **kwargs): + if not check_auth(): + return redirect(url_for("login_page")) + return f(*args, **kwargs) + return decorated + +def get_db(): + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + return conn + + +@app.route("/login", methods=["GET", "POST"]) +def login_page(): + error = None + if request.method == "POST": + pwd = request.form.get("password", "") + if pwd == AUTH_PASSWORD: + session["authenticated"] = True + session.permanent = True + app.permanent_session_lifetime = __import__("datetime").timedelta(seconds=AUTH_COOKIE_MAX_AGE) + return redirect(url_for("index")) + error = "密码不对哦，再试试 🤔" + return render_template_string(LOGIN_TEMPLATE, error=error) + + +@app.route("/logout") +def logout(): + session.clear() + return redirect(url_for("login_page")) + + +# ===== 🔴 关机密令 ===== +SHUTDOWN_TOKEN = "shutdown_2026" + + +@app.route("/shutdown") +def shutdown_server(): + """通过密令关闭服务，访问 /shutdown?token=xxx 即可关机""" + token = request.args.get("token", "") + if token != SHUTDOWN_TOKEN: + return "❌ 无效密令", 403 + # 在新线程中关闭，避免阻塞当前请求 + threading.Thread(target=stop_server, daemon=True).start() + return "🛑 服务已关闭" + + +@app.route("/") +@login_required +def index(): + return render_template_string(HTML_TEMPLATE, shutdown_token=SHUTDOWN_TOKEN) + + +@app.route("/api/bills") +@login_required +def api_bills(): + """返回所有账单数据，支持排序和筛选""" + conn = get_db() + cur = conn.cursor() + + # 筛选参数 + payto = request.args.get("payto") # 收入/支出 + payqd = request.args.get("payqd") # 渠道 + paybox = request.args.get("paybox") # 钱包 + paytype = request.args.get("paytype") # 交易类型 + keyword = request.args.get("keyword") # 搜索关键词 + date_from = request.args.get("date_from") # 起始日期 + date_to = request.args.get("date_to") # 截止日期 + + # 排序参数 + sort = request.args.get("sort", "paydate") + order = request.args.get("order", "DESC") + + allowed_sorts = { + "paydate", "paysize", "paytype", "payname", + "payto", "paystu", "paybox", "payqd", "id" + } + if sort not in allowed_sorts: + sort = "paydate" + order = "ASC" if order.upper() == "ASC" else "DESC" + + query = "SELECT * FROM zhangdan WHERE 1=1" + params = [] + + if payto: + query += " AND payto = ?" + params.append(payto) + if payqd: + query += " AND payqd = ?" + params.append(payqd) + if paybox: + query += " AND paybox = ?" + params.append(paybox) + if paytype: + query += " AND paytype = ?" + params.append(paytype) + if keyword: + query += " AND (payname LIKE ? OR paytype LIKE ?)" + params.extend([f"%{keyword}%", f"%{keyword}%"]) + if date_from: + query += " AND paydate >= ?" + params.append(date_from) + if date_to: + query += " AND paydate <= ?" + params.append(date_to + " 23:59:59") + + # paysize 存的是文本，需要特殊处理排序 + if sort == "paysize": + query += f" ORDER BY CAST({sort} AS REAL) {order}" + else: + query += f" ORDER BY {sort} {order}" + + cur.execute(query, params) + rows = [dict(r) for r in cur.fetchall()] + conn.close() + return jsonify(rows) + + +@app.route("/api/summary") +@login_required +def api_summary(): + """汇总统计 — 支持与 /api/bills 相同的筛选参数""" + conn = get_db() + cur = conn.cursor() + + # 收集筛选条件（与 /api/bills 一致） + payto = request.args.get("payto") + payqd = request.args.get("payqd") + paybox = request.args.get("paybox") + paytype = request.args.get("paytype") + keyword = request.args.get("keyword") + date_from = request.args.get("date_from") + date_to = request.args.get("date_to") + + where = "WHERE 1=1" + params = [] + if payto: + where += " AND payto = ?" + params.append(payto) + if payqd: + where += " AND payqd = ?" + params.append(payqd) + if paybox: + where += " AND paybox = ?" + params.append(paybox) + if paytype: + where += " AND paytype = ?" + params.append(paytype) + if keyword: + where += " AND (payname LIKE ? OR paytype LIKE ?)" + params.extend([f"%{keyword}%", f"%{keyword}%"]) + if date_from: + where += " AND paydate >= ?" + params.append(date_from) + if date_to: + where += " AND paydate <= ?" + params.append(date_to + " 23:59:59") + + # 总收入/支出 + cur.execute(f""" + SELECT payto, + COUNT(*) as count, + SUM(CAST(paysize AS REAL)) as total + FROM zhangdan {where} + GROUP BY payto + """, params) + overview = [dict(r) for r in cur.fetchall()] + + # 按类型汇总 + cur.execute(f""" + SELECT paytype, payto, + COUNT(*) as count, + SUM(CAST(paysize AS REAL)) as total + FROM zhangdan {where} + GROUP BY paytype, payto + ORDER BY total DESC + """, params) + by_type = [dict(r) for r in cur.fetchall()] + + # 按渠道汇总 + cur.execute(f""" + SELECT payqd, payto, + COUNT(*) as count, + SUM(CAST(paysize AS REAL)) as total + FROM zhangdan {where} + GROUP BY payqd, payto + ORDER BY total DESC + """, params) + by_channel = [dict(r) for r in cur.fetchall()] + + # 按钱包汇总 + cur.execute(f""" + SELECT paybox, payto, + COUNT(*) as count, + SUM(CAST(paysize AS REAL)) as total + FROM zhangdan {where} + GROUP BY paybox, payto + ORDER BY total DESC + """, params) + by_wallet = [dict(r) for r in cur.fetchall()] + + conn.close() + return jsonify({ + "overview": overview, + "by_type": by_type, + "by_channel": by_channel, + "by_wallet": by_wallet, + }) + + +@app.route("/api/filters") +@login_required +def api_filters(): + """获取所有可选的筛选值""" + conn = get_db() + cur = conn.cursor() + result = {} + for col in ["payto", "payqd", "paybox", "paytype"]: + cur.execute(f"SELECT DISTINCT {col} FROM zhangdan ORDER BY {col}") + result[col] = [r[0] for r in cur.fetchall()] + conn.close() + return jsonify(result) + + + +LOGIN_TEMPLATE = r""" + + + + +🔐 验证身份 + + + + +
+
🔒
+

账单查询

+

请输入密码以继续

+ + {% if error %} +
{{ error }}
+ {% endif %} + +
+
+ + +
+ +
+ +

数据已加密保护 🔐

+
+ + + + +""" + +HTML_TEMPLATE = r""" + + + + +💰 账单查询 + + + + +
+
+

💰 账单查询

+

数据来源: zhangdan.db · Flask API 驱动

+
+ +
+ +
+
按类型
+
按渠道
+
按钱包
+
+
+ +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+ + +
+ +
+ + + +
+
+ + +
+ · Flask + SQLite · 0 条记录 + · 退出登录 🔒 + · 关闭服务 ⏻ +
+
+ + + + +"""