88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
import toml
|
|
import pymysql
|
|
from pymysql.cursors import DictCursor
|
|
from pymysql.connections import Connection
|
|
from models.logger import setup_logger
|
|
|
|
logger = setup_logger()
|
|
|
|
def connect_db() -> Connection | None:
|
|
"""连接数据库，如果没有则初始化数据库和表"""
|
|
try:
|
|
config = toml.load("config.toml")
|
|
db_host = config["db_host"]
|
|
db_port = config["db_port"]
|
|
db_user = config["db_user"]
|
|
db_password = config["db_password"]
|
|
db_database = config["db_database"]
|
|
except Exception as e:
|
|
logger.error(f"读取数据库配置错误{e}")
|
|
return None
|
|
|
|
connection = pymysql.connect(
|
|
host=db_host,
|
|
port=int(db_port),
|
|
user=db_user,
|
|
password=db_password,
|
|
database=db_database,
|
|
charset='utf8mb4',
|
|
cursorclass=DictCursor
|
|
)
|
|
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
# 创建 emoji 表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS emoji (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
uuid VARCHAR(255) NOT NULL,
|
|
description VARCHAR(255),
|
|
url VARCHAR(255) NOT NULL,
|
|
hash VARCHAR(255) NOT NULL
|
|
)
|
|
''')
|
|
|
|
# 创建 approved 表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS approved
|
|
(
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
uuid VARCHAR(255) NOT NULL,
|
|
approved BOOL,
|
|
approver VARCHAR(255)
|
|
)
|
|
''')
|
|
|
|
# 创建 token 表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS token
|
|
(
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
token VARCHAR(255) NOT NULL ,
|
|
userid INT NOT NULL ,
|
|
expires TIMESTAMP NOT NULL
|
|
)
|
|
''')
|
|
|
|
# 创建 user 表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS user
|
|
(
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(255) NOT NULL,
|
|
passwd VARCHAR(255) NOT NULL,
|
|
email VARCHAR(255),
|
|
lastlogin TIMESTAMP,
|
|
submission_count INT DEFAULT 0
|
|
)
|
|
''')
|
|
|
|
return connection
|
|
except Exception as e:
|
|
logger.error(f"数据库连接或表创建失败: {e}")
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
conn = connect_db()
|
|
conn.cursor() |