forked from Lesin/reposync
75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
import aiomysql
|
|
import asyncio
|
|
import os
|
|
|
|
async def test_connection():
|
|
# 数据库连接配置
|
|
db_config = {
|
|
'host': '127.0.0.1',
|
|
'port': 3307,
|
|
'user': 'root',
|
|
'password': 'anye97!.!'
|
|
}
|
|
|
|
try:
|
|
# 测试连接
|
|
conn = await aiomysql.connect(**db_config)
|
|
print("数据库连接成功！")
|
|
await conn.close()
|
|
except Exception as e:
|
|
print(f"数据库连接失败: {str(e)}")
|
|
|
|
async def init_database():
|
|
# 读取SQL文件
|
|
with open('sql/table.sql', 'r', encoding='utf-8') as f:
|
|
sql_content = f.read()
|
|
|
|
# 数据库连接配置
|
|
db_config = {
|
|
'host': '127.0.0.1',
|
|
'port': 3307,
|
|
'user': 'root',
|
|
'password': 'anye97!.!',
|
|
'db': 'ob_sync'
|
|
}
|
|
|
|
try:
|
|
# 首先创建数据库
|
|
conn = await aiomysql.connect(
|
|
host=db_config['host'],
|
|
port=db_config['port'],
|
|
user=db_config['user'],
|
|
password=db_config['password']
|
|
)
|
|
|
|
async with conn.cursor() as cur:
|
|
# 创建数据库
|
|
await cur.execute(f"CREATE DATABASE IF NOT EXISTS {db_config['db']}")
|
|
print(f"数据库 {db_config['db']} 创建成功")
|
|
|
|
await conn.close()
|
|
|
|
# 连接到新创建的数据库
|
|
conn = await aiomysql.connect(**db_config)
|
|
|
|
async with conn.cursor() as cur:
|
|
# 执行SQL文件中的内容
|
|
for statement in sql_content.split(';'):
|
|
if statement.strip():
|
|
await cur.execute(statement)
|
|
print("执行SQL语句成功")
|
|
|
|
await conn.commit()
|
|
print("数据库初始化完成")
|
|
|
|
except Exception as e:
|
|
print(f"发生错误: {str(e)}")
|
|
finally:
|
|
if 'conn' in locals():
|
|
await conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
# 先测试数据库连接
|
|
asyncio.run(test_connection())
|
|
# 如果连接成功，再初始化数据库
|
|
# asyncio.run(init_database()) |