def unzip_kbf(input_files=None):
global tp_kbf_file_name
if input_files is None:
if global_constant.full_update or global_constant.sdcard_root_path is None:
input_file_abs_path = os.path.join(tp_path.parent.parent, 'f_input/')
else:
input_file_abs_path = os.path.join(global_constant.sdcard_root_path, '.mymoney', 'backup/')
else:
if global_constant.full_update or global_constant.sdcard_root_path is None:
input_file_abs_path = os.path.join(tp_path.parent.parent, 'f_input/', input_files)
else:
input_file_abs_path = os.path.join(global_constant.sdcard_root_path, '.mymoney', 'backup/', input_files)
if not os.path.exists(input_file_abs_path):
print(f'{input_file_abs_path}不存在')
raise FileNotFoundError(f'输入文件kbf不存在-->{input_file_abs_path}')
print(f'输入待解压kbf文件路径:{input_file_abs_path}')
if os.path.isdir(input_file_abs_path):
files = os.listdir(input_file_abs_path)
for f in files:
if f.endswith('.kbf'):
global tp_kbf_file_name
tp_kbf_file_name = f
zf = zipfile.ZipFile(os.path.join(input_file_abs_path, f))
zf.extractall(path=os.path.join(tp_path.parent.parent, 'f_input/'))
break
elif os.path.isfile(input_file_abs_path):
if input_files.lower().endswith('.kbf'):
tp_kbf_file_name = input_files
zf = zipfile.ZipFile(input_file_abs_path)
zf.extractall(path=os.path.join(tp_path.parent.parent, 'f_input/'))
pass
def ssj_kbf_sqlite_convert(input_file=None, output_file=None, convert=ConvertType.Exchanged):
"""
convert ssj data, after kbf unzip to sqlite,convert it to normal sqlite database file
:param convert: 0 means convert it auto, 1 kbf format to sqlite, 2 sqlite to kbf format
:param input_file: the mymoney.sqlite file path
:param output_file: the convert mymoney.sqlite file path
:return:
"""
if input_file is None:
input_file = os.path.join(tp_path.parent.parent, 'f_input/mymoney.sqlite')
if output_file is None:
output_file = os.path.join(tp_path.parent.parent, 'f_output/mymoney.sqlite')
sqlite_header = (0x53, 0x51, 0x4C, 0x69,
0x74, 0x65, 0x20, 0x66,
0x6F, 0x72, 0x6D, 0x61,
0x74, 0x20, 0x33, 0x0)
kbf_header = (0x0, 0x0, 0x0, 0x0,
0x0, 0x0, 0x0, 0x0,
0x0, 0x0, 0x0, 0x0,
0x0, 0x46, 0xFF, 0x0)
if global_constant.print_repeat_data_info:
read_file_header(input_file)
if os.path.exists(output_file):
os.remove(output_file)
with open(input_file, mode='rb') as f:
with open(output_file, mode='wb') as fw:
data_buffer = f.read()
if data_buffer[0] == 0x53:
kbf2sqlite = False
print("当前为SQLite文件格式")
if data_buffer[0] == 0x00:
kbf2sqlite = True
print("当前为KBF文件格式")
write_buffer = bytearray(data_buffer)
index = 0
while index < len(kbf_header) and index < len(sqlite_header):
if convert == ConvertType.Exchanged:
if kbf2sqlite is True:
write_buffer[index] = sqlite_header[index]
else:
write_buffer[index] = kbf_header[index]
elif convert == ConvertType.SqliteToKbf and not kbf2sqlite:
write_buffer[index] = kbf_header[index]
elif convert == ConvertType.KbfToSqlite and kbf2sqlite:
write_buffer[index] = sqlite_header[index]
index = index + 1
fw.write(write_buffer)
pass
if global_constant.print_repeat_data_info:
read_file_header(output_file)
pass
import zipfile
import os
def unzip_kbf(input_files=None):
zf = zipfile.ZipFile(input_files)
zf.extractall(path='./')
def ssj_kbf_sqlite_convert(input_file, output_file):
"""
convert ssj data, after kbf unzip to sqlite,convert it to normal sqlite database file
:param input_file: the mymoney.sqlite file path
:param output_file: the convert mymoney.sqlite file path
:return:
"""
sqlite_header = (0x53, 0x51, 0x4C, 0x69,
0x74, 0x65, 0x20, 0x66,
0x6F, 0x72, 0x6D, 0x61,
0x74, 0x20, 0x33, 0x0)
if os.path.exists(output_file):
os.remove(output_file)
with open(input_file, mode='rb') as f:
with open(output_file, mode='wb') as fw:
data_buffer = f.read()
write_buffer = bytearray(data_buffer)
index = 0
while index < len(sqlite_header):
write_buffer[index] = sqlite_header[index]
index = index + 1
fw.write(write_buffer)
print("convert done")
# 执行kbf文件解密
unzip_kbf("record.kbf")
ssj_kbf_sqlite_convert("mymoney.sqlite", "record_decrypt.sqlite")
转为一木记账可用的导入的模板的代码：
需要安装三方库：
pip install pandas xlwt
代码：
import sqlite3
import pandas as pd
import xlwt
# 连接到SQLite数据库，如果数据库文件不存在，会自动在当前目录创建一个
# 数据库文件名为example.db
conn = sqlite3.connect('record_decrypt.sqlite')
# 创建一个游标对象，用于执行SQL语句
cursor = conn.cursor()
sql = '''SELECT strftime('%Y-%m-%d %H:%M', a.tradeTime / 1000 + 8 * 3600, 'unixepoch') as 日期,
case
when a.type = 1 then '收入'
when a.type = 0 then '支出'
end as 收支类型,
case
when a.type = 1 then (select case
when (select b.currencyType from t_account b where b.accountPOID = a.sellerAccountPOID) = 'CNY' then a.buyerMoney
else (select round(a.buyerMoney * d.rate, 2) from
(select b.currencyType from t_account b where b.accountPOID = a.sellerAccountPOID) c,
t_exchange d where c.currencyType = d.sell)
end)
when a.type = 0 then (select case
when (select b.currencyType from t_account b where b.accountPOID = a.buyerAccountPOID) = 'CNY' then a.buyerMoney
else (select round(a.buyerMoney * d.rate, 2) from
(select b.currencyType from t_account b where b.accountPOID = a.buyerAccountPOID) c,
t_exchange d where c.currencyType = d.sell)
end)
end as 金额,
case
when a.type = 1 then (select d.name from (select b.parentCategoryPOID from t_category b
where b.categoryPOID = a.buyerCategoryPOID) c, t_category d
where c.parentCategoryPOID = d.categoryPOID)
when a.type = 0 then (select d.name from (select b.parentCategoryPOID from t_category b
where b.categoryPOID = a.sellerCategoryPOID) c, t_category d
where c.parentCategoryPOID = d.categoryPOID)
end as 类别,
case
when a.type = 1 then (select b.name from t_category b where b.categoryPOID = a.buyerCategoryPOID)
when a.type = 0 then (select b.name from t_category b where b.categoryPOID = a.sellerCategoryPOID)
end as 子类,
'日常账本' as 所属账本,
case
when a.type = 1 then (select b.name from t_account b where b.accountPOID = a.sellerAccountPOID)
when a.type = 0 then (select b.name from t_account b where b.accountPOID = a.buyerAccountPOID)
end as 收支账户,
a.memo as 备注,
(select c.name
from t_transaction_projectcategory_map b, t_tag c
where b.transactionPOID = a.transactionPOID
and b.projectCategoryPOID = c.tagPOID
and b.type = 2) as 标签,
'' as 地址
FROM t_transaction a
order by a.tradetime desc;'''
# 运行一个查询语句
select_sql = sql
# 使用游标执行查询
cursor.execute(select_sql)
# 获取所有查询结果
results = cursor.fetchall()
# 打印结果
for row in results:
print(row)
# 将结果转换为DataFrame
columns = [column[0] for column in cursor.description] # 获取列名
df = pd.DataFrame(results, columns=columns)
# 关闭游标和连接
cursor.close()
conn.close()
# 使用xlwt库将DataFrame写入Excel文件
excel_filename = '一木记账导入.xls'
# 创建一个Excel工作簿
wb = xlwt.Workbook()
# 添加一个工作表
ws = wb.add_sheet('Sheet 1')
# 将DataFrame数据写入Excel工作表
for col_num, col_data in enumerate(df.columns):
ws.write(0, col_num, col_data) # 写入列名
for row_num, row_data in enumerate(df.values):
for col_num, col_data in enumerate(row_data):
ws.write(row_num + 1, col_num, col_data) # 写入单元格数据
# 保存Excel文件
wb.save(excel_filename)
print(f'数据已成功写入到 {excel_filename}')
SELECT strftime('%Y-%m-%d', a.tradeTime / 1000, 'unixepoch') as 日期,
case
when a.type = 1 then '收入'
when a.type = 0 then '支出'
end as 收支类型,
case
when a.type = 1 then (select case
when (select b.currencyType from t_account b where b.accountPOID = a.sellerAccountPOID) = 'CNY' then a.buyerMoney
else (select round(a.buyerMoney * d.rate, 2) from
(select b.currencyType from t_account b where b.accountPOID = a.sellerAccountPOID) c,
t_exchange d where c.currencyType = d.sell)
end)
when a.type = 0 then (select case
when (select b.currencyType from t_account b where b.accountPOID = a.buyerAccountPOID) = 'CNY' then a.buyerMoney
else (select round(a.buyerMoney * d.rate, 2) from
(select b.currencyType from t_account b where b.accountPOID = a.buyerAccountPOID) c,
t_exchange d where c.currencyType = d.sell)
end)
end as 金额,
case
when a.type = 1 then (select d.name from (select b.parentCategoryPOID from t_category b
where b.categoryPOID = a.buyerCategoryPOID) c, t_category d
where c.parentCategoryPOID = d.categoryPOID)
when a.type = 0 then (select d.name from (select b.parentCategoryPOID from t_category b
where b.categoryPOID = a.sellerCategoryPOID) c, t_category d
where c.parentCategoryPOID = d.categoryPOID)
end as 类别,
case
when a.type = 1 then (select b.name from t_category b where b.categoryPOID = a.buyerCategoryPOID)
when a.type = 0 then (select b.name from t_category b where b.categoryPOID = a.sellerCategoryPOID)
end as 子类,
'日常账本' as 所属账本,
case
when a.type = 1 then (select b.name from t_account b where b.accountPOID = a.sellerAccountPOID)
when a.type = 0 then (select b.name from t_account b where b.accountPOID = a.buyerAccountPOID)
end as 收支账户,
a.memo as 备注,
(select c.name
from t_transaction_projectcategory_map b, t_tag c
where b.transactionPOID = a.transactionPOID
and b.projectCategoryPOID = c.tagPOID
and b.type = 1) as 标签,
'' as 地址
FROM t_transaction a
order by a.tradetime desc;
-- 增加一个成员字段取出，可以看到这比消费是谁消费的
SELECT strftime('%Y-%m-%d', a.tradeTime / 1000, 'unixepoch') as 日期,
case
when a.type = 1 then '收入'
when a.type = 0 then '支出'
end as 收支类型,
case
when a.type = 1 then (select case
when (select b.currencyType from t_account b where b.accountPOID = a.sellerAccountPOID) = 'CNY' then a.buyerMoney
else (select round(a.buyerMoney * d.rate, 2) from
(select b.currencyType from t_account b where b.accountPOID = a.sellerAccountPOID) c,
t_exchange d where c.currencyType = d.sell)
end)
when a.type = 0 then (select case
when (select b.currencyType from t_account b where b.accountPOID = a.buyerAccountPOID) = 'CNY' then a.buyerMoney
else (select round(a.buyerMoney * d.rate, 2) from
(select b.currencyType from t_account b where b.accountPOID = a.buyerAccountPOID) c,
t_exchange d where c.currencyType = d.sell)
end)
end as 金额,
case
when a.type = 1 then (select d.name from (select b.parentCategoryPOID from t_category b
where b.categoryPOID = a.buyerCategoryPOID) c, t_category d
where c.parentCategoryPOID = d.categoryPOID)
when a.type = 0 then (select d.name from (select b.parentCategoryPOID from t_category b
where b.categoryPOID = a.sellerCategoryPOID) c, t_category d
where c.parentCategoryPOID = d.categoryPOID)
end as 类别,
case
when a.type = 1 then (select b.name from t_category b where b.categoryPOID = a.buyerCategoryPOID)
when a.type = 0 then (select b.name from t_category b where b.categoryPOID = a.sellerCategoryPOID)
end as 子类,
'日常账本' as 所属账本,
case
when a.type = 1 then (select b.name from t_account b where b.accountPOID = a.sellerAccountPOID)
when a.type = 0 then (select b.name from t_account b where b.accountPOID = a.buyerAccountPOID)
end as 收支账户,
a.memo as 备注,
(select c.name
from t_transaction_projectcategory_map b, t_tag c
where b.transactionPOID = a.transactionPOID
and b.projectCategoryPOID = c.tagPOID
and b.type = 1) as 标签,
'' as 地址,
(select c.name
from t_transaction_projectcategory_map b, t_tag c
where b.transactionPOID = a.transactionPOID
and b.projectCategoryPOID = c.tagPOID
and b.type = 2) as 成员
FROM t_transaction a
order by a.tradetime desc;
[复制链接]
发表于 2023-9-13 14:50
|
发表于 2023-9-13 15:12
