"""测试数据填充脚本。 用法：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 "), ] 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()