lsd-cat · February 2, 2025 21:01 · fidojones · Nov 25, 2024 · arbv · Dec 1, 2024
diff --git a/Readme.md b/Readme.md
diff --git a/migrate.py b/migrate.py
 # Courtesy of ChatGPT :)

 import sqlite3
 import argparse

 def get_schema(db_path):
    """Extracts the schema of all user-defined tables from the SQLite database."""
    schema = {}
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        
        # Get all user-defined table names, excluding internal SQLite tables
        cursor.execute("""
            SELECT name FROM sqlite_master 
            WHERE type='table' AND name NOT LIKE 'sqlite_%'
        """)
        tables = [row[0] for row in cursor.fetchall()]

        for table in tables:
            cursor.execute(f"PRAGMA table_info(\"{table}\")")
            columns = cursor.fetchall()

            # Store as {table_name: {column_name: column_type}}
            schema[table] = {column[1]: column[2] for column in columns}

        conn.close()
    except sqlite3.Error as e:
        print(f"Error reading database {db_path}: {e}")

    return schema

 def clear_tables(db_path):
    """Deletes all records from all user-defined tables in the database."""
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # Get all user-defined table names, excluding internal SQLite tables
        cursor.execute("""
            SELECT name FROM sqlite_master 
            WHERE type='table' AND name NOT LIKE 'sqlite_%'
        """)
        tables = [row[0] for row in cursor.fetchall()]

        # Clear each table
        for table in tables:
            cursor.execute(f"DELETE FROM \"{table}\"")
            print(f"Cleared all records from table '{table}'.")

        conn.commit()
        conn.close()

        print("Database cleared of all records.")
    except sqlite3.Error as e:
        print(f"Error clearing database {db_path}: {e}")

 def migrate_data(db1_path, db2_path, schema1, schema2):
    """Migrates data from db1 to db2, omitting missing tables/columns."""
    try:
        conn1 = sqlite3.connect(db1_path)
        conn2 = sqlite3.connect(db2_path)
        cursor1 = conn1.cursor()
        cursor2 = conn2.cursor()

        for table in schema2.keys():
            if table not in schema1:
                print(f"Skipping table '{table}' as it is missing in the source database.")
                continue

            # Get columns common to both schemas
            common_columns = set(schema1[table].keys()).intersection(set(schema2[table].keys()))
            if not common_columns:
                print(f"Skipping table '{table}' as no columns match between databases.")
                continue

            # Generate column list for SELECT and INSERT
            column_list = ", ".join([f"\"{col}\"" for col in common_columns])
            query = f"SELECT {column_list} FROM \"{table}\""

            # Fetch data from source database
            cursor1.execute(query)
            rows = cursor1.fetchall()

            # Prepare insert statement for target database
            placeholders = ", ".join(["?"] * len(common_columns))
            insert_query = f"INSERT INTO \"{table}\" ({column_list}) VALUES ({placeholders})"

            # Insert data into the target database
            for row in rows:
                cursor2.execute(insert_query, row)

            print(f"Migrated table '{table}' with {len(rows)} rows.")
        
        conn2.commit()
        conn1.close()
        conn2.close()

        print("Migration complete.")

    except sqlite3.Error as e:
        print(f"Error during migration: {e}")

 def main():
    parser = argparse.ArgumentParser(description="Perform lossy migration between SQLite databases.")
    parser.add_argument("db1", help="Path to the source SQLite database.")
    parser.add_argument("db2", help="Path to the target SQLite database.")
    args = parser.parse_args()

    schema1 = get_schema(args.db1)
    schema2 = get_schema(args.db2)

    # Clear database 2
    clear_tables(args.db2)

    # Migrate data from database 1 to database 2
    migrate_data(args.db1, args.db2, schema1, schema2)

 if __name__ == "__main__":
    main()
	# Courtesy of ChatGPT :)

	import sqlite3
	import argparse

	def get_schema(db_path):
	"""Extracts the schema of all user-defined tables from the SQLite database."""
	schema = {}
	try:
	conn = sqlite3.connect(db_path)
	cursor = conn.cursor()

	# Get all user-defined table names, excluding internal SQLite tables
	cursor.execute("""
	SELECT name FROM sqlite_master
	WHERE type='table' AND name NOT LIKE 'sqlite_%'
	""")
	tables = [row[0] for row in cursor.fetchall()]

	for table in tables:
	cursor.execute(f"PRAGMA table_info(\"{table}\")")
	columns = cursor.fetchall()

	# Store as {table_name: {column_name: column_type}}
	schema[table] = {column[1]: column[2] for column in columns}

	conn.close()
	except sqlite3.Error as e:
	print(f"Error reading database {db_path}: {e}")

	return schema

	def clear_tables(db_path):
	"""Deletes all records from all user-defined tables in the database."""
	try:
	conn = sqlite3.connect(db_path)
	cursor = conn.cursor()

	# Get all user-defined table names, excluding internal SQLite tables
	cursor.execute("""
	SELECT name FROM sqlite_master
	WHERE type='table' AND name NOT LIKE 'sqlite_%'
	""")
	tables = [row[0] for row in cursor.fetchall()]

	# Clear each table
	for table in tables:
	cursor.execute(f"DELETE FROM \"{table}\"")
	print(f"Cleared all records from table '{table}'.")

	conn.commit()
	conn.close()

	print("Database cleared of all records.")
	except sqlite3.Error as e:
	print(f"Error clearing database {db_path}: {e}")

	def migrate_data(db1_path, db2_path, schema1, schema2):
	"""Migrates data from db1 to db2, omitting missing tables/columns."""
	try:
	conn1 = sqlite3.connect(db1_path)
	conn2 = sqlite3.connect(db2_path)
	cursor1 = conn1.cursor()
	cursor2 = conn2.cursor()

	for table in schema2.keys():
	if table not in schema1:
	print(f"Skipping table '{table}' as it is missing in the source database.")
	continue

	# Get columns common to both schemas
	common_columns = set(schema1[table].keys()).intersection(set(schema2[table].keys()))
	if not common_columns:
	print(f"Skipping table '{table}' as no columns match between databases.")
	continue

	# Generate column list for SELECT and INSERT
	column_list = ", ".join([f"\"{col}\"" for col in common_columns])
	query = f"SELECT {column_list} FROM \"{table}\""

	# Fetch data from source database
	cursor1.execute(query)
	rows = cursor1.fetchall()

	# Prepare insert statement for target database
	placeholders = ", ".join(["?"] * len(common_columns))
	insert_query = f"INSERT INTO \"{table}\" ({column_list}) VALUES ({placeholders})"

	# Insert data into the target database
	for row in rows:
	cursor2.execute(insert_query, row)

	print(f"Migrated table '{table}' with {len(rows)} rows.")

	conn2.commit()
	conn1.close()
	conn2.close()

	print("Migration complete.")

	except sqlite3.Error as e:
	print(f"Error during migration: {e}")

	def main():
	parser = argparse.ArgumentParser(description="Perform lossy migration between SQLite databases.")
	parser.add_argument("db1", help="Path to the source SQLite database.")
	parser.add_argument("db2", help="Path to the target SQLite database.")
	args = parser.parse_args()

	schema1 = get_schema(args.db1)
	schema2 = get_schema(args.db2)

	# Clear database 2
	clear_tables(args.db2)

	# Migrate data from database 1 to database 2
	migrate_data(args.db1, args.db2, schema1, schema2)

	if __name__ == "__main__":
	main()