231 lines
6.6 KiB
Python
231 lines
6.6 KiB
Python
import sqlite3
|
||
import os
|
||
import shutil
|
||
|
||
# 数据库放到 Windows 常见的 APPDATA 目录（避免写在项目目录）
|
||
APP_NAME = "CleanDesktopOrganizer"
|
||
|
||
_PROJECT_DB_PATH = os.path.join(
|
||
os.path.dirname(os.path.dirname(__file__)), "data.db"
|
||
)
|
||
|
||
_BASE_DIR = os.environ.get("APPDATA") or os.environ.get("LOCALAPPDATA") or ""
|
||
DB_DIR = os.path.join(_BASE_DIR, APP_NAME) if _BASE_DIR else None
|
||
DB_PATH = os.path.join(DB_DIR, "data.db") if DB_DIR else _PROJECT_DB_PATH
|
||
|
||
|
||
def _ensure_db_dir():
|
||
if DB_DIR:
|
||
os.makedirs(DB_DIR, exist_ok=True)
|
||
|
||
|
||
def _migrate_old_db_if_needed():
|
||
"""
|
||
首次升级：把旧版项目目录下的 data.db 自动拷贝到 APPDATA，
|
||
避免你之前分组/设置丢失。
|
||
"""
|
||
try:
|
||
if os.path.isfile(_PROJECT_DB_PATH) and not os.path.isfile(DB_PATH):
|
||
_ensure_db_dir()
|
||
shutil.copy2(_PROJECT_DB_PATH, DB_PATH)
|
||
except Exception:
|
||
# 迁移失败不影响程序运行（会在新目录重新创建库）
|
||
pass
|
||
|
||
|
||
def get_conn():
|
||
_ensure_db_dir()
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
_migrate_old_db_if_needed()
|
||
conn = get_conn()
|
||
c = conn.cursor()
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS groups (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
position INTEGER DEFAULT 0,
|
||
folder_path TEXT DEFAULT ''
|
||
)
|
||
""")
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS items (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
group_id INTEGER NOT NULL,
|
||
name TEXT NOT NULL,
|
||
path TEXT NOT NULL,
|
||
icon_path TEXT,
|
||
position INTEGER DEFAULT 0,
|
||
FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
|
||
)
|
||
""")
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS settings (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
)
|
||
""")
|
||
# 迁移：旧库没有 folder_path 字段时自动添加
|
||
try:
|
||
c.execute("ALTER TABLE groups ADD COLUMN folder_path TEXT DEFAULT ''")
|
||
conn.commit()
|
||
except Exception:
|
||
pass # 字段已存在则忽略
|
||
|
||
# 默认分组
|
||
c.execute("SELECT COUNT(*) FROM groups")
|
||
if c.fetchone()[0] == 0:
|
||
c.execute("INSERT INTO groups (name, position) VALUES ('常用程序', 0)")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
_migrate_item_shortcut_paths_to_targets()
|
||
|
||
|
||
def _migrate_item_shortcut_paths_to_targets():
|
||
"""将旧数据中仍保存为 .lnk 的路径改存为快捷方式目标（若可解析）。仅执行一次。"""
|
||
conn = get_conn()
|
||
row = conn.execute(
|
||
"SELECT value FROM settings WHERE key='items_lnk_targets_migrated'"
|
||
).fetchone()
|
||
conn.close()
|
||
if row and row["value"] == "1":
|
||
return
|
||
|
||
from shortcut_target import path_for_storage
|
||
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT id, path FROM items").fetchall()
|
||
for r in rows:
|
||
iid, p = r["id"], (r["path"] or "")
|
||
pl = p.lower()
|
||
if not pl.endswith(".lnk"):
|
||
continue
|
||
try:
|
||
if not os.path.isfile(p):
|
||
continue
|
||
except OSError:
|
||
continue
|
||
new_p = path_for_storage(p)
|
||
if new_p != p:
|
||
conn.execute("UPDATE items SET path=? WHERE id=?", (new_p, iid))
|
||
conn.execute(
|
||
"INSERT INTO settings (key, value) VALUES (?,?) "
|
||
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
|
||
("items_lnk_targets_migrated", "1"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ── Settings ─────────────────────────────────────────────
|
||
def get_setting(key: str, default: str = "") -> str:
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
|
||
conn.close()
|
||
return row["value"] if row else default
|
||
|
||
|
||
def set_setting(key: str, value: str):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"INSERT INTO settings (key, value) VALUES (?,?) "
|
||
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
|
||
(key, value)
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ── Groups ──────────────────────────────────────────────
|
||
def get_groups():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM groups ORDER BY position").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def add_group(name, folder_path=""):
|
||
conn = get_conn()
|
||
conn.execute("INSERT INTO groups (name, folder_path) VALUES (?,?)", (name, folder_path))
|
||
conn.commit()
|
||
gid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||
conn.close()
|
||
return gid
|
||
|
||
|
||
def rename_group(gid, name):
|
||
conn = get_conn()
|
||
conn.execute("UPDATE groups SET name=? WHERE id=?", (name, gid))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_group(gid):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM groups WHERE id=?", (gid,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def reorder_groups(id_list):
|
||
conn = get_conn()
|
||
for pos, gid in enumerate(id_list):
|
||
conn.execute("UPDATE groups SET position=? WHERE id=?", (pos, gid))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ── Items ────────────────────────────────────────────────
|
||
def get_items(group_id):
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM items WHERE group_id=? ORDER BY position", (group_id,)
|
||
).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def add_item(group_id, name, path, icon_path=None):
|
||
from shortcut_target import path_for_storage
|
||
|
||
path = path_for_storage(path)
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"INSERT INTO items (group_id, name, path, icon_path) VALUES (?,?,?,?)",
|
||
(group_id, name, path, icon_path),
|
||
)
|
||
conn.commit()
|
||
iid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||
conn.close()
|
||
return iid
|
||
|
||
|
||
def delete_item(item_id):
|
||
conn = get_conn()
|
||
conn.execute("DELETE FROM items WHERE id=?", (item_id,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def move_item(item_id, new_group_id, new_position):
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"UPDATE items SET group_id=?, position=? WHERE id=?",
|
||
(new_group_id, new_position, item_id),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def reorder_items(group_id, id_list):
|
||
conn = get_conn()
|
||
for pos, iid in enumerate(id_list):
|
||
conn.execute("UPDATE items SET position=? WHERE id=?", (pos, iid))
|
||
conn.commit()
|
||
conn.close()
|