bdzxb
3e277afddc
modified: README.md modified: __pycache__/main.cpython-312.pyc modified: _conf_schema.json new file: image.png modified: main.py deleted: static/files/ceshi.txt deleted: static/img/server-icon.png new file: tools/.flask.pid new file: tools/__pycache__/mail_tool.cpython-312.pyc new file: tools/__pycache__/zhangdan_flask.cpython-312.pyc new file: tools/mail_tool.py new file: tools/zhangdan_flask.py
290 lines
12 KiB
Python
290 lines
12 KiB
Python
from astrbot.api.event import filter, AstrMessageEvent
|
||
from astrbot.api.star import Context, Star, register, StarTools
|
||
from astrbot.api import logger, AstrBotConfig
|
||
import os
|
||
import zipfile
|
||
from pyzipper import AESZipFile
|
||
import pandas as pd
|
||
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.email_usr = None
|
||
# 授权码
|
||
self.emcil_paswd = None
|
||
# imap服务器地址
|
||
self.emcil_imap = None
|
||
# 略过附件列表
|
||
self.no_file = None
|
||
# 发件人白名单
|
||
self.emcil_white = None
|
||
# 插件文件位置
|
||
self.file_path = "data/plugins/astrbot_bdzxb_bill"
|
||
# 配置文件
|
||
self.config = config
|
||
# 插件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):
|
||
"""可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。"""
|
||
# 读取配置信息
|
||
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")
|
||
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")
|
||
# 创建游标对象
|
||
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")
|
||
# 读取配置文件 查看石头需要直接启动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 = "/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(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}")
|
||
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()
|
||
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)
|
||
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.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()
|
||
yield event.plain_result(f"{jieg}")
|
||
except Exception as e:
|
||
yield event.plain_result(f"查询失败：{e}")
|
||
logger.error(f"查询失败{e}")
|
||
self.sqllitaa.commit()
|
||
cursor.close()
|
||
self.sqllitaa.close()
|
||
|
||
# 分析文件并返回SQL语句
|
||
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.error(f"解析文件失败 原因{e}")
|
||
return
|
||
|
||
# 解压文件
|
||
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:
|
||
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}")
|
||
# 通过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
|
||
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:
|
||
logger.error(f"文件下载失败 原因：{e}")
|
||
return
|
||
|
||
async def terminate(self):
|
||
"""可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。"""
|