a8fe6ed7b3
- Replace TinyDB (JSON file) with sqlite3 for data persistence - Add schema.py: table creation + data migration from db.json - Rewrite database.py: all CRUD operations use sqlite3 directly - All data retains original IDs via migration script - Remove tinydb dependency from pyproject.toml
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
import sqlite3
|
|
import os
|
|
import json
|
|
import config
|
|
|
|
|
|
CREATE_USERS = """
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT NOT NULL UNIQUE,
|
|
password_hash TEXT NOT NULL,
|
|
role TEXT NOT NULL DEFAULT 'user',
|
|
max_goals INTEGER NOT NULL DEFAULT 5
|
|
)
|
|
"""
|
|
|
|
CREATE_GOALS = """
|
|
CREATE TABLE IF NOT EXISTS goals (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
title TEXT NOT NULL,
|
|
activated INTEGER NOT NULL DEFAULT 1,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
)
|
|
"""
|
|
|
|
CREATE_TASKS = """
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
goal_id INTEGER NOT NULL,
|
|
title TEXT NOT NULL,
|
|
desc TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'todo',
|
|
start_time TEXT,
|
|
finished_time TEXT,
|
|
"order" REAL NOT NULL DEFAULT 0.0,
|
|
FOREIGN KEY (goal_id) REFERENCES goals(id)
|
|
)
|
|
"""
|
|
|
|
|
|
def get_connection():
|
|
os.makedirs(os.path.dirname(config.DB_PATH), exist_ok=True)
|
|
conn = sqlite3.connect(config.DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
return conn
|
|
|
|
|
|
def init_db():
|
|
conn = get_connection()
|
|
try:
|
|
conn.execute(CREATE_USERS)
|
|
conn.execute(CREATE_GOALS)
|
|
conn.execute(CREATE_TASKS)
|
|
conn.commit()
|
|
|
|
import bcrypt
|
|
cur = conn.execute("SELECT COUNT(*) FROM users WHERE role = 'admin'")
|
|
if cur.fetchone()[0] == 0:
|
|
password_hash = bcrypt.hashpw(
|
|
config.DEFAULT_ADMIN_PASSWORD.encode("utf-8"),
|
|
bcrypt.gensalt()
|
|
).decode("utf-8")
|
|
conn.execute(
|
|
"INSERT INTO users (username, password_hash, role, max_goals) VALUES (?, ?, ?, ?)",
|
|
(config.DEFAULT_ADMIN_USERNAME, password_hash, "admin", 100)
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def migrate_from_tinydb():
|
|
json_path = os.path.join(os.path.dirname(config.DB_PATH), "db.json")
|
|
if not os.path.exists(json_path):
|
|
return
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with open(json_path, "r") as f:
|
|
data = json.load(f)
|
|
|
|
if "users" in data:
|
|
for doc_id, record in data["users"].items():
|
|
record["id"] = int(doc_id)
|
|
if conn.execute("SELECT COUNT(*) FROM users WHERE id = ?", (record["id"],)).fetchone()[0] == 0:
|
|
conn.execute(
|
|
"INSERT INTO users (id, username, password_hash, role, max_goals) VALUES (?, ?, ?, ?, ?)",
|
|
(record["id"], record["username"], record["password_hash"], record["role"], record["max_goals"])
|
|
)
|
|
|
|
if "goals" in data:
|
|
for doc_id, record in data["goals"].items():
|
|
record["id"] = int(doc_id)
|
|
if conn.execute("SELECT COUNT(*) FROM goals WHERE id = ?", (record["id"],)).fetchone()[0] == 0:
|
|
conn.execute(
|
|
"INSERT INTO goals (id, user_id, title, activated) VALUES (?, ?, ?, ?)",
|
|
(record["id"], record["user_id"], record["title"], 1 if record.get("activated", True) else 0)
|
|
)
|
|
|
|
if "tasks" in data:
|
|
for doc_id, record in data["tasks"].items():
|
|
record["id"] = int(doc_id)
|
|
if conn.execute("SELECT COUNT(*) FROM tasks WHERE id = ?", (record["id"],)).fetchone()[0] == 0:
|
|
conn.execute(
|
|
"""INSERT INTO tasks (id, goal_id, title, desc, status, start_time, finished_time, "order")
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(record["id"], record["goal_id"], record["title"], record.get("desc", ""),
|
|
record.get("status", "todo"), record.get("start_time"), record.get("finished_time"),
|
|
record.get("order", 0.0))
|
|
)
|
|
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|