GitLab

import sqlite3
import pymysql
from pymysql import Error, IntegrityError

def connect_to_sqlite(db_file):
    conn = None
    try:
        conn = sqlite3.connect(db_file)
    except Error as e:
        print(e)
    return conn

def connect_to_mysql(user, password, host, database):
    conn = None
    try:
        conn = pymysql.connect(host=host, user=user, password=password,
                               database=database, cursorclass=pymysql.cursors.DictCursor)
        print('Connected to MySQL database')
    except Error as e:
        print(e)
    return conn

def migrate_data(sqlite_db, mysql_conn, namespace_name):
    sqlite_conn = connect_to_sqlite(sqlite_db)
    if sqlite_conn is not None and mysql_conn is not None:
        sqlite_cursor = sqlite_conn.cursor()
        mysql_cursor = mysql_conn.cursor()

        # Получаем идентификатор пространства имен по его имени
        sqlite_cursor.execute("SELECT id FROM namespace WHERE name = ?", (namespace_name,))
        namespace_id = sqlite_cursor.fetchone()
        if namespace_id is None:
            print(f"Namespace '{namespace_name}' not found.")
            return

        # Выполняем выборку доменов из базы данных SQLite, используя идентификатор пространства имен
        sqlite_cursor.execute("SELECT * FROM domain WHERE namespace_id = ?", (namespace_id[0],))
        domains = sqlite_cursor.fetchall()

        # Мигрируем данные в базу данных MySQL
        for domain in domains:
            dtype_upper = domain[5].upper() if domain[5] else None
            try:
                mysql_cursor.execute("INSERT INTO domains (name, type) VALUES (%s, %s)", (domain[1], dtype_upper))
                mysql_conn.commit()
                domain_id = mysql_cursor.lastrowid
            except IntegrityError as e:
                print(f"Duplicate entry for {domain[1]}. Trying to update records instead.")
                mysql_cursor.execute("SELECT id FROM domains WHERE name = %s", (domain[1],))
                result = mysql_cursor.fetchone()
                if result is not None:
                    domain_id = result['id']
                else:
                    print(f"Could not find existing domain for name {domain[1]}. Skipping.")
                    continue

            # Получаем данные из таблицы record в SQLite для данного домена
            sqlite_cursor.execute("SELECT * FROM record WHERE domain_id=?", (domain[0],))
            records = sqlite_cursor.fetchall()
            for record in records:
                sqlite_cursor.execute("SELECT value FROM record_values WHERE record_id=? AND name='prio'", (record[0],))
                prio_value = sqlite_cursor.fetchone()
                if prio_value:
                    prio = int(prio_value[0])
                else:
                    prio = 0
                try:
                    mysql_cursor.execute("""
                                        INSERT INTO records (domain_id, name, type, content, ttl, prio) 
                                        VALUES (%s, %s, %s, %s, %s, %s)
                                        """, (domain_id, record[2], record[3].upper(), record[4], record[5], prio))
                    mysql_conn.commit()
                except IntegrityError as e:
                    print(f"Error updating record for domain {record[2]}: {e}")

        sqlite_cursor.close()
        mysql_cursor.close()
    else:
        print("Error on databases connection")

# Параметры подключения к MySQL
mysql_conn = connect_to_mysql('your_mysql_user', 'your_mysql_password', 'your_mysql_host', 'your_mysql_db')
# Путь к файлу SQLite базы данных
sqlite_db = 'path_to_your_sqlite3_db.db'

# Имя пространства имен
namespace_name = 'my_namespace_name'
# Выполнение миграции данных
migrate_data(sqlite_db, mysql_conn, namespace_name)

# Закрытие соединения с MySQL
if mysql_conn is not None:
    mysql_conn.close()