81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""数据库初始化脚本。
|
|
|
|
用法：
|
|
python scripts/init_db.py # 根据模型创建表
|
|
python scripts/init_db.py --sql <文件> # 导入 SQL 结构文件
|
|
"""
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 将 src 目录加入 sys.path
|
|
SRC = Path(__file__).resolve().parent.parent / "src"
|
|
sys.path.insert(0, str(SRC))
|
|
|
|
from config.database import DATABASE_URL
|
|
from extensions.sqlalchemy import Base, engine
|
|
|
|
|
|
def import_sql(sql_file: Path) -> None:
|
|
"""导入 SQL 文件（按分号+换行拆分逐条执行，避免字符串内分号误切）。"""
|
|
from sqlalchemy import text
|
|
|
|
if not sql_file.exists():
|
|
print(f"SQL 文件不存在: {sql_file}")
|
|
sys.exit(1)
|
|
|
|
content = sql_file.read_text(encoding="utf-8", errors="replace")
|
|
statements = [s.strip() for s in content.split(";\n") if s.strip()]
|
|
|
|
with engine.begin() as conn:
|
|
for statement in statements:
|
|
conn.execute(text(statement))
|
|
print(f"SQL 导入完成: {sql_file}（共 {len(statements)} 条语句）")
|
|
|
|
|
|
def create_tables() -> None:
|
|
"""根据模型定义创建全部数据表。"""
|
|
# 导入各模块 models 触发注册
|
|
import modules.auth.models
|
|
import modules.user.models
|
|
import modules.user_role.models
|
|
import modules.role.models
|
|
import modules.role_menu.models
|
|
import modules.menu.models
|
|
import modules.dept.models
|
|
import modules.category.models
|
|
import modules.article.models
|
|
import modules.param.models
|
|
import modules.file_template.models
|
|
import modules.job.models
|
|
import modules.dictionary.dict.models
|
|
import modules.dictionary.dict_item.models
|
|
import modules.configuration.config.models
|
|
import modules.configuration.config_item.models
|
|
import modules.level.models
|
|
import modules.position.models
|
|
import modules.city.models
|
|
import modules.notice.models
|
|
import modules.link.models
|
|
import modules.operation_log.models
|
|
import modules.login_log.models
|
|
import modules.job_log.models
|
|
import modules.example.models
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
print(f"数据表创建完成，目标数据库: {DATABASE_URL.rsplit('/', 1)[-1]}")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="数据库初始化")
|
|
parser.add_argument("--sql", type=Path, default=None, help="SQL 结构文件路径")
|
|
args = parser.parse_args()
|
|
if args.sql:
|
|
import_sql(args.sql)
|
|
else:
|
|
create_tables()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|