New-API 部署与迁移

import sqlite3
import pymysql
import re

# 从环境变量中获取 MySQL 连接信息
sql_dsn = "用户名:密码@tcp(主机名:端口号)/数据库名"

if not sql_dsn:
    raise ValueError("环境变量 SQL_DSN 未设置。请按照格式 SQL_DSN=用户名:密码@tcp(主机名:端口号)/数据库名 来设置。")

# 使用正则表达式解析 SQL_DSN
pattern = r'(?P<user>[^:]+):(?P<password>[^@]+)@tcp\((?P<host>[^:]+):(?P<port>\d+)\)/(?P<database>.+)'
match = re.match(pattern, sql_dsn)

if not match:
    raise ValueError("环境变量 SQL_DSN 格式不正确，应为 用户名:密码@tcp(主机名:端口号)/数据库名")

mysql_config = match.groupdict()
# 将端口号转换为整数
mysql_config['port'] = int(mysql_config['port'])

# 连接到 SQLite 数据库
sqlite_conn = sqlite3.connect('one-api.db')
sqlite_cursor = sqlite_conn.cursor()
print("成功连接到 SQLite 数据库。")

# 连接到 MySQL 数据库
try:
    mysql_conn = pymysql.connect(
        host=mysql_config['host'],
        port=mysql_config['port'],
        user=mysql_config['user'],
        password=mysql_config['password'],
        database=mysql_config['database'],
        charset='utf8mb4'
    )
    mysql_cursor = mysql_conn.cursor()
    print("成功连接到 MySQL 数据库。")
except Exception as e:
    print(f"连接 MySQL 数据库时出现错误：{e}")
    exit(1)

# 获取 SQLite 中 channels 表的列名
sqlite_cursor.execute('PRAGMA table_info(channels)')
sqlite_columns_info = sqlite_cursor.fetchall()
sqlite_column_names = [column[1] for column in sqlite_columns_info]
print(f"SQLite 数据库中 channels 表的列名：{sqlite_column_names}")

# 获取 MySQL 中 channels 表的列名
mysql_cursor.execute('DESCRIBE channels')
mysql_columns_info = mysql_cursor.fetchall()
mysql_column_names = [column[0] for column in mysql_columns_info]
print(f"MySQL 数据库中 channels 表的列名：{mysql_column_names}")

# 找出两个表中共同的列名（排除自增主键 'id'）
common_columns = [col for col in sqlite_column_names if col in mysql_column_names and col != 'id']
print(f"将要插入的公共列名：{common_columns}")

# 获取所有行数据
sqlite_cursor.execute('SELECT * FROM channels')
rows = sqlite_cursor.fetchall()
print(f"从 SQLite 获取了 {len(rows)} 条记录。")

# 创建列名到索引的映射，方便从行数据中获取对应的值
sqlite_column_to_index = {column_name: index for index, column_name in enumerate(sqlite_column_names)}

# 过滤 rows，只保留共同列的数据
filtered_rows = []
for row in rows:
    filtered_row = tuple(row[sqlite_column_to_index[col]] for col in common_columns)
    filtered_rows.append(filtered_row)

# 构建插入 MySQL 的 SQL 语句
placeholders = ', '.join(['%s'] * len(common_columns))
columns = ', '.join(f"`{col}`" for col in common_columns)
insert_sql = f"INSERT INTO channels ({columns}) VALUES ({placeholders})"
print(f"生成的 INSERT 语句：{insert_sql}")

# 将数据插入到 MySQL 数据库中
try:
    mysql_cursor.executemany(insert_sql, filtered_rows)
    mysql_conn.commit()
    print(f"成功插入 {mysql_cursor.rowcount} 条记录到 MySQL 数据库中。")
except Exception as e:
    mysql_conn.rollback()
    print(f"数据插入过程中出现错误：{e}")
finally:
    # 关闭连接
    sqlite_conn.close()
    mysql_cursor.close()
    mysql_conn.close()
    print("关闭数据库连接。")