import json
|
import pymysql
|
import logging
|
|
# 设置日志记录
|
logging.basicConfig(level=logging.INFO)
|
logger = logging.getLogger(__name__)
|
|
|
class DatabaseConfig:
|
def __init__(self, config_file="config.json"):
|
"""从JSON文件初始化数据库配置"""
|
try:
|
with open(config_file) as f:
|
config = json.load(f)
|
except FileNotFoundError:
|
logger.error(f"配置文件 {config_file} 未找到")
|
raise
|
except json.JSONDecodeError:
|
logger.error(f"配置文件 {config_file} 中JSON格式无效")
|
raise
|
|
if "database" not in config:
|
logger.error("配置文件中未找到数据库配置")
|
raise ValueError("数据库配置缺失")
|
|
db_config = config["database"]
|
|
self.host = db_config.get("host", "localhost")
|
self.port = int(db_config.get("port", 3306))
|
self.user = db_config.get("user", "root")
|
self.password = db_config.get("password", "")
|
self.database_name = db_config.get("database_name", "admin_db")
|
self.charset = db_config.get("charset", "utf8mb4")
|
|
def get_connection(self):
|
"""建立并返回MySQL数据库连接"""
|
try:
|
connection = pymysql.connect(
|
host=self.host,
|
port=self.port,
|
user=self.user,
|
password=self.password,
|
database=self.database_name,
|
charset=self.charset,
|
cursorclass=pymysql.cursors.DictCursor,
|
)
|
return connection
|
except Exception as e:
|
logger.error(f"连接数据库时出错: {e}")
|
return None
|
|
|
def init_database(config_file="config.json"):
|
"""初始化数据库和所需表"""
|
try:
|
with open(config_file) as f:
|
config = json.load(f)
|
except FileNotFoundError:
|
logger.error(f"配置文件 {config_file} 未找到")
|
raise
|
except json.JSONDecodeError:
|
logger.error(f"配置文件 {config_file} 中JSON格式无效")
|
raise
|
|
if "database" not in config:
|
logger.error("配置文件中未找到数据库配置")
|
raise ValueError("数据库配置缺失")
|
|
db_config = config["database"]
|
|
# 不指定数据库连接以创建数据库（如果需要）
|
try:
|
connection = pymysql.connect(
|
host=db_config.get("host", "localhost"),
|
port=int(db_config.get("port", 3306)),
|
user=db_config.get("user", "root"),
|
password=db_config.get("password", ""),
|
charset=db_config.get("charset", "utf8mb4"),
|
)
|
except Exception as e:
|
logger.error(f"建立初始数据库连接时出错: {e}")
|
raise
|
|
try:
|
with connection.cursor() as cursor:
|
# 如果数据库不存在则创建
|
cursor.execute(
|
f"CREATE DATABASE IF NOT EXISTS {db_config.get('database_name', 'admin_db')}"
|
)
|
connection.commit()
|
|
# 选择数据库
|
cursor.execute(f"USE {db_config.get('database_name', 'admin_db')}")
|
|
# 创建ContentType表 - 包含id、code、name字段
|
create_contenttype_table_sql = """
|
CREATE TABLE IF NOT EXISTS content_type (
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
code VARCHAR(50) NOT NULL UNIQUE,
|
name VARCHAR(255) NOT NULL,
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
INDEX idx_code (code)
|
)
|
"""
|
cursor.execute(create_contenttype_table_sql)
|
|
# 创建Content表 - 包含id、Type(外键)、Question、Answer字段
|
create_content_table_sql = """
|
CREATE TABLE IF NOT EXISTS content (
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
type_id INT NOT NULL,
|
question TEXT NOT NULL,
|
answer TEXT NOT NULL,
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
FOREIGN KEY (type_id) REFERENCES content_type(id) ON DELETE CASCADE,
|
INDEX idx_type (type_id)
|
)
|
"""
|
cursor.execute(create_content_table_sql)
|
|
connection.commit()
|
|
logger.info("数据库和表初始化成功!")
|
except Exception as e:
|
logger.error(f"初始化数据库时出错: {e}")
|
raise
|
finally:
|
connection.close()
|
