149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
# core/database.py
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
DB_FILE = Path("app.db")
|
|
|
|
|
|
class DatabaseManager:
|
|
def __init__(self, db_path: Path = DB_FILE):
|
|
self.db_path = db_path
|
|
self.conn: sqlite3.Connection | None = None
|
|
|
|
# -------------------------
|
|
# Połączenie z bazą
|
|
# -------------------------
|
|
def connect(self):
|
|
"""Nawiązuje połączenie z bazą i tworzy tabele, jeśli ich nie ma."""
|
|
self.conn = sqlite3.connect(self.db_path)
|
|
self.conn.row_factory = sqlite3.Row
|
|
self._create_tables()
|
|
|
|
def disconnect(self):
|
|
if self.conn:
|
|
self.conn.close()
|
|
self.conn = None
|
|
|
|
# -------------------------
|
|
# Tworzenie tabel
|
|
# -------------------------
|
|
def _create_tables(self):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS colors (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT UNIQUE NOT NULL,
|
|
icon_path TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS media (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
color_id INTEGER NOT NULL,
|
|
media_path TEXT NOT NULL,
|
|
file_type TEXT CHECK(file_type IN ('photo','video')),
|
|
timestamp TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY(color_id) REFERENCES colors(id) ON DELETE CASCADE
|
|
)
|
|
"""
|
|
)
|
|
|
|
self.conn.commit()
|
|
|
|
# -------------------------
|
|
# Operacje na kolorach
|
|
# -------------------------
|
|
def add_color(self, name: str, icon_path: str):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute(
|
|
"INSERT OR IGNORE INTO colors (name, icon_path) VALUES (?, ?)",
|
|
(name, icon_path),
|
|
)
|
|
self.conn.commit()
|
|
|
|
def get_color_id(self, name: str) -> int | None:
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT id FROM colors WHERE name = ?", (name,))
|
|
row = cur.fetchone()
|
|
return row["id"] if row else None
|
|
|
|
def get_all_colors(self) -> list[dict]:
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT * FROM colors ORDER BY name")
|
|
rows = cur.fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def update_color_name(self, old_name: str, new_name: str):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("UPDATE colors SET name = ? WHERE name = ?", (new_name, old_name))
|
|
self.conn.commit()
|
|
|
|
def update_color_icon(self, name: str, icon_path: str):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("UPDATE colors SET icon_path = ? WHERE name = ?", (icon_path, name))
|
|
self.conn.commit()
|
|
|
|
def delete_color(self, name: str):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("DELETE FROM colors WHERE name = ?", (name,))
|
|
self.conn.commit()
|
|
|
|
# -------------------------
|
|
# Operacje na plikach
|
|
# -------------------------
|
|
def add_media(self, color_id: int, media_path: str, file_type: str, timestamp: str | None = None):
|
|
if timestamp is None:
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO media (color_id, media_path, file_type, timestamp) VALUES (?, ?, ?, ?)",
|
|
(color_id, media_path, file_type, timestamp),
|
|
)
|
|
self.conn.commit()
|
|
|
|
def get_media_for_color(self, color_id: int) -> list[dict]:
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT * FROM media WHERE color_id = ? ORDER BY timestamp DESC", (color_id,))
|
|
rows = cur.fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def delete_media(self, color_id: int, media_path: str):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("DELETE FROM media WHERE color_id = ? AND media_path = ?", (color_id, media_path))
|
|
self.conn.commit()
|
|
|
|
def delete_all_media_for_color(self, color_id: int):
|
|
if self.conn is None:
|
|
raise RuntimeError("Database not connected")
|
|
cur = self.conn.cursor()
|
|
cur.execute("DELETE FROM media WHERE color_id = ?", (color_id,))
|
|
self.conn.commit()
|