main.py - master - warriorpacis/Connector

main.py
189 строк8 KB
fedotrick
Успешно внедрили функциональность определения имени компьютера и выбора соответствующих путей к SQLite базам данных
03 окт 2025, 08:50
03 окт 2025, 08:5012d5d3d
import sqlite3
import mysql.connector
import pandas as pd
import glob
import os
import re
import numpy as np
import platform

# 🔹 Настройки MySQL (главная база для подключения)
mysql_config = {
    "host": "localhost",       # или IP сервера
    "user": "root",
    "password": "19912018",
    "port": 3306,
    "charset": "utf8mb4",      # 🔥 Добавлено для правильной кодировки
    "use_unicode": True        # 🔥 Добавлено для поддержки Unicode
}

# 🔹 Определяем имя компьютера
hostname = platform.node()
print(f"Имя компьютера: {hostname}")

# 🔹 Папки с SQLite-базами в зависимости от имени компьютера
if hostname == "SADLitCehDesktop":
    sqlite_folders = [
        r"C:\Users\1\Telegram\FoamFusionLab\data",
        r"C:\Users\1\Telegram\MetalFusionX",
        r"C:\Users\1\Telegram\TrimCraftBot",
        r"C:\Users\1\Telegram\HeatControlPRO",
        r"C:\Users\1\Telegram\SurfaceStrike",
        r"C:\Users\1\Telegram\AnvilAssistant",
    ]
    print("Используются пути для SADLitCehDesktop")
elif hostname == "SADGlEnergy2Desk":
    sqlite_folders = [
        r"Y:\Telegram\FoamFusionLab\data",
        r"Y:\Telegram\MetalFusionX",
        r"Y:\Telegram\TrimCraftBot",
        r"Y:\Telegram\HeatControlPRO",
        r"Y:\Telegram\SurfaceStrike",
        r"Y:\Telegram\AnvilAssistant",
    ]
    print("Используются пути для SADGlEnergy2Desk")
else:
    # 🔹 Папки с SQLite-базами (по умолчанию)
    sqlite_folders = [
        r"Y:\Telegram\FoamFusionLab\data",
        r"Y:\Telegram\MetalFusionX",
        r"Y:\Telegram\TrimCraftBot",
        r"Y:\Telegram\HeatControlPRO",
        r"Y:\Telegram\SurfaceStrike",
        r"Y:\Telegram\AnvilAssistant",
    ]
    print("Используются пути по умолчанию")

# 🔹 Собираем все .db файлы
sqlite_files = []
for folder in sqlite_folders:
    sqlite_files.extend(glob.glob(os.path.join(folder, "*.db")))

print("Нашли базы:", sqlite_files)

# 🔹 Сопоставление типов SQLite → MySQL
TYPE_MAPPING = {
    "INTEGER": "DECIMAL(65,0)",        # 🔥 изменено для поддержки больших значений
    "REAL": "DOUBLE",
    "TEXT": "TEXT",
    "BLOB": "LONGBLOB",
    "NUMERIC": "DECIMAL(18,6)"
}

# 🔹 Функция нормализации имени (для MySQL)
def normalize_name(name):
    return re.sub(r'\W+', '_', name).lower()

# 🔹 Функция для преобразования значений в совместимые типы
def convert_value(value):
    if pd.isna(value) or value is np.nan or str(value).lower() == 'nan':
        return None
    elif isinstance(value, (np.integer, np.floating)):
        return value.item()  # Преобразуем numpy типы в стандартные Python типы
    elif isinstance(value, np.ndarray):
        return value.tolist()  # Преобразуем массивы в списки
    else:
        return value

# 🔹 Подключение к MySQL без указания базы данных
mysql_conn = mysql.connector.connect(
    host=mysql_config["host"],
    user=mysql_config["user"],
    password=mysql_config["password"],
    port=mysql_config["port"],
    charset=mysql_config["charset"],          # 🔥 Добавлено для правильной кодировки
    use_unicode=mysql_config["use_unicode"]   # 🔥 Добавлено для поддержки Unicode
)
mysql_cursor = mysql_conn.cursor()

# 🔹 Создаем единую базу данных "Fabrica"
mysql_cursor.execute("CREATE DATABASE IF NOT EXISTS `Fabrica` DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;")
mysql_conn.commit()

# 🔹 Подключаемся к единой базе данных "Fabrica"
mysql_cursor.execute("USE `Fabrica`;")

for sqlite_file in sqlite_files:
    # Получаем имя базы данных из имени файла
    db_name = normalize_name(os.path.splitext(os.path.basename(sqlite_file))[0])
    print(f"\n📂 Обработка {sqlite_file} → таблицы с префиксом `{db_name}`")
    
    sqlite_conn = sqlite3.connect(sqlite_file)
    sqlite_cursor = sqlite_conn.cursor()
    
    # Получаем список таблиц
    sqlite_cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
    tables = sqlite_cursor.fetchall()
    
    for table_name, in tables:
        # Создаем имя таблицы с префиксом
        table_mysql = f"{db_name}_{normalize_name(table_name)}"
        print(f"  ➡️ Копируем таблицу {table_name} → Fabrica.{table_mysql}")
        
        # Читаем схему таблицы
        sqlite_cursor.execute(f"PRAGMA table_info({table_name})")
        columns_info = sqlite_cursor.fetchall()
        
        if not columns_info:
            print(f"    ⚠️ Таблица {table_name} пуста (нет колонок)")
            continue
        
        # Формируем CREATE TABLE
        col_defs = []
        primary_keys = []
        for col in columns_info:
            col_name = normalize_name(col[1])
            col_type = (col[2] or "TEXT").upper()
            mysql_type = TYPE_MAPPING.get(col_type.split("(")[0], "TEXT")
            
            # Для текстовых полей, используемых как первичный ключ, добавляем длину
            if col[5] == 1:  # Primary Key
                if mysql_type == "TEXT":
                    col_defs.append(f"`{col_name}` VARCHAR(255) PRIMARY KEY")
                else:
                    col_defs.append(f"`{col_name}` {mysql_type} PRIMARY KEY")
                primary_keys.append(col_name)
            else:
                # Для обычных текстовых полей также используем VARCHAR вместо TEXT в ключах
                if mysql_type == "TEXT":
                    col_defs.append(f"`{col_name}` TEXT")
                else:
                    col_defs.append(f"`{col_name}` {mysql_type}")
        
        # Удаляем таблицу, если она существует, и создаем заново
        mysql_cursor.execute(f"DROP TABLE IF EXISTS `{table_mysql}`")
        create_sql = f"CREATE TABLE `{table_mysql}` ({', '.join(col_defs)})"
        mysql_cursor.execute(create_sql)
        
        # Загружаем данные через pandas
        df = pd.read_sql_query(f"SELECT * FROM {table_name}", sqlite_conn)
        
        if df.empty:
            print(f"    ⚠️ Таблица {table_name} пустая, пропускаем.")
            continue
        
        # Переименовываем колонки
        df.columns = [normalize_name(c) for c in df.columns]
        
        # Заменяем NaN значения на None
        df = df.where(pd.notnull(df), None)
        
        # Обрабатываем значения
        for col in df.columns:
            df[col] = df[col].apply(convert_value)
        
        # Вставляем строки
        for _, row in df.iterrows():
            # Преобразуем значения в кортеж, заменяя NaN на None
            values = tuple(convert_value(val) for val in row.values)
            placeholders = ", ".join(["%s"] * len(values))
            sql = f"INSERT INTO `{table_mysql}` VALUES ({placeholders})"
            mysql_cursor.execute(sql, values)
        
        mysql_conn.commit()
    
    sqlite_conn.close()

mysql_cursor.close()
mysql_conn.close()
print("\n✅ Все базы перенесены в единую базу данных 'Fabrica'! 🎉")