| 1 |
"""
|
| 2 |
This allows for some of the lower level db utilities, such as creating the inital Gromulus database, db backup, export, import for migration, etc. I'm assuming I'll use the create database once and then maybe each time I update the schema.
|
| 3 |
I'm not sure how I'll best do data migration manually.
|
| 4 |
"""
|
| 5 |
|
| 6 |
import sqlite3
|
| 7 |
import os
|
| 8 |
import shutil
|
| 9 |
import json
|
| 10 |
|
| 11 |
|
| 12 |
class GromulusDatabaseUtilities:
|
| 13 |
def __init__(self, db_path: str):
|
| 14 |
self.db_path = db_path
|
| 15 |
|
| 16 |
def create_database(self):
|
| 17 |
"""Creates the Gromulus SQLite database with all required tables."""
|
| 18 |
if os.path.exists(self.db_path):
|
| 19 |
os.remove(self.db_path)
|
| 20 |
|
| 21 |
conn = sqlite3.connect(self.db_path)
|
| 22 |
cursor = conn.cursor()
|
| 23 |
|
| 24 |
cursor.execute("""
|
| 25 |
CREATE TABLE main_app_system (
|
| 26 |
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| 27 |
system_name TEXT,
|
| 28 |
system_description TEXT
|
| 29 |
)
|
| 30 |
""")
|
| 31 |
|
| 32 |
cursor.execute("""
|
| 33 |
CREATE TABLE main_app (
|
| 34 |
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| 35 |
rom_filename TEXT,
|
| 36 |
rom_hash TEXT,
|
| 37 |
rom_filepath TEXT,
|
| 38 |
system_id INTEGER,
|
| 39 |
is_duplicate INTEGER,
|
| 40 |
is_missing_metadata INTEGER,
|
| 41 |
notes TEXT,
|
| 42 |
FOREIGN KEY(system_id) REFERENCES main_app_system(id)
|
| 43 |
)
|
| 44 |
""")
|
| 45 |
|
| 46 |
cursor.execute("""
|
| 47 |
CREATE TABLE no_intro_system (
|
| 48 |
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| 49 |
system_name TEXT,
|
| 50 |
system_description TEXT
|
| 51 |
)
|
| 52 |
""")
|
| 53 |
|
| 54 |
cursor.execute("""
|
| 55 |
CREATE TABLE no_intro_main (
|
| 56 |
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| 57 |
game_name TEXT,
|
| 58 |
rom_hash TEXT,
|
| 59 |
rom_filename TEXT,
|
| 60 |
rom_size INTEGER,
|
| 61 |
system_id INTEGER,
|
| 62 |
FOREIGN KEY(system_id) REFERENCES no_intro_system(id)
|
| 63 |
)
|
| 64 |
""")
|
| 65 |
|
| 66 |
cursor.execute("""
|
| 67 |
CREATE TABLE tosec_system (
|
| 68 |
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| 69 |
system_name TEXT,
|
| 70 |
system_description TEXT
|
| 71 |
)
|
| 72 |
""")
|
| 73 |
|
| 74 |
cursor.execute("""
|
| 75 |
CREATE TABLE tosec_main (
|
| 76 |
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| 77 |
game_name TEXT,
|
| 78 |
rom_hash TEXT,
|
| 79 |
rom_filename TEXT,
|
| 80 |
rom_size INTEGER,
|
| 81 |
system_id INTEGER,
|
| 82 |
FOREIGN KEY(system_id) REFERENCES tosec_system(id)
|
| 83 |
)
|
| 84 |
""")
|
| 85 |
|
| 86 |
conn.commit()
|
| 87 |
conn.close()
|
| 88 |
print("Database created successfully at:", self.db_path)
|
| 89 |
|
| 90 |
def backup_database(self, backup_path):
|
| 91 |
"""Copies the current database to a new location."""
|
| 92 |
if not os.path.exists(self.db_path):
|
| 93 |
raise FileNotFoundError("Database file not found.")
|
| 94 |
shutil.copy2(self.db_path, backup_path)
|
| 95 |
print(f"Database backed up to: {backup_path}")
|
| 96 |
|
| 97 |
def export_data_to_json(self, export_path):
|
| 98 |
"""Exports all table data to a JSON file."""
|
| 99 |
conn = sqlite3.connect(self.db_path)
|
| 100 |
cursor = conn.cursor()
|
| 101 |
|
| 102 |
data = {}
|
| 103 |
for table in [
|
| 104 |
"main_app_system", "main_app",
|
| 105 |
"no_intro_system", "no_intro_main",
|
| 106 |
"tosec_system", "tosec_main"
|
| 107 |
]:
|
| 108 |
cursor.execute(f"SELECT * FROM {table}")
|
| 109 |
columns = [description[0] for description in cursor.description]
|
| 110 |
rows = cursor.fetchall()
|
| 111 |
data[table] = [dict(zip(columns, row)) for row in rows]
|
| 112 |
|
| 113 |
with open(export_path, "w", encoding="utf-8") as f:
|
| 114 |
json.dump(data, f, indent=2)
|
| 115 |
|
| 116 |
conn.close()
|
| 117 |
print(f"Exported data to JSON: {export_path}")
|
| 118 |
|
| 119 |
def import_data_from_json(self, import_path):
|
| 120 |
"""Imports table data from a JSON file (assumes schema is already created)."""
|
| 121 |
if not os.path.exists(import_path):
|
| 122 |
raise FileNotFoundError("Import file not found.")
|
| 123 |
|
| 124 |
with open(import_path, "r", encoding="utf-8") as f:
|
| 125 |
data = json.load(f)
|
| 126 |
|
| 127 |
conn = sqlite3.connect(self.db_path)
|
| 128 |
cursor = conn.cursor()
|
| 129 |
|
| 130 |
for table, rows in data.items():
|
| 131 |
if not rows:
|
| 132 |
continue
|
| 133 |
columns = rows[0].keys()
|
| 134 |
placeholders = ", ".join(["?"] * len(columns))
|
| 135 |
col_list = ", ".join(columns)
|
| 136 |
|
| 137 |
for row in rows:
|
| 138 |
values = tuple(row[col] for col in columns)
|
| 139 |
cursor.execute(f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})", values)
|
| 140 |
|
| 141 |
conn.commit()
|
| 142 |
conn.close()
|
| 143 |
print(f"Imported data from JSON: {import_path}")
|
