86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
from utils.sql_utils import SQLUtils
|
||
import datetime
|
||
import os
|
||
import logging
|
||
from utils.config_loader import ConfigLoader
|
||
|
||
def init_database():
|
||
# 获取SQLite数据源的路径
|
||
config_loader = ConfigLoader.get_instance()
|
||
sqlite_config = config_loader.get_database_config('sqlite')
|
||
db_path = sqlite_config.get('path', 'db/jtDB.db')
|
||
|
||
# 确保db目录存在
|
||
db_dir = os.path.dirname(db_path)
|
||
os.makedirs(db_dir, exist_ok=True)
|
||
|
||
logging.info(f"初始化数据库: {db_path}")
|
||
db = SQLUtils('sqlite', database=db_path)
|
||
|
||
try:
|
||
db.begin_transaction()
|
||
|
||
# 创建用户表
|
||
create_user_table_sql = """
|
||
CREATE TABLE IF NOT EXISTS user (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
username VARCHAR(50) NOT NULL UNIQUE,
|
||
password VARCHAR(100) NOT NULL,
|
||
create_time TIMESTAMP NOT NULL,
|
||
create_by VARCHAR(50) NOT NULL,
|
||
update_time TIMESTAMP,
|
||
update_by VARCHAR(50),
|
||
is_deleted BOOLEAN DEFAULT FALSE
|
||
);
|
||
"""
|
||
db.execute_query(create_user_table_sql)
|
||
|
||
# 创建电力消耗表
|
||
create_electricity_table_sql = """
|
||
CREATE TABLE IF NOT EXISTS wsbz_electricity_consumption (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
sync_time TIMESTAMP,
|
||
electricity_number REAL
|
||
);
|
||
"""
|
||
db.execute_query(create_electricity_table_sql)
|
||
logging.info("已创建电力消耗表")
|
||
|
||
# 获取当前时间
|
||
current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
# 插入系统用户
|
||
insert_system_user_sql = """
|
||
INSERT OR IGNORE INTO user (
|
||
username, password, create_time, create_by, is_deleted
|
||
) VALUES (
|
||
'system', '123456', ?, 'system', FALSE
|
||
);
|
||
"""
|
||
db.execute_query(insert_system_user_sql, (current_time,))
|
||
|
||
# 检查是否存在schema.sql文件，如果存在则执行其中的SQL
|
||
schema_file = 'db/schema.sql'
|
||
if os.path.exists(schema_file):
|
||
with open(schema_file, 'r', encoding='utf-8') as f:
|
||
schema_sql = f.read()
|
||
|
||
# 按语句分割并执行SQL
|
||
statements = schema_sql.split(';')
|
||
for statement in statements:
|
||
statement = statement.strip()
|
||
if statement: # 跳过空语句
|
||
db.execute_query(statement)
|
||
|
||
logging.info("已执行schema.sql中的数据库初始化脚本")
|
||
|
||
db.commit_transaction()
|
||
logging.info("数据库初始化成功!")
|
||
except Exception as e:
|
||
db.rollback_transaction()
|
||
logging.error(f"数据库初始化失败: {str(e)}")
|
||
finally:
|
||
db.close()
|
||
|
||
if __name__ == "__main__":
|
||
init_database() |