Update db_manager.py
This commit is contained in:
GitHub
251
db_manager.py
251
db_manager.py
@@ -1,111 +1,170 @@
|
||||
import sqlite3
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
DB_FILE = "app_data.db"
|
||||
JSON_BACKUP = "settings.json" # The old file you want to import from
|
||||
class DatabaseManager:
|
||||
def __init__(self, db_path="app_data.db"):
|
||||
self.db_path = db_path
|
||||
self.conn = None
|
||||
self.connect()
|
||||
self.init_tables()
|
||||
|
||||
def get_db_connection():
|
||||
conn = sqlite3.connect(DB_FILE)
|
||||
conn.row_factory = sqlite3.Row # Allows accessing columns by name
|
||||
return conn
|
||||
def connect(self):
|
||||
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
|
||||
def init_db():
|
||||
"""Creates tables and imports JSON if DB is new."""
|
||||
conn = get_db_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 1. Create Settings Table (Keyed by Path)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS settings (
|
||||
path TEXT PRIMARY KEY,
|
||||
params TEXT,
|
||||
updated_at TIMESTAMP
|
||||
)
|
||||
''')
|
||||
|
||||
# 2. Create Batch History Table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS batch_log (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
timestamp TEXT,
|
||||
path TEXT,
|
||||
status TEXT,
|
||||
details TEXT
|
||||
)
|
||||
''')
|
||||
|
||||
# 3. Check for Migration (If DB is empty but JSON exists)
|
||||
cursor.execute('SELECT count(*) FROM settings')
|
||||
count = cursor.fetchone()[0]
|
||||
|
||||
if count == 0 and os.path.exists(JSON_BACKUP):
|
||||
print(f"Migration: Database empty. Importing from {JSON_BACKUP}...")
|
||||
try:
|
||||
with open(JSON_BACKUP, 'r') as f:
|
||||
data = json.load(f)
|
||||
# Assumes JSON structure: {"/path/to/img1": {config}, "/path/to/img2": {config}}
|
||||
for path, config in data.items():
|
||||
cursor.execute(
|
||||
'INSERT OR REPLACE INTO settings (path, params, updated_at) VALUES (?, ?, ?)',
|
||||
(path, json.dumps(config), datetime.now())
|
||||
)
|
||||
conn.commit()
|
||||
print("Migration successful.")
|
||||
except Exception as e:
|
||||
print(f"Migration failed: {e}")
|
||||
def init_tables(self):
|
||||
"""Create tables for App Config, Project Settings, and Snippets."""
|
||||
cursor = self.conn.cursor()
|
||||
|
||||
# 1. Project Settings (Replaces your per-file JSONs)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS projects (
|
||||
path TEXT PRIMARY KEY,
|
||||
parent_dir TEXT,
|
||||
filename TEXT,
|
||||
data TEXT,
|
||||
is_batch INTEGER,
|
||||
updated_at TIMESTAMP
|
||||
)
|
||||
''')
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
# 2. Global App Config (Replaces config.json)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS app_config (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT
|
||||
)
|
||||
''')
|
||||
|
||||
# 3. Snippets (Replaces snippets.json)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS snippets (
|
||||
name TEXT PRIMARY KEY,
|
||||
content TEXT
|
||||
)
|
||||
''')
|
||||
self.conn.commit()
|
||||
|
||||
# --- SETTINGS FUNCTIONS ---
|
||||
# --- MIGRATION LOGIC ---
|
||||
def is_empty(self):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM projects")
|
||||
return cursor.fetchone()[0] == 0
|
||||
|
||||
def load_settings_for_path(path):
|
||||
"""Returns a dict of settings for the specific path, or None."""
|
||||
conn = get_db_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT params FROM settings WHERE path = ?", (path,))
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
if row:
|
||||
return json.loads(row['params'])
|
||||
return None
|
||||
def migrate_from_json(self, root_dir):
|
||||
"""Scans folder for .json files and imports them if DB is empty."""
|
||||
print("Starting Migration from JSON...")
|
||||
root = Path(root_dir)
|
||||
count = 0
|
||||
|
||||
# 1. Migrate Projects
|
||||
for json_file in root.glob("*.json"):
|
||||
if json_file.name in [".editor_config.json", ".editor_snippets.json"]:
|
||||
continue
|
||||
|
||||
try:
|
||||
with open(json_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
is_batch = 1 if "batch_data" in data or isinstance(data, list) else 0
|
||||
self.save_project(json_file, data, is_batch)
|
||||
count += 1
|
||||
except Exception as e:
|
||||
print(f"Failed to migrate {json_file}: {e}")
|
||||
|
||||
def save_settings_for_path(path, settings_dict):
|
||||
"""Saves the settings dict into the DB associated with the path."""
|
||||
conn = get_db_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
json_str = json.dumps(settings_dict)
|
||||
timestamp = datetime.now()
|
||||
|
||||
cursor.execute('''
|
||||
INSERT INTO settings (path, params, updated_at)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(path) DO UPDATE SET
|
||||
params=excluded.params,
|
||||
# 2. Migrate Config
|
||||
config_path = root / ".editor_config.json"
|
||||
if config_path.exists():
|
||||
with open(config_path, 'r') as f:
|
||||
self.save_app_config(json.load(f))
|
||||
|
||||
# 3. Migrate Snippets
|
||||
snip_path = root / ".editor_snippets.json"
|
||||
if snip_path.exists():
|
||||
with open(snip_path, 'r') as f:
|
||||
snippets = json.load(f)
|
||||
for k, v in snippets.items():
|
||||
self.save_snippet(k, v)
|
||||
|
||||
print(f"Migration complete. Imported {count} files.")
|
||||
|
||||
# --- PROJECT OPERATIONS ---
|
||||
def get_projects_in_dir(self, directory):
|
||||
"""Returns list of filenames for a specific directory."""
|
||||
cursor = self.conn.cursor()
|
||||
# Ensure directory path format matches how we save it
|
||||
dir_str = str(Path(directory).absolute())
|
||||
|
||||
cursor.execute("SELECT filename FROM projects WHERE parent_dir = ?", (dir_str,))
|
||||
return [row['filename'] for row in cursor.fetchall()]
|
||||
|
||||
def load_project(self, full_path):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT data FROM projects WHERE path = ?", (str(full_path),))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return json.loads(row['data'])
|
||||
return None
|
||||
|
||||
def save_project(self, full_path, data, is_batch=False):
|
||||
path_obj = Path(full_path)
|
||||
path_str = str(path_obj.absolute())
|
||||
parent_str = str(path_obj.parent.absolute())
|
||||
filename = path_obj.name
|
||||
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute('''
|
||||
INSERT INTO projects (path, parent_dir, filename, data, is_batch, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(path) DO UPDATE SET
|
||||
data=excluded.data,
|
||||
is_batch=excluded.is_batch,
|
||||
updated_at=excluded.updated_at
|
||||
''', (path, json_str, timestamp))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return f"Saved settings for: {path}"
|
||||
''', (path_str, parent_str, filename, json.dumps(data), int(is_batch), datetime.now()))
|
||||
self.conn.commit()
|
||||
|
||||
# --- BATCH FUNCTIONS ---
|
||||
def delete_project(self, full_path):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT count(*) FROM projects WHERE path = ?", (str(full_path),)) # Check existence
|
||||
cursor.execute("DELETE FROM projects WHERE path = ?", (str(full_path),))
|
||||
self.conn.commit()
|
||||
|
||||
def log_batch_run(path, status, details_dict):
|
||||
conn = get_db_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
INSERT INTO batch_log (timestamp, path, status, details)
|
||||
VALUES (?, ?, ?, ?)
|
||||
''', (datetime.now(), path, status, json.dumps(details_dict)))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
# --- CONFIG OPERATIONS ---
|
||||
def load_app_config(self):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT value FROM app_config WHERE key = 'main_config'")
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return json.loads(row['value'])
|
||||
# Default Config
|
||||
return {"last_dir": str(Path.cwd()), "favorites": []}
|
||||
|
||||
# Initialize DB immediately when this module is imported
|
||||
init_db()
|
||||
def save_app_config(self, config_dict):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute('''
|
||||
INSERT INTO app_config (key, value) VALUES ('main_config', ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value=excluded.value
|
||||
''', (json.dumps(config_dict),))
|
||||
self.conn.commit()
|
||||
|
||||
# --- SNIPPET OPERATIONS ---
|
||||
def load_snippets(self):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT * FROM snippets")
|
||||
return {row['name']: row['content'] for row in cursor.fetchall()}
|
||||
|
||||
def save_snippet(self, name, content):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute('''
|
||||
INSERT INTO snippets (name, content) VALUES (?, ?)
|
||||
ON CONFLICT(name) DO UPDATE SET content=excluded.content
|
||||
''', (name, content))
|
||||
self.conn.commit()
|
||||
|
||||
def delete_snippet(self, name):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("DELETE FROM snippets WHERE name = ?", (name,))
|
||||
self.conn.commit()
|
||||
|
||||
Reference in New Issue
Block a user