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
#	User	Rev	Content
1	nino.borges	795	"""
2	nino.borges	905	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	nino.borges	795	"""
5
6			import sqlite3
7	nino.borges	905	import os
8			import shutil
9			import json
10	nino.borges	795
11
12	nino.borges	905	class GromulusDatabaseUtilities:
13			def __init__(self, db_path: str):
14			self.db_path = db_path
15	nino.borges	976	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	nino.borges	795
81	nino.borges	905	def create_database(self):
82	nino.borges	976	"""Creates a fresh Gromulus SQLite database using the canonical schema."""
83	nino.borges	905	if os.path.exists(self.db_path):
84			os.remove(self.db_path)
85	nino.borges	795
86	nino.borges	905	conn = sqlite3.connect(self.db_path)
87			cursor = conn.cursor()
88	nino.borges	795
89	nino.borges	976	for ddl in self._canonical_tables.values():
90			cursor.execute(ddl)
91			self._create_indexes(cursor)
92	nino.borges	805
93	nino.borges	976	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	nino.borges	905	)
110	nino.borges	976	if cursor.fetchone() is None:
111			cursor.execute(ddl)
112	nino.borges	795
113	nino.borges	976	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	nino.borges	905	)
165	nino.borges	976	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	nino.borges	806
193	nino.borges	976	# 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	nino.borges	905	)
199	nino.borges	976	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	nino.borges	806
205	nino.borges	976	# 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	nino.borges	905	)
211	nino.borges	976	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	nino.borges	806
222	nino.borges	976	# 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	nino.borges	905	)
228	nino.borges	976	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	nino.borges	806
234	nino.borges	976	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	nino.borges	905
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	nino.borges	976	"tosec_main"
257	nino.borges	905	]:
258	nino.borges	976	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	nino.borges	905	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}")