197 lines
8.4 KiB
Python
197 lines
8.4 KiB
Python
"""测试数据填充脚本。
|
||
|
||
用法：python scripts/seed.py
|
||
- 创建默认管理员账号（admin / 123456）
|
||
- 填充任务调度示例数据：任务菜单/权限节点、示例任务、示例任务日志
|
||
"""
|
||
import sys
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
from sqlalchemy import inspect
|
||
|
||
SRC = Path(__file__).resolve().parent.parent / "src"
|
||
sys.path.insert(0, str(SRC))
|
||
|
||
from common.utils.crypto import set_password
|
||
from config.app import TORNADO_PORT
|
||
from extensions.sqlalchemy import SessionLocal
|
||
from modules.user.models import User
|
||
|
||
|
||
def seed_admin() -> None:
|
||
"""创建默认管理员账号（幂等）。"""
|
||
db = SessionLocal()
|
||
try:
|
||
exists = db.query(User).filter(User.username == "admin").first()
|
||
if exists:
|
||
print("管理员账号已存在，跳过。")
|
||
return
|
||
salt, hashed = set_password("123456")
|
||
admin = User(
|
||
realname="超级管理员",
|
||
nickname="管理员",
|
||
username="admin",
|
||
password=hashed,
|
||
salt=salt,
|
||
gender=1,
|
||
status=1,
|
||
sort=0,
|
||
)
|
||
db.add(admin)
|
||
db.commit()
|
||
print("管理员账号创建成功：admin / 123456")
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(f"填充失败: {e}")
|
||
sys.exit(1)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _ensure_table(db, required: tuple) -> bool:
|
||
"""校验所需数据表是否存在，不存在则提示并返回 False。"""
|
||
existing = set(inspect(db.bind).get_table_names())
|
||
missing = [t for t in required if t not in existing]
|
||
if missing:
|
||
print(f"表不存在，跳过任务数据填充：{', '.join(missing)}（请先执行 alembic upgrade head）")
|
||
return False
|
||
return True
|
||
|
||
|
||
def seed_job_data() -> None:
|
||
"""填充任务调度示例数据（幂等）：菜单/权限节点 + 示例任务 + 示例任务日志。"""
|
||
db = SessionLocal()
|
||
try:
|
||
from modules.job.models import Job
|
||
from modules.job_log.models import JobLog
|
||
from modules.menu.models import Menu
|
||
|
||
if not _ensure_table(db, ("tornado_menu", "tornado_job", "tornado_job_log")):
|
||
return
|
||
|
||
# ---------- 1. 菜单与权限节点（挂到"系统管理"下） ----------
|
||
system = db.query(Menu).filter(Menu.title == "系统管理", Menu.parent_id == 0).first()
|
||
parent_id = system.id if system else 0
|
||
|
||
# 任务调度菜单
|
||
job_menu = db.query(Menu).filter(Menu.permission == "sys:job:view").first()
|
||
if not job_menu:
|
||
job_menu = Menu(title="任务调度", icon="el-icon-alarm-clock", parent_id=parent_id,
|
||
path="/system/job", component="/system/job", target="_self",
|
||
permission="sys:job:view", type=0, status=1, hide=0, sort=40)
|
||
db.add(job_menu)
|
||
db.flush()
|
||
print("创建菜单：任务调度")
|
||
else:
|
||
print("菜单已存在：任务调度")
|
||
|
||
# 任务日志菜单
|
||
joblog_menu = db.query(Menu).filter(Menu.permission == "sys:joblog:view").first()
|
||
if not joblog_menu:
|
||
joblog_menu = Menu(title="任务日志", icon="el-icon-document", parent_id=parent_id,
|
||
path="/system/joblog", component="/system/joblog", target="_self",
|
||
permission="sys:joblog:view", type=0, status=1, hide=0, sort=45)
|
||
db.add(joblog_menu)
|
||
db.flush()
|
||
print("创建菜单：任务日志")
|
||
else:
|
||
print("菜单已存在：任务日志")
|
||
|
||
# 任务调度权限节点（type=1，按钮级）
|
||
job_nodes = [
|
||
("查询任务", "sys:job:list", "/job/list", 1),
|
||
("任务详情", "sys:job:detail", "/job/detail", 5),
|
||
("添加任务", "sys:job:add", "/job/add", 10),
|
||
("修改任务", "sys:job:update", "/job/update", 15),
|
||
("删除任务", "sys:job:delete", "/job/delete", 20),
|
||
("设置状态", "sys:job:status", "/job/status", 25),
|
||
("立即执行", "sys:job:run", "/job/run", 30),
|
||
]
|
||
for title, perm, path, sort in job_nodes:
|
||
if not db.query(Menu).filter(Menu.permission == perm).first():
|
||
db.add(Menu(title=title, icon="", parent_id=job_menu.id, path=path,
|
||
component="", target="_self", permission=perm, type=1,
|
||
status=1, hide=0, sort=sort))
|
||
print(f"创建权限节点：{title}（{perm}）")
|
||
|
||
# 任务日志权限节点
|
||
joblog_nodes = [
|
||
("查询日志", "sys:joblog:list", "/joblog/list", 1),
|
||
("日志详情", "sys:joblog:detail", "/joblog/detail", 5),
|
||
("删除日志", "sys:joblog:delete", "/joblog/delete", 10),
|
||
]
|
||
for title, perm, path, sort in joblog_nodes:
|
||
if not db.query(Menu).filter(Menu.permission == perm).first():
|
||
db.add(Menu(title=title, icon="", parent_id=joblog_menu.id, path=path,
|
||
component="", target="_self", permission=perm, type=1,
|
||
status=1, hide=0, sort=sort))
|
||
print(f"创建权限节点：{title}（{perm}）")
|
||
|
||
# ---------- 2. 示例任务（回调指向本服务健康检查接口，可真实触发并产生日志） ----------
|
||
base_url = f"http://127.0.0.1:{TORNADO_PORT}/api/v1/health"
|
||
sample_jobs = [
|
||
dict(job_name="应用健康检查", job_alias="health", job_group="系统任务",
|
||
job_trigger="cron", status=1, cron_expression="*/5 * * * *",
|
||
execute_policy=1, is_sync=0, url=base_url,
|
||
note="每5分钟检测应用健康状态（示例任务）"),
|
||
dict(job_name="夜间数据备份", job_alias="backup", job_group="系统任务",
|
||
job_trigger="cron", status=2, cron_expression="0 0 2 * * ?",
|
||
execute_policy=1, is_sync=0, url=base_url,
|
||
note="每日凌晨2点执行数据备份（示例任务，已暂停）"),
|
||
dict(job_name="订单数据同步", job_alias="sync", job_group="业务任务",
|
||
job_trigger="cron", status=0, cron_expression="0 */30 * * * *",
|
||
execute_policy=2, is_sync=0, url=base_url,
|
||
note="每30分钟同步订单数据（示例任务，未发布）"),
|
||
]
|
||
for item in sample_jobs:
|
||
if db.query(Job).filter(Job.job_name == item["job_name"]).first():
|
||
print(f"任务已存在：{item['job_name']}")
|
||
continue
|
||
db.add(Job(**item))
|
||
print(f"创建示例任务：{item['job_name']}")
|
||
# flush 让新增任务可见，供下方示例日志回填真实 job_id
|
||
db.flush()
|
||
|
||
# ---------- 3. 示例任务日志（仅表为空时填充，按任务名回查真实 job_id） ----------
|
||
def _job_id(name: str):
|
||
job = db.query(Job).filter(Job.job_name == name).first()
|
||
return job.id if job else None
|
||
|
||
if db.query(JobLog).count() == 0:
|
||
now = datetime.now()
|
||
refs = [
|
||
("应用健康检查", 0, 156, None),
|
||
("应用健康检查", 0, 203, None),
|
||
("应用健康检查", 1, 10001, "urlopen error <urlopen error timed out>"),
|
||
]
|
||
sample_logs = []
|
||
for idx, (name, status, consume, exc) in enumerate(refs):
|
||
jid = _job_id(name)
|
||
if jid is None:
|
||
continue
|
||
start = now - timedelta(minutes=(idx + 1) * 5)
|
||
sample_logs.append(JobLog(
|
||
job_id=jid, job_name=name, job_group="系统任务", job_trigger="cron",
|
||
job_message="任务触发成功" if status == 0 else "任务触发失败",
|
||
cron_expression="*/5 * * * *", status=status,
|
||
start_time=start, end_time=start + timedelta(milliseconds=consume),
|
||
consume_time=consume, exception_info=exc,
|
||
))
|
||
db.add_all(sample_logs)
|
||
print(f"创建示例任务日志 {len(sample_logs)} 条")
|
||
|
||
db.commit()
|
||
print("任务调度示例数据填充完成。")
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(f"任务数据填充失败: {e}")
|
||
sys.exit(1)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
seed_admin()
|
||
seed_job_data()
|