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

    def create_database(self):
        """Creates the Gromulus SQLite database with all required tables."""
        if os.path.exists(self.db_path):
            os.remove(self.db_path)

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

        cursor.execute("""
            CREATE TABLE main_app_system (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                system_name TEXT,
                system_description TEXT
            )
        """)

        cursor.execute("""
            CREATE TABLE main_app (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                rom_filename TEXT,
                rom_hash TEXT,
                rom_filepath TEXT,
                system_id INTEGER,
                is_duplicate INTEGER,
                is_missing_metadata INTEGER,
                notes TEXT,
                FOREIGN KEY(system_id) REFERENCES main_app_system(id)
            )
        """)

        cursor.execute("""
            CREATE TABLE no_intro_system (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                system_name TEXT,
                system_description TEXT
            )
        """)

        cursor.execute("""
            CREATE TABLE no_intro_main (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                game_name TEXT,
                rom_hash TEXT,
                rom_filename TEXT,
                rom_size INTEGER,
                system_id INTEGER,
                FOREIGN KEY(system_id) REFERENCES no_intro_system(id)
            )
        """)

        cursor.execute("""
            CREATE TABLE tosec_system (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                system_name TEXT,
                system_description TEXT
            )
        """)

        cursor.execute("""
            CREATE TABLE tosec_main (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                game_name TEXT,
                rom_hash TEXT,
                rom_filename TEXT,
                rom_size INTEGER,
                system_id INTEGER,
                FOREIGN KEY(system_id) REFERENCES tosec_system(id)
            )
        """)

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

    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_system", "tosec_main"
        ]:
            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:	905
Committed:	Fri Jun 6 22:14:57 2025 UTC (9 months, 2 weeks ago) by nino.borges
Content type:	text/x-python
File size:	4766 byte(s)
Log Message:	Changed the file name CreateGromulusDB to GromulusDatabaseUtilities and changed it to be a class of utilities, instead of a simple make new db script. Now allows for, in additon to making the db, db backup and export and import to-from a JSON file. Also changed the schema to add a tosec system table, instead of using the no intro one.
#	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	795
16	nino.borges	905	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	nino.borges	795
21	nino.borges	905	conn = sqlite3.connect(self.db_path)
22			cursor = conn.cursor()
23	nino.borges	795
24	nino.borges	905	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	nino.borges	805
32	nino.borges	905	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	nino.borges	795
46	nino.borges	905	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	nino.borges	806
54	nino.borges	905	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	nino.borges	806
66	nino.borges	905	cursor.execute("""
67			CREATE TABLE tosec_system (
68			id INTEGER PRIMARY KEY AUTOINCREMENT,
69			system_name TEXT,
70			system_description TEXT
71			)
72			""")
73	nino.borges	806
74	nino.borges	905	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	nino.borges	806
86	nino.borges	905	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}")