dev: 初步支持加载kps

lib/db_columns_def.py

@@ -0,0 +1,23 @@
+# coding: utf8
+from .Sqlite3Helper import Column, DataType
+
+columns_d = {
+    "entry_id": Column("entry_id", DataType.INTEGER, primary_key=True),
+    "title": Column("title", DataType.BLOB),
+    "username": Column("username", DataType.BLOB),
+    "password": Column("password", DataType.BLOB),
+    "opt": Column("opt", DataType.TEXT),
+    "url": Column("url", DataType.BLOB),
+    "notes": Column("notes", DataType.BLOB),
+    "path": Column("path", DataType.BLOB, nullable=False),
+}
+all_columns = [
+    columns_d["entry_id"],
+    columns_d["title"],
+    columns_d["username"],
+    columns_d["password"],
+    columns_d["opt"],
+    columns_d["url"],
+    columns_d["notes"],
+    columns_d["path"],
+]

lib/kps_operations.py

@@ -0,0 +1,47 @@
+# coding: utf8
+from os import PathLike
+from pykeepass import PyKeePass
+from .Sqlite3Helper import Sqlite3Worker, BlobType, Column
+
+
+def trim_str(value):
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value.strip()
+    return value
+
+
+def extract_otp(otp: str) -> str:
+    if otp is None:
+        return ""
+    params = otp.split("?", 1)[1]
+    secret = params.split("&")[0]
+    return secret.split("=")[1]
+
+
+def blob_fy(value: str) -> BlobType:
+    if value is None:
+        return BlobType()
+    return BlobType(value.encode("utf-8"))
+
+
+def read_kps_to_db(kps_file: str | PathLike[str], password: str,
+                   columns: list[Column], sqh: Sqlite3Worker) -> PyKeePass:
+    kp = PyKeePass(kps_file, password=password)
+
+    values = []
+    for group in kp.groups:
+        for entry in group.entries:
+            values.append([
+                blob_fy(trim_str(entry.title)),
+                blob_fy(trim_str(entry.username)),
+                blob_fy(entry.password),
+                extract_otp(entry.otp),
+                blob_fy(trim_str(entry.url)),
+                blob_fy(entry.notes),
+                blob_fy("::".join(entry.path[:-1])),
+            ])
+
+    sqh.insert_into("entries", columns, values)
+    return kp