julian/KpsUnifier
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dev: 增加删除功能

Browse Source
This commit is contained in:
Julian Freeman
2024-07-19 22:51:03 -04:00
parent dc96e1a747
commit 1fcb707472
11 changed files with 233 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/Sqlite3Helper.py
View File

@@ -363,7 +363,7 @@ class Sqlite3Worker(object):
return ", ".join(columns_str_ls)
def insert_into(self, table_name: str, columns: list[Column | str],
values: list[list[str | int | float]],
values: list[list[str | int | float | BlobType]],
*, execute: bool = True, commit: bool = True) -> str:
col_count = len(columns)
columns_str = self._columns_to_string(columns)

9
lib/config_utils.py
View File

@@ -42,8 +42,8 @@ def get_app_dir(org_name: str, app_name: str) -> Path:
def get_config_path(org_name: str, app_name: str) -> Path:
data_dir = get_app_dir(org_name, app_name)
return Path(data_dir, "config.json")
app_dir = get_app_dir(org_name, app_name)
return Path(app_dir, "config.json")
def read_config(org_name: str, app_name: str) -> dict:
@@ -71,3 +71,8 @@ def get_default_db_path(config: dict, org_name: str, app_name: str) -> str:
app_dir = get_app_dir(org_name, app_name)
return str(app_dir / f"default.db")
return config["last_db_path"]
def get_secrets_path(org_name: str, app_name: str) -> str:
app_dir = get_app_dir(org_name, app_name)
return str(app_dir / "secrets.db")

6
lib/db_columns_def.py
View File

@@ -13,6 +13,7 @@ columns_d = {
"filepath": Column("filepath", DataType.BLOB, nullable=False),
"path": Column("path", DataType.BLOB),
"status": Column("status", DataType.TEXT), # 只有三种状态：keep, transfer, delete
"deleted": Column("deleted", DataType.INTEGER, has_default=True, default=0), # 布尔，只有 1 或者 0
}
all_columns = [
@@ -27,10 +28,11 @@ all_columns = [
columns_d["filepath"],
columns_d["path"],
columns_d["status"],
columns_d["deleted"],
]
# 插入数据时使用的列
insert_columns = all_columns[1:-1]
insert_columns = all_columns[1:-2]
# 查询数据时使用的列
query_columns = [
@@ -47,6 +49,8 @@ sim_columns = [
columns_d["filepath"],
]
uuid_col = columns_d["uuid"]
filepath_col = columns_d["filepath"]
entry_id_col = columns_d["entry_id"]
status_col = columns_d["status"]
deleted_col = columns_d["deleted"]

19
lib/sec_db_columns_def.py Normal file
View File

@@ -0,0 +1,19 @@
# coding: utf8
from .Sqlite3Helper import Column, DataType
sec_columns_d = {
"secret_id": Column("secret_id", DataType.INTEGER, primary_key=True, unique=True),
"filepath": Column("filepath", DataType.BLOB),
"password": Column("password", DataType.BLOB),
}
sec_all_columns = [
sec_columns_d["secret_id"],
sec_columns_d["filepath"],
sec_columns_d["password"],
]
insert_sec_columns = sec_all_columns[1:]
sec_filepath_col = sec_columns_d["filepath"]
sec_password_col = sec_columns_d["password"]

9
main.py
View File

@@ -3,7 +3,11 @@ import sys
from PySide6.QtWidgets import QApplication
from lib.config_utils import get_default_db_path, read_config
from lib.config_utils import (
get_default_db_path,
read_config,
get_secrets_path,
)
from src.mw_kps_unifier import KpsUnifier
import src.rc_kps_unifier
@@ -21,8 +25,9 @@ def main():
config = read_config(ORG_NAME, APP_NAME)
db_path = get_default_db_path(config, ORG_NAME, APP_NAME)
secrets_path = get_secrets_path(ORG_NAME, APP_NAME)
win = KpsUnifier(db_path, config, __version__)
win = KpsUnifier(db_path, secrets_path, config, __version__)
win.show()
return app.exec()

21
src/gbx_kps_login.py
View File

@@ -1,24 +1,30 @@
# coding: utf8
from pathlib import Path
from PySide6 import QtWidgets, QtGui, QtCore
from pykeepass import PyKeePass
from pykeepass.exceptions import CredentialsError
from lib.Sqlite3Helper import Sqlite3Worker
from lib.kps_operations import read_kps_to_db
from lib.kps_operations import read_kps_to_db, blob_fy
from lib.sec_db_columns_def import insert_sec_columns
class GbxKpsLogin(QtWidgets.QGroupBox):
def __init__(
self,
path: str,
sqh: Sqlite3Worker,
config: dict,
file_kp: dict[str, PyKeePass],
sqh: Sqlite3Worker,
sec_sqh: Sqlite3Worker,
parent: QtWidgets.QWidget = None
):
super().__init__(parent)
self.sqh = sqh
self.config = config
self.path = path
self.config = config
self.file_kp = file_kp
self.sqh = sqh
self.sec_sqh = sec_sqh
self.icon_eye = QtGui.QIcon(":/asset/img/eye.svg")
self.icon_eye_off = QtGui.QIcon(":/asset/img/eye-off.svg")
@@ -81,7 +87,7 @@ class GbxKpsLogin(QtWidgets.QGroupBox):
def on_pbn_load_clicked(self):
try:
read_kps_to_db(
kp = read_kps_to_db(
kps_file=self.lne_path.text(),
password=self.lne_password.text(),
table_name=self.config["table_name"],
@@ -92,6 +98,11 @@ class GbxKpsLogin(QtWidgets.QGroupBox):
f"{self.lne_path.text()}\n密码错误")
return
self.file_kp[self.lne_path.text()] = kp
self.sec_sqh.insert_into("secrets", insert_sec_columns, [
[blob_fy(self.lne_path.text()), blob_fy(self.lne_password.text())]
])
self.lne_password.clear()
self.set_loaded(True)
loaded_mem = self.config["loaded_memory"]

27
src/mw_kps_unifier.py
View File

@@ -1,5 +1,6 @@
# coding: utf8
from PySide6 import QtWidgets, QtCore, QtGui
from pykeepass import PyKeePass
from .page_load import PageLoad
from .page_query import PageQuery
@@ -8,6 +9,7 @@ from .page_similar import PageSimilar
from .cmbx_styles import StyleComboBox
from lib.Sqlite3Helper import Sqlite3Worker
from lib.db_columns_def import all_columns
from lib.sec_db_columns_def import sec_all_columns
from lib.config_utils import write_config
@@ -16,7 +18,9 @@ class UiKpsUnifier(object):
self,
default_db_path: str,
config: dict,
file_kp: dict[str, PyKeePass],
sqh: Sqlite3Worker,
sec_sqh: Sqlite3Worker,
window: QtWidgets.QMainWindow
):
window.setWindowTitle('KeePassXC 多合一')
@@ -56,9 +60,9 @@ class UiKpsUnifier(object):
self.sw_m = QtWidgets.QStackedWidget(self.cw)
self.vly_m.addWidget(self.sw_m)
self.page_load = PageLoad(sqh, config, self.cw)
self.page_load = PageLoad(config, file_kp, sqh, sec_sqh, self.cw)
self.sw_m.addWidget(self.page_load)
self.page_query = PageQuery(sqh, config, self.cw)
self.page_query = PageQuery(config, file_kp, sqh, sec_sqh, self.cw)
self.sw_m.addWidget(self.page_query)
self.page_similar = PageSimilar(sqh, config, self.cw)
self.sw_m.addWidget(self.page_similar)
@@ -69,14 +73,24 @@ class UiKpsUnifier(object):
class KpsUnifier(QtWidgets.QMainWindow):
def __init__(self, db_path: str, config: dict, version: str, parent=None):
def __init__(
self,
db_path: str,
secrets_path: str,
config: dict,
version: str,
parent: QtWidgets.QWidget = None,
):
super().__init__(parent)
self.db_path = db_path
self.secrets_path = secrets_path
self.config = config
self.version = version
self.file_kp: dict[str, PyKeePass] = {}
self.sqh = self.init_db()
self.sec_sqh = self.init_secrets_db()
self.ui = UiKpsUnifier(self.db_path, self.config, self.sqh, self)
self.ui = UiKpsUnifier(self.db_path, self.config, self.file_kp, self.sqh, self.sec_sqh, self)
self.ui.act_new.triggered.connect(self.on_act_new_triggered)
self.ui.act_open.triggered.connect(self.on_act_open_triggered)
@@ -139,3 +153,8 @@ class KpsUnifier(QtWidgets.QMainWindow):
def on_act_about_qt_triggered(self):
QtWidgets.QMessageBox.aboutQt(self, "关于 Qt")
def init_secrets_db(self) -> Sqlite3Worker:
sec_sqh = Sqlite3Worker(self.secrets_path)
sec_sqh.create_table("secrets", sec_all_columns, if_not_exists=True)
return sec_sqh

17
src/page_load.py
View File

@@ -2,6 +2,7 @@
import sqlite3
from pathlib import Path
from PySide6 import QtWidgets
from pykeepass import PyKeePass
from .gbx_kps_login import GbxKpsLogin
from .utils import accept_warning
@@ -11,13 +12,17 @@ from lib.Sqlite3Helper import Sqlite3Worker
class WgLoadKps(QtWidgets.QWidget):
def __init__(
self,
sqh: Sqlite3Worker,
config: dict,
file_kp: dict[str, PyKeePass],
sqh: Sqlite3Worker,
sec_sqh: Sqlite3Worker,
parent: QtWidgets.QWidget = None
):
super().__init__(parent)
self.sqh = sqh
self.sec_sqh = sec_sqh
self.config = config
self.file_kp = file_kp
self.kps_wgs: list[GbxKpsLogin] = []
self.vly_m = QtWidgets.QVBoxLayout()
@@ -38,7 +43,7 @@ class WgLoadKps(QtWidgets.QWidget):
QtWidgets.QMessageBox.warning(self, "警告", "该 KPS 文件已添加。")
return
wg = GbxKpsLogin(path, self.sqh, self.config, self)
wg = GbxKpsLogin(path, self.config, self.file_kp, self.sqh, self.sec_sqh, self)
wg.pbn_remove.clicked_with_item.connect(self.on_item_pbn_remove_clicked)
# 从倒数第二个位置插入，保证弹簧始终在最后
self.vly_m.insertWidget(self.vly_m.count() - 1, wg)
@@ -61,9 +66,11 @@ class WgLoadKps(QtWidgets.QWidget):
class PageLoad(QtWidgets.QWidget):
def __init__(
self,
sqh: Sqlite3Worker,
config: dict,
parent: QtWidgets.QWidget = None
file_kp: dict[str, PyKeePass],
sqh: Sqlite3Worker,
sec_sqh: Sqlite3Worker,
parent: QtWidgets.QWidget = None,
):
super().__init__(parent)
self.sqh = sqh
@@ -91,7 +98,7 @@ class PageLoad(QtWidgets.QWidget):
self.sa_m = QtWidgets.QScrollArea(self)
self.sa_m.setWidgetResizable(True)
self.hly_m.addWidget(self.sa_m)
self.wg_sa = WgLoadKps(sqh, config, self.sa_m)
self.wg_sa = WgLoadKps(config, file_kp, sqh, sec_sqh, self.sa_m)
self.sa_m.setWidget(self.wg_sa)
self.pbn_add_kps.clicked.connect(self.on_pbn_add_kps_clicked)

156
src/page_query.py
View File

@@ -1,10 +1,18 @@
# coding: utf8
import json
from uuid import UUID
from PySide6 import QtWidgets, QtCore, QtGui
from pykeepass import PyKeePass
from .da_entry_info import DaEntryInfo
from .utils import HorizontalLine, get_filepath_uuids_map, accept_warning
from lib.Sqlite3Helper import Sqlite3Worker, Expression, Operand
from lib.db_columns_def import query_columns, status_col, entry_id_col
from lib.db_columns_def import (
query_columns, status_col, entry_id_col,
sim_columns, deleted_col, uuid_col, filepath_col,
)
from lib.sec_db_columns_def import sec_password_col, sec_filepath_col
from lib.kps_operations import blob_fy
class QueryTableModel(QtCore.QAbstractTableModel):
@@ -21,9 +29,9 @@ class QueryTableModel(QtCore.QAbstractTableModel):
"none": QtGui.QBrush(QtCore.Qt.GlobalColor.transparent)
}
self.dark_status_colors = {
"keep": QtGui.QBrush(QtGui.QColor("green")),
"transfer": QtGui.QBrush(QtGui.QColor("orange")),
"delete": QtGui.QBrush(QtGui.QColor("orangered")),
"keep": QtGui.QBrush(QtGui.QColor("forestgreen")),
"transfer": QtGui.QBrush(QtGui.QColor("darksalmon")),
"delete": QtGui.QBrush(QtGui.QColor("darkred")),
"none": QtGui.QBrush(QtCore.Qt.GlobalColor.transparent)
}
if QtWidgets.QApplication.style().name() == "windowsvista":
@@ -60,29 +68,21 @@ class QueryTableModel(QtCore.QAbstractTableModel):
return self.headers[section]
class PageQuery(QtWidgets.QWidget):
def __init__(self, sqh: Sqlite3Worker, config: dict, parent=None):
super().__init__(parent)
self.sqh = sqh
self.config = config
class UiPageQuery(object):
def __init__(self, config: dict, window: QtWidgets.QWidget):
# 右键菜单
self.menu_ctx = QtWidgets.QMenu(self)
self.act_keep = ActionWithStr("keep", "保留", self)
self.act_transfer = ActionWithStr("transfer", "转移", self)
self.act_delete = ActionWithStr("delete", "删除", self)
self.menu_ctx = QtWidgets.QMenu(window)
self.act_keep = ActionWithStr("keep", "保留", window)
self.act_transfer = ActionWithStr("transfer", "转移", window)
self.act_delete = ActionWithStr("delete", "删除", window)
self.menu_ctx.addActions([self.act_keep, self.act_transfer, self.act_delete])
self.act_keep.triggered_with_str.connect(self.on_act_mark_triggered_with_str)
self.act_transfer.triggered_with_str.connect(self.on_act_mark_triggered_with_str)
self.act_delete.triggered_with_str.connect(self.on_act_mark_triggered_with_str)
# 主布局
self.hly_m = QtWidgets.QHBoxLayout()
self.setLayout(self.hly_m)
window.setLayout(self.hly_m)
self.sa_left = QtWidgets.QScrollArea(self)
self.sa_left = QtWidgets.QScrollArea(window)
self.sa_left.setWidgetResizable(True)
self.sa_left.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
self.sa_left.setSizePolicy(QtWidgets.QSizePolicy.Policy.Minimum, QtWidgets.QSizePolicy.Policy.Ignored)
@@ -93,26 +93,66 @@ class PageQuery(QtWidgets.QWidget):
self.sa_wg.setLayout(self.vly_sa_wg)
self.sa_left.setWidget(self.sa_wg)
self.pbn_execute = QtWidgets.QPushButton("执行操作", window)
self.pbn_execute.setMinimumWidth(config["button_min_width"])
self.vly_sa_wg.addWidget(self.pbn_execute)
self.pbn_set_target = QtWidgets.QPushButton("目标文件", window)
self.pbn_set_target.setMinimumWidth(config["button_min_width"])
self.vly_sa_wg.addWidget(self.pbn_set_target)
self.hln_1 = HorizontalLine(window)
self.vly_sa_wg.addWidget(self.hln_1)
self.pbn_all = QtWidgets.QPushButton("全部", self.sa_wg)
self.pbn_all.setMinimumWidth(config["button_min_width"])
self.vly_sa_wg.addWidget(self.pbn_all)
self.pbn_deleted = QtWidgets.QPushButton("已删除", self.sa_wg)
self.pbn_deleted.setMinimumWidth(config["button_min_width"])
self.vly_sa_wg.addWidget(self.pbn_deleted)
self.vly_sa_wg.addStretch(1)
self.pbn_read_filters = QtWidgets.QPushButton("更多过滤", self)
self.pbn_read_filters = QtWidgets.QPushButton("更多过滤", window)
self.pbn_read_filters.setMinimumWidth(config["button_min_width"])
self.vly_sa_wg.addWidget(self.pbn_read_filters)
self.trv_m = QtWidgets.QTreeView(self)
self.trv_m = QtWidgets.QTreeView(window)
self.trv_m.setSelectionMode(QtWidgets.QTreeView.SelectionMode.ExtendedSelection)
self.trv_m.setContextMenuPolicy(QtCore.Qt.ContextMenuPolicy.CustomContextMenu)
# self.trv_m.setSortingEnabled(True)
self.hly_m.addWidget(self.trv_m)
self.pbn_all.clicked.connect(self.on_pbn_all_clicked)
self.pbn_read_filters.clicked.connect(self.on_pbn_read_filters_clicked)
self.trv_m.doubleClicked.connect(self.on_trv_m_double_clicked)
self.trv_m.customContextMenuRequested.connect(self.on_trv_m_custom_context_menu_requested)
class PageQuery(QtWidgets.QWidget):
def __init__(
self,
config: dict,
file_kp: dict[str, PyKeePass],
sqh: Sqlite3Worker,
sec_sqh: Sqlite3Worker,
parent: QtWidgets.QWidget = None,
):
super().__init__(parent)
self.sqh = sqh
self.sec_sqh = sec_sqh
self.config = config
self.file_kp = file_kp
self.ui = UiPageQuery(config, self)
self.ui.act_keep.triggered_with_str.connect(self.on_act_mark_triggered_with_str)
self.ui.act_transfer.triggered_with_str.connect(self.on_act_mark_triggered_with_str)
self.ui.act_delete.triggered_with_str.connect(self.on_act_mark_triggered_with_str)
self.ui.pbn_execute.clicked.connect(self.on_pbn_execute_clicked)
self.ui.pbn_set_target.clicked.connect(self.on_pbn_set_target_clicked)
self.ui.pbn_all.clicked.connect(self.on_pbn_all_clicked)
self.ui.pbn_deleted.clicked.connect(self.on_pbn_deleted_clicked)
self.ui.pbn_read_filters.clicked.connect(self.on_pbn_read_filters_clicked)
self.ui.trv_m.doubleClicked.connect(self.on_trv_m_double_clicked)
self.ui.trv_m.customContextMenuRequested.connect(self.on_trv_m_custom_context_menu_requested)
self.set_default_filters()
@@ -135,9 +175,9 @@ class PageQuery(QtWidgets.QWidget):
self.set_filter_button(fil)
def set_filter_button(self, fil: dict):
pbn_fil = PushButtonWithData(fil, self.sa_wg, fil["name"])
pbn_fil = PushButtonWithData(fil, self.ui.sa_wg, fil["name"])
pbn_fil.setMinimumWidth(self.config["button_min_width"])
self.vly_sa_wg.insertWidget(self.vly_sa_wg.count() - 2, pbn_fil)
self.ui.vly_sa_wg.insertWidget(self.ui.vly_sa_wg.count() - 2, pbn_fil)
pbn_fil.clicked_with_data.connect(self.on_custom_filters_clicked_with_data)
def on_pbn_read_filters_clicked(self):
@@ -153,17 +193,24 @@ class PageQuery(QtWidgets.QWidget):
def on_custom_filters_clicked_with_data(self, data: dict):
_, results = self.sqh.select(self.config["table_name"], query_columns,
where=Expression(data["where"]))
where=Expression(data["where"]).and_(Operand(deleted_col).equal_to(0)))
model = QueryTableModel(results, self)
self.trv_m.setModel(model)
self.ui.trv_m.setModel(model)
def update_sqh(self, sqh: Sqlite3Worker):
self.sqh = sqh
def on_pbn_all_clicked(self):
_, results = self.sqh.select(self.config["table_name"], query_columns)
_, results = self.sqh.select(self.config["table_name"], query_columns,
where=Operand(deleted_col).equal_to(0))
model = QueryTableModel(results, self)
self.trv_m.setModel(model)
self.ui.trv_m.setModel(model)
def on_pbn_deleted_clicked(self):
_, results = self.sqh.select(self.config["table_name"], query_columns,
where=Operand(deleted_col).equal_to(1))
model = QueryTableModel(results, self)
self.ui.trv_m.setModel(model)
def on_trv_m_double_clicked(self, index: QtCore.QModelIndex):
entry_id = index.siblingAtColumn(0).data(QtCore.Qt.ItemDataRole.DisplayRole)
@@ -171,10 +218,10 @@ class PageQuery(QtWidgets.QWidget):
da_entry_info.exec()
def on_trv_m_custom_context_menu_requested(self, pos: QtCore.QPoint):
self.menu_ctx.exec(self.trv_m.viewport().mapToGlobal(pos))
self.ui.menu_ctx.exec(self.ui.trv_m.viewport().mapToGlobal(pos))
def on_act_mark_triggered_with_str(self, info: str):
indexes = self.trv_m.selectedIndexes()
indexes = self.ui.trv_m.selectedIndexes()
entry_ids = [
index.data(QtCore.Qt.ItemDataRole.DisplayRole)
for index in indexes if index.column() == 0
@@ -183,6 +230,47 @@ class PageQuery(QtWidgets.QWidget):
self.sqh.update(self.config["table_name"], [(status_col, info)],
where=Operand(entry_id_col).in_(entry_ids))
def on_pbn_execute_clicked(self):
if accept_warning(self, True, "警告", "你确定要执行转移和删除操作吗？"):
return
# 删除功能
_, results = self.sqh.select(self.config["table_name"], sim_columns,
where=Operand(status_col).equal_to("delete"))
file_uuids = get_filepath_uuids_map(results)
total, success, invalid = 0, 0, 0
for file in file_uuids:
if file not in self.file_kp:
_, results = self.sec_sqh.select("secrets", [sec_password_col],
where=Operand(sec_filepath_col).equal_to(blob_fy(file)))
password = results[-1][0].decode("utf-8")
kp = PyKeePass(file, password)
else:
kp = self.file_kp[file]
for u in file_uuids[file]:
total += 1
self.sqh.update(self.config["table_name"], [(deleted_col, 1)],
where=Operand(uuid_col).equal_to(u).and_(
Operand(filepath_col).equal_to(blob_fy(file))))
entry = kp.find_entries(uuid=UUID(u), first=True)
if entry is None:
invalid += 1
continue
kp.delete_entry(entry)
success += 1
kp.save()
QtWidgets.QMessageBox.information(self, "提示",
f"共 {total} 条标记的条目，已删除 {success} 条，无效 {invalid} 条。")
def on_pbn_set_target_clicked(self):
pass
class PushButtonWithData(QtWidgets.QPushButton):

9
src/page_similar.py
View File

@@ -4,7 +4,7 @@ from uuid import UUID
from PySide6 import QtWidgets, QtCore
from PySide6.QtCore import QAbstractTableModel
from .utils import accept_warning
from .utils import accept_warning, get_filepath_uuids_map
from lib.Sqlite3Helper import Sqlite3Worker, Operand
from lib.db_columns_def import sim_columns, filepath_col
from lib.config_utils import path_not_exist
@@ -71,12 +71,7 @@ class PageSimilar(QtWidgets.QWidget):
def on_pbn_read_db_clicked(self):
_, results = self.sqh.select(self.config["table_name"], sim_columns)
file_uuids: dict[str, list[UUID]] = {}
for u, filepath in results:
filepath = filepath.decode("utf8")
if filepath not in file_uuids:
file_uuids[filepath] = []
file_uuids[filepath].append(u)
file_uuids = get_filepath_uuids_map(results)
files = file_uuids.keys()
if len(files) < 2:

19
src/utils.py
View File

@@ -9,3 +9,22 @@ def accept_warning(widget: QtWidgets.QWidget, condition: bool,
if b == QtWidgets.QMessageBox.StandardButton.No:
return True
return False
class HorizontalLine(QtWidgets.QFrame):
def __init__(self, parent: QtWidgets.QWidget = None):
super().__init__(parent)
self.setFrameShape(QtWidgets.QFrame.Shape.HLine)
self.setFrameShadow(QtWidgets.QFrame.Shadow.Sunken)
def get_filepath_uuids_map(query_results: list[tuple]) -> dict[str, list[str]]:
file_uuids: dict[str, list[str]] = {}
for u, filepath in query_results:
filepath = filepath.decode("utf8")
if filepath not in file_uuids:
file_uuids[filepath] = []
file_uuids[filepath].append(u)
return file_uuids