自建rarbg_db搜索站

docker:
nichea6/rarbg_search
nichea6/rarbg_search_without_db

数据库转换脚本

import mysql.connector
from tqdm import tqdm
import logging

# 配置日志记录
logging.basicConfig(filename='error.log', level=logging.ERROR,
                    format='%(asctime)s - %(levelname)s - %(message)s')


def transfer_data(sqlite_db_path, mysql_host, mysql_user, mysql_password, mysql_database):
    sqlite_conn = None
    mysql_conn = None
    batch_size = 5000
    has_error = False

    try:
        # 连接数据库
        sqlite_conn = sqlite3.connect(sqlite_db_path)
        mysql_conn = mysql.connector.connect(
            host=mysql_host,
            user=mysql_user,
            password=mysql_password,
            database=mysql_database,
            use_pure=True  # 提升批量插入性能
        )
        mysql_conn.autocommit = False

        # 获取 SQLite 表结构
        sqlite_result = sqlite_conn.execute("PRAGMA table_info(items)")
        if sqlite_result is None:
            raise ValueError("无法获取SQLite表结构信息")
        sqlite_cols = []
        for col in sqlite_result:
            col_name = col[1]
            col_type = col[2]
            # 简单映射 SQLite 到 MySQL 数据类型
            if col_type.upper().startswith('INT'):
                col_type = 'INT'
            elif col_type.upper().startswith('TEXT'):
                col_type = 'TEXT'
            elif col_type.upper().startswith('REAL'):
                col_type = 'DOUBLE'
            sqlite_cols.append(f"`{col_name}` {col_type}")

        # 在 MySQL 中创建表
        create_table_sql = f"CREATE TABLE IF NOT EXISTS items ({', '.join(sqlite_cols)})"
        mysql_cursor = mysql_conn.cursor()
        mysql_cursor.execute(create_table_sql)
        mysql_conn.commit()

        # 再次获取 MySQL 表结构以验证
        mysql_cursor.execute("DESCRIBE items")
        mysql_result = mysql_cursor.fetchall()
        if mysql_result is None:
            raise ValueError("无法获取MySQL表结构信息")
        mysql_cols = [col[0] for col in mysql_result]

        if len(sqlite_cols) != len(mysql_cols):
            raise ValueError(f"表结构不一致\nSQLite: {sqlite_cols}\nMySQL: {mysql_cols}")

        # 准备插入语句
        insert_sql = f"INSERT INTO items ({', '.join([f'`{c}`' for c in mysql_cols])}) VALUES ({', '.join(['%s'] * len(mysql_cols))})"

        # 分批处理
        with tqdm(desc="数据迁移进度", unit="rows") as pbar:
            sqlite_cursor = sqlite_conn.cursor()
            sqlite_cursor.execute("SELECT * FROM items")
            while True:
                batch = sqlite_cursor.fetchmany(batch_size)
                if not batch:
                    break

                try:
                    mysql_cursor.executemany(insert_sql, batch)
                    mysql_conn.commit()
                    pbar.update(len(batch))
                except mysql.connector.Error as err:
                    mysql_conn.rollback()
                    tqdm.write(f"错误跳过批次: {err}")
                    # 记录错误信息到日志
                    logging.error(f"错误跳过批次: {err}, 批次数据: {batch}")

    except Exception as e:
        has_error = True
        print(f"\n迁移中止: {str(e)}")
        logging.error(f"迁移中止: {str(e)}")
        if mysql_conn:
            mysql_conn.rollback()
    finally:
        if sqlite_conn:
            sqlite_conn.close()
        if mysql_conn:
            mysql_conn.close()
        if not has_error:
            print("\n数据迁移成功完成")

if __name__ == "__main__":
    # 配置参数...
    sqlite_db_path = 'rarbg_db.sqlite'
    mysql_host = 'your_db_host'
    mysql_user = 'your_db_user'
    mysql_password = 'your_db_password'
    mysql_database = 'your_db'
    transfer_data(sqlite_db_path, mysql_host, mysql_user, mysql_password, mysql_database)

rarbg_db.sqlite:
链接：https://pan.quark.cn/s/cab1760c7020
提取码：pW9a