Active_prgs/Gromulus/GromulusDatabaseUtilities.py

"""
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.
I'm not sure how I'll best do data migration manually.
"""

import sqlite3
import os
import shutil
import json


class GromulusDatabaseUtilities:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._canonical_tables = {
            "main_app_system": """
                CREATE TABLE main_app_system(
                    id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
                    name TEXT,
                    short_name TEXT,
                    relative_file_path TEXT
                )
            """,
            "main_app": """
                CREATE TABLE main_app(
                    id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
                    game_name TEXT,
                    game_name_scraped TEXT,
                    container_file_name TEXT,
                    description TEXT,
                    description_scraped TEXT,
                    system_console INTEGER,
                    system_console_scraped TEXT,
                    path_to_screenshot_box TEXT,
                    path_to_screenshot_title TEXT,
                    path_to_screenshot_ingame TEXT,
                    path_to_video TEXT,
                    user_notes TEXT,
                    container_md5_hash TEXT,
                    version TEXT
                )
            """,
            "no_intro_system": """
                CREATE TABLE no_intro_system(
                    id INTEGER NOT NULL PRIMARY KEY,
                    name TEXT,
                    description TEXT,
                    dat_version TEXT
                )
            """,
            "no_intro_main": """
                CREATE TABLE no_intro_main(
                    game_name TEXT,
                    no_intro_id TEXT,
                    clone_of_id TEXT,
                    description TEXT,
                    rom_name TEXT,
                    crc TEXT,
                    md5 TEXT,
                    sha1 TEXT,
                    sha256 TEXT,
                    status TEXT,
                    no_intro_system_id INTEGER NOT NULL,
                    FOREIGN KEY(no_intro_system_id) REFERENCES no_intro_system(id)
                )
            """,
            "tosec_main": """
                CREATE TABLE tosec_main(
                    game_name TEXT,
                    description TEXT,
                    rom_name TEXT,
                    crc TEXT,
                    md5 TEXT,
                    sha1 TEXT,
                    system_id INTEGER NOT NULL,
                    FOREIGN KEY(system_id) REFERENCES no_intro_system(id)
                )
            """,
        }

    def create_database(self):
        """Creates a fresh Gromulus SQLite database using the canonical schema."""
        if os.path.exists(self.db_path):
            os.remove(self.db_path)

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        for ddl in self._canonical_tables.values():
            cursor.execute(ddl)
        self._create_indexes(cursor)

        conn.commit()
        conn.close()
        print("Database created successfully at:", self.db_path)

    def normalize_schema(self):
        """
        Normalizes an existing DB to the canonical schema without dropping data.
        Adds missing tables/columns and backfills known old column names.
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        for table_name, ddl in self._canonical_tables.items():
            cursor.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
                (table_name,),
            )
            if cursor.fetchone() is None:
                cursor.execute(ddl)

        self._ensure_columns(
            cursor,
            "main_app",
            [
                "game_name", "game_name_scraped", "container_file_name", "description",
                "description_scraped", "system_console", "system_console_scraped",
                "path_to_screenshot_box", "path_to_screenshot_title",
                "path_to_screenshot_ingame", "path_to_video", "user_notes",
                "container_md5_hash", "version",
            ],
        )
        self._ensure_columns(cursor, "main_app_system", ["name", "short_name", "relative_file_path"])
        self._ensure_columns(cursor, "no_intro_system", ["name", "description", "dat_version"])
        self._ensure_columns(
            cursor,
            "no_intro_main",
            [
                "game_name", "no_intro_id", "clone_of_id", "description", "rom_name",
                "crc", "md5", "sha1", "sha256", "status", "no_intro_system_id",
            ],
        )
        self._ensure_columns(
            cursor,
            "tosec_main",
            ["game_name", "description", "rom_name", "crc", "md5", "sha1", "system_id"],
        )

        self._backfill_aliases(cursor)
        self._create_indexes(cursor)
        conn.commit()
        conn.close()
        print("Schema normalized successfully at:", self.db_path)

    def _ensure_columns(self, cursor, table_name, required_columns):
        cursor.execute(f"PRAGMA table_info({table_name})")
        existing_columns = {row[1] for row in cursor.fetchall()}
        for column_name in required_columns:
            if column_name not in existing_columns:
                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} TEXT")

    def _column_exists(self, cursor, table_name, column_name):
        cursor.execute(f"PRAGMA table_info({table_name})")
        return any(row[1] == column_name for row in cursor.fetchall())

    def _backfill_aliases(self, cursor):
        # main_app alias mapping
        if self._column_exists(cursor, "main_app", "file_name"):
            cursor.execute(
                "UPDATE main_app SET container_file_name=file_name "
                "WHERE (container_file_name IS NULL OR container_file_name='') "
                "AND file_name IS NOT NULL"
            )
        if self._column_exists(cursor, "main_app", "md5"):
            cursor.execute(
                "UPDATE main_app SET container_md5_hash=md5 "
                "WHERE (container_md5_hash IS NULL OR container_md5_hash='') AND md5 IS NOT NULL"
            )
        if self._column_exists(cursor, "main_app", "system_id"):
            cursor.execute(
                "UPDATE main_app SET system_console=system_id "
                "WHERE (system_console IS NULL OR system_console='') AND system_id IS NOT NULL"
            )
        if self._column_exists(cursor, "main_app", "rom_filename"):
            cursor.execute(
                "UPDATE main_app SET container_file_name=rom_filename "
                "WHERE (container_file_name IS NULL OR container_file_name='') "
                "AND rom_filename IS NOT NULL"
            )
        if self._column_exists(cursor, "main_app", "rom_hash"):
            cursor.execute(
                "UPDATE main_app SET container_md5_hash=rom_hash "
                "WHERE (container_md5_hash IS NULL OR container_md5_hash='') "
                "AND rom_hash IS NOT NULL"
            )
        if self._column_exists(cursor, "main_app", "notes"):
            cursor.execute(
                "UPDATE main_app SET user_notes=notes "
                "WHERE (user_notes IS NULL OR user_notes='') AND notes IS NOT NULL"
            )

        # system table alias mapping
        if self._column_exists(cursor, "no_intro_system", "system_name"):
            cursor.execute(
                "UPDATE no_intro_system SET name=system_name "
                "WHERE (name IS NULL OR name='') AND system_name IS NOT NULL"
            )
        if self._column_exists(cursor, "no_intro_system", "system_description"):
            cursor.execute(
                "UPDATE no_intro_system SET description=system_description "
                "WHERE (description IS NULL OR description='') AND system_description IS NOT NULL"
            )

        # no_intro_main alias mapping
        if self._column_exists(cursor, "no_intro_main", "rom_hash"):
            cursor.execute(
                "UPDATE no_intro_main SET md5=rom_hash "
                "WHERE (md5 IS NULL OR md5='') AND rom_hash IS NOT NULL"
            )
        if self._column_exists(cursor, "no_intro_main", "rom_filename"):
            cursor.execute(
                "UPDATE no_intro_main SET rom_name=rom_filename "
                "WHERE (rom_name IS NULL OR rom_name='') AND rom_filename IS NOT NULL"
            )
        if self._column_exists(cursor, "no_intro_main", "system_id"):
            cursor.execute(
                "UPDATE no_intro_main SET no_intro_system_id=system_id "
                "WHERE (no_intro_system_id IS NULL OR no_intro_system_id='') AND system_id IS NOT NULL"
            )

        # tosec_main alias mapping
        if self._column_exists(cursor, "tosec_main", "rom_hash"):
            cursor.execute(
                "UPDATE tosec_main SET md5=rom_hash "
                "WHERE (md5 IS NULL OR md5='') AND rom_hash IS NOT NULL"
            )
        if self._column_exists(cursor, "tosec_main", "rom_filename"):
            cursor.execute(
                "UPDATE tosec_main SET rom_name=rom_filename "
                "WHERE (rom_name IS NULL OR rom_name='') AND rom_filename IS NOT NULL"
            )

    def _create_indexes(self, cursor):
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_main_app_md5 ON main_app(container_md5_hash)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_main_app_system ON main_app(system_console)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_no_intro_md5 ON no_intro_main(md5)")
        cursor.execute("CREATE INDEX IF NOT EXISTS idx_tosec_md5 ON tosec_main(md5)")

    def backup_database(self, backup_path):
        """Copies the current database to a new location."""
        if not os.path.exists(self.db_path):
            raise FileNotFoundError("Database file not found.")
        shutil.copy2(self.db_path, backup_path)
        print(f"Database backed up to: {backup_path}")

    def export_data_to_json(self, export_path):
        """Exports all table data to a JSON file."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        data = {}
        for table in [
            "main_app_system", "main_app",
            "no_intro_system", "no_intro_main",
            "tosec_main"
        ]:
            cursor.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
                (table,),
            )
            if cursor.fetchone() is None:
                continue
            cursor.execute(f"SELECT * FROM {table}")
            columns = [description[0] for description in cursor.description]
            rows = cursor.fetchall()
            data[table] = [dict(zip(columns, row)) for row in rows]

        with open(export_path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2)

        conn.close()
        print(f"Exported data to JSON: {export_path}")

    def import_data_from_json(self, import_path):
        """Imports table data from a JSON file (assumes schema is already created)."""
        if not os.path.exists(import_path):
            raise FileNotFoundError("Import file not found.")

        with open(import_path, "r", encoding="utf-8") as f:
            data = json.load(f)

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        for table, rows in data.items():
            if not rows:
                continue
            columns = rows[0].keys()
            placeholders = ", ".join(["?"] * len(columns))
            col_list = ", ".join(columns)

            for row in rows:
                values = tuple(row[col] for col in columns)
                cursor.execute(f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})", values)

        conn.commit()
        conn.close()
        print(f"Imported data from JSON: {import_path}")
Revision:	976
Committed:	Fri Feb 13 02:00:16 2026 UTC (6 weeks, 1 day ago) by nino.borges
Content type:	text/x-python
File size:	12222 byte(s)
Log Message:	Refactor catalog data flow: unify schema usage, fix UI data contracts, and match metadata via main_app_file_hash
#	Content
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	self._canonical_tables = {
16	"main_app_system": """
17	CREATE TABLE main_app_system(
18	id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
19	name TEXT,
20	short_name TEXT,
21	relative_file_path TEXT
22	)
23	""",
24	"main_app": """
25	CREATE TABLE main_app(
26	id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
27	game_name TEXT,
28	game_name_scraped TEXT,
29	container_file_name TEXT,
30	description TEXT,
31	description_scraped TEXT,
32	system_console INTEGER,
33	system_console_scraped TEXT,
34	path_to_screenshot_box TEXT,
35	path_to_screenshot_title TEXT,
36	path_to_screenshot_ingame TEXT,
37	path_to_video TEXT,
38	user_notes TEXT,
39	container_md5_hash TEXT,
40	version TEXT
41	)
42	""",
43	"no_intro_system": """
44	CREATE TABLE no_intro_system(
45	id INTEGER NOT NULL PRIMARY KEY,
46	name TEXT,
47	description TEXT,
48	dat_version TEXT
49	)
50	""",
51	"no_intro_main": """
52	CREATE TABLE no_intro_main(
53	game_name TEXT,
54	no_intro_id TEXT,
55	clone_of_id TEXT,
56	description TEXT,
57	rom_name TEXT,
58	crc TEXT,
59	md5 TEXT,
60	sha1 TEXT,
61	sha256 TEXT,
62	status TEXT,
63	no_intro_system_id INTEGER NOT NULL,
64	FOREIGN KEY(no_intro_system_id) REFERENCES no_intro_system(id)
65	)
66	""",
67	"tosec_main": """
68	CREATE TABLE tosec_main(
69	game_name TEXT,
70	description TEXT,
71	rom_name TEXT,
72	crc TEXT,
73	md5 TEXT,
74	sha1 TEXT,
75	system_id INTEGER NOT NULL,
76	FOREIGN KEY(system_id) REFERENCES no_intro_system(id)
77	)
78	""",
79	}
80
81	def create_database(self):
82	"""Creates a fresh Gromulus SQLite database using the canonical schema."""
83	if os.path.exists(self.db_path):
84	os.remove(self.db_path)
85
86	conn = sqlite3.connect(self.db_path)
87	cursor = conn.cursor()
88
89	for ddl in self._canonical_tables.values():
90	cursor.execute(ddl)
91	self._create_indexes(cursor)
92
93	conn.commit()
94	conn.close()
95	print("Database created successfully at:", self.db_path)
96
97	def normalize_schema(self):
98	"""
99	Normalizes an existing DB to the canonical schema without dropping data.
100	Adds missing tables/columns and backfills known old column names.
101	"""
102	conn = sqlite3.connect(self.db_path)
103	cursor = conn.cursor()
104
105	for table_name, ddl in self._canonical_tables.items():
106	cursor.execute(
107	"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
108	(table_name,),
109	)
110	if cursor.fetchone() is None:
111	cursor.execute(ddl)
112
113	self._ensure_columns(
114	cursor,
115	"main_app",
116	[
117	"game_name", "game_name_scraped", "container_file_name", "description",
118	"description_scraped", "system_console", "system_console_scraped",
119	"path_to_screenshot_box", "path_to_screenshot_title",
120	"path_to_screenshot_ingame", "path_to_video", "user_notes",
121	"container_md5_hash", "version",
122	],
123	)
124	self._ensure_columns(cursor, "main_app_system", ["name", "short_name", "relative_file_path"])
125	self._ensure_columns(cursor, "no_intro_system", ["name", "description", "dat_version"])
126	self._ensure_columns(
127	cursor,
128	"no_intro_main",
129	[
130	"game_name", "no_intro_id", "clone_of_id", "description", "rom_name",
131	"crc", "md5", "sha1", "sha256", "status", "no_intro_system_id",
132	],
133	)
134	self._ensure_columns(
135	cursor,
136	"tosec_main",
137	["game_name", "description", "rom_name", "crc", "md5", "sha1", "system_id"],
138	)
139
140	self._backfill_aliases(cursor)
141	self._create_indexes(cursor)
142	conn.commit()
143	conn.close()
144	print("Schema normalized successfully at:", self.db_path)
145
146	def _ensure_columns(self, cursor, table_name, required_columns):
147	cursor.execute(f"PRAGMA table_info({table_name})")
148	existing_columns = {row[1] for row in cursor.fetchall()}
149	for column_name in required_columns:
150	if column_name not in existing_columns:
151	cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} TEXT")
152
153	def _column_exists(self, cursor, table_name, column_name):
154	cursor.execute(f"PRAGMA table_info({table_name})")
155	return any(row[1] == column_name for row in cursor.fetchall())
156
157	def _backfill_aliases(self, cursor):
158	# main_app alias mapping
159	if self._column_exists(cursor, "main_app", "file_name"):
160	cursor.execute(
161	"UPDATE main_app SET container_file_name=file_name "
162	"WHERE (container_file_name IS NULL OR container_file_name='') "
163	"AND file_name IS NOT NULL"
164	)
165	if self._column_exists(cursor, "main_app", "md5"):
166	cursor.execute(
167	"UPDATE main_app SET container_md5_hash=md5 "
168	"WHERE (container_md5_hash IS NULL OR container_md5_hash='') AND md5 IS NOT NULL"
169	)
170	if self._column_exists(cursor, "main_app", "system_id"):
171	cursor.execute(
172	"UPDATE main_app SET system_console=system_id "
173	"WHERE (system_console IS NULL OR system_console='') AND system_id IS NOT NULL"
174	)
175	if self._column_exists(cursor, "main_app", "rom_filename"):
176	cursor.execute(
177	"UPDATE main_app SET container_file_name=rom_filename "
178	"WHERE (container_file_name IS NULL OR container_file_name='') "
179	"AND rom_filename IS NOT NULL"
180	)
181	if self._column_exists(cursor, "main_app", "rom_hash"):
182	cursor.execute(
183	"UPDATE main_app SET container_md5_hash=rom_hash "
184	"WHERE (container_md5_hash IS NULL OR container_md5_hash='') "
185	"AND rom_hash IS NOT NULL"
186	)
187	if self._column_exists(cursor, "main_app", "notes"):
188	cursor.execute(
189	"UPDATE main_app SET user_notes=notes "
190	"WHERE (user_notes IS NULL OR user_notes='') AND notes IS NOT NULL"
191	)
192
193	# system table alias mapping
194	if self._column_exists(cursor, "no_intro_system", "system_name"):
195	cursor.execute(
196	"UPDATE no_intro_system SET name=system_name "
197	"WHERE (name IS NULL OR name='') AND system_name IS NOT NULL"
198	)
199	if self._column_exists(cursor, "no_intro_system", "system_description"):
200	cursor.execute(
201	"UPDATE no_intro_system SET description=system_description "
202	"WHERE (description IS NULL OR description='') AND system_description IS NOT NULL"
203	)
204
205	# no_intro_main alias mapping
206	if self._column_exists(cursor, "no_intro_main", "rom_hash"):
207	cursor.execute(
208	"UPDATE no_intro_main SET md5=rom_hash "
209	"WHERE (md5 IS NULL OR md5='') AND rom_hash IS NOT NULL"
210	)
211	if self._column_exists(cursor, "no_intro_main", "rom_filename"):
212	cursor.execute(
213	"UPDATE no_intro_main SET rom_name=rom_filename "
214	"WHERE (rom_name IS NULL OR rom_name='') AND rom_filename IS NOT NULL"
215	)
216	if self._column_exists(cursor, "no_intro_main", "system_id"):
217	cursor.execute(
218	"UPDATE no_intro_main SET no_intro_system_id=system_id "
219	"WHERE (no_intro_system_id IS NULL OR no_intro_system_id='') AND system_id IS NOT NULL"
220	)
221
222	# tosec_main alias mapping
223	if self._column_exists(cursor, "tosec_main", "rom_hash"):
224	cursor.execute(
225	"UPDATE tosec_main SET md5=rom_hash "
226	"WHERE (md5 IS NULL OR md5='') AND rom_hash IS NOT NULL"
227	)
228	if self._column_exists(cursor, "tosec_main", "rom_filename"):
229	cursor.execute(
230	"UPDATE tosec_main SET rom_name=rom_filename "
231	"WHERE (rom_name IS NULL OR rom_name='') AND rom_filename IS NOT NULL"
232	)
233
234	def _create_indexes(self, cursor):
235	cursor.execute("CREATE INDEX IF NOT EXISTS idx_main_app_md5 ON main_app(container_md5_hash)")
236	cursor.execute("CREATE INDEX IF NOT EXISTS idx_main_app_system ON main_app(system_console)")
237	cursor.execute("CREATE INDEX IF NOT EXISTS idx_no_intro_md5 ON no_intro_main(md5)")
238	cursor.execute("CREATE INDEX IF NOT EXISTS idx_tosec_md5 ON tosec_main(md5)")
239
240	def backup_database(self, backup_path):
241	"""Copies the current database to a new location."""
242	if not os.path.exists(self.db_path):
243	raise FileNotFoundError("Database file not found.")
244	shutil.copy2(self.db_path, backup_path)
245	print(f"Database backed up to: {backup_path}")
246
247	def export_data_to_json(self, export_path):
248	"""Exports all table data to a JSON file."""
249	conn = sqlite3.connect(self.db_path)
250	cursor = conn.cursor()
251
252	data = {}
253	for table in [
254	"main_app_system", "main_app",
255	"no_intro_system", "no_intro_main",
256	"tosec_main"
257	]:
258	cursor.execute(
259	"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
260	(table,),
261	)
262	if cursor.fetchone() is None:
263	continue
264	cursor.execute(f"SELECT * FROM {table}")
265	columns = [description[0] for description in cursor.description]
266	rows = cursor.fetchall()
267	data[table] = [dict(zip(columns, row)) for row in rows]
268
269	with open(export_path, "w", encoding="utf-8") as f:
270	json.dump(data, f, indent=2)
271
272	conn.close()
273	print(f"Exported data to JSON: {export_path}")
274
275	def import_data_from_json(self, import_path):
276	"""Imports table data from a JSON file (assumes schema is already created)."""
277	if not os.path.exists(import_path):
278	raise FileNotFoundError("Import file not found.")
279
280	with open(import_path, "r", encoding="utf-8") as f:
281	data = json.load(f)
282
283	conn = sqlite3.connect(self.db_path)
284	cursor = conn.cursor()
285
286	for table, rows in data.items():
287	if not rows:
288	continue
289	columns = rows[0].keys()
290	placeholders = ", ".join(["?"] * len(columns))
291	col_list = ", ".join(columns)
292
293	for row in rows:
294	values = tuple(row[col] for col in columns)
295	cursor.execute(f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})", values)
296
297	conn.commit()
298	conn.close()
299	print(f"Imported data from JSON: {import_path}")