Files
lovife 1f3c485ceb feat(role,scope)!: 角色与授权可配 + 需求收敛(4 角色 / 10 菜单 / 删两模块)
角色与授权(交接摘要 §一·三十五):
- 权限判定合并两个来源:内置角色按代码(BuiltinRole)、自定义角色按库
  (sys_role_permission)。实现放 module 层的 DbRoleGrantSource,security 只依赖
  接口 RoleGrantSource —— 否则角色服务要调 permissionChecker,两个包迟早成环
- 权限判定必须加缓存:此前是零 IO,而压测基线是"SQL 条数恒定"。缓存放 Redis,
  空结果也缓存(否则已删角色的 token 会把库打穿)、Redis 挂了退化成直读库、TTL 只兜底
- V18 建 sys_role.user_type(账号的 user_type 一直从角色反推,自定义角色不在代码映射里)
  与 sys_role_menu;8 个接口 + 错误码 40045;契约 119 -> 129
- 内置角色只读(权限码是前后端契约,做成可改 = 一次误操作就能把管理员锁在系统外);
  平台级能力两道防线(界面禁用 + 服务端 40301);防自锁;改完立即生效(写授权清缓存)
- RoleProvisioner 从"已存在就 continue"改成幂等回填 + 每日 03:20 定时任务 ——
  修的是"代码改了权限、已开通租户永远停在发放那一刻",这种漂移不报错也没测试会红
- scripts/e2e-m10-role.py 77/77

需求收敛(交接摘要 §一·三十六,本节的数字为准):
- 删副班主任 / 审核员两个内置角色(6 -> 4)、历史整理器(module/cleanup)、
  加入申请(module/join + 学生端 pages/join)、collect_task_org.deputy_user_ids 整列;
  审核范围三级回退 -> 两级,连带 6 个 mapper 的 LIKE '%,id,%' 分支
- V19 迁移处理存量:先解 sys_user_role 绑定再软删角色;账号只在"除这两个角色外没有
  别的角色"时才软删。必须迁而不能只删代码 —— 角色码删掉后 mergePermissions 返回空集,
  存量账号会"能登录、处处 40301",不抛异常不打日志
- 契约 129 -> 119、错误码 59 -> 54、权限码 30 -> 28、PC 菜单 12 -> 10、
  小程序页 10 -> 9、原型 32 -> 29 屏、E2E 39 -> 37 个(断言 1695 条)
- UserService.list 对不认识的角色码由"静默丢掉过滤条件"改成失败关闭
  (之前传个不存在的 roleCode 会返回全校账号,且看起来完全正常)
- 品牌元信息同步改名「深学在线」:pom 名称、/doc.html 标题、配置与启动类注释、
  存储自检探针内容;学生端 app 名与全局导航栏标题
2026-10-10 16:48:54 +08:00

411 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""摆一份**可用于界面验收的审核台数据**(开发用,不属于 E2E,不会清理)。
为什么要这么一个脚本
--------------------
审核台(PC)与班主任端页面验收时需要"名单里有人已提交"的状态,而 E2E 脚本
跑完会**清掉名册绑定**。手工在页面上造:建任务 → 配班级 → 配材料 → 导名册 →
三个学生各自核验 + 填表 + 传图 + 提交 —— 一次十几分钟,而且每次都要重来。
它做的全部事情都是**走真实接口**(建任务 / 配班级 / 应用材料模板 / 导名册 /
门禁核验 / 填表 / 传材料 / 提交 / 初审通过),所以摆出来的状态与真实使用完全一致。
跑法
----
python scripts/dev-seed-audit.py # 建演示任务 + 3 个学生 + 3 份提交单
python scripts/dev-seed-audit.py --reset # 先删掉旧的演示任务再重建
python scripts/dev-seed-audit.py --clean # 只清理演示任务
产出(浏览器里怎么用)
----------------------
· 任务「【演示】审核台」→ 名单管理:3 人,3 人已提交
· 审核 → 初审:**演示甲(干净,可批量通过)/ 演示乙(有黄色警告)**
· 审核 → 终审:**演示丙(已初审通过,等待终审)**
"""
from __future__ import annotations
import argparse
import time
import importlib.util
import os
import sys
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
BASE = "http://localhost:8080"
TASK_NAME = "【演示】审核台"
TMP_ROSTER = os.path.join(ROOT, "scripts", "_seed-audit-roster.xlsx")
# 演示班主任:**密码写死**,因为这个账号的用途就是"登录进去看初审那一屏"。
# 生产当然不能这么干,但它是 dev 演示数据(`--clean` 会连任务一起删)。
TEACHER_USERNAME = "demot1"
TEACHER_PASSWORD = "Demo1234"
TEACHER_NAME = "演示班主任"
TEACHER_PHONE = "13900139001"
def _load(module_name: str, filename: str):
"""
复用既有 E2E 脚本的工具(登录 / 造名册 / 造学生 / 校验位计算)。
⚠️ 不自己再写一份 `call`/`upload`/`mint`:两边一改一漏,
表现是"这个脚本单独跑得过、和 E2E 一起跑就红",而根因极难找。
"""
path = os.path.join(ROOT, "scripts", filename)
spec = importlib.util.spec_from_file_location(module_name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
R = _load("e2e_m2_roster", "e2e-m2-roster.py") # 服务端工具:登录 / 导入名册 / SQL
S = _load("e2e_m3_submit", "e2e-m3-submit.py") # 学生侧:填表 / 传材料 / 提交
A = _load("e2e_m5_audit", "e2e-m5-audit.py") # big_jpeg / prepare_student
# 三个学生:姓名必须与名册一致;证件号用 R.with_checksum 生成(校验位合法)
STUDENTS = [
(993101, "演示甲", R.with_checksum("36102920080101001"), True), # 干净 → 留作初审待办
(993102, "演示乙", R.with_checksum("36102920080202002"), False), # 重复件 → 黄色警告
(993103, "演示丙", R.with_checksum("36102920080303003"), False), # 干净但先被初审通过
]
def login() -> str:
st, body = R.login(BASE, "admin", "Admin@123", "demo")
if st != 200 or body.get("code") != 0:
raise SystemExit(f"登录失败:status={st} body={body}")
return body["data"]["token"]
def find_demo_tasks(token: str) -> list[dict]:
"""找出所有演示任务(**只认干净名字**,被清理过的那些已改名,不会再被选中)。"""
st, body = R.call("GET", f"{BASE}/api/v1/tasks?pageNum=1&pageSize=200", token)
return [t for t in ((body.get("data") or {}).get("list") or [])
if t.get("taskName") == TASK_NAME]
def clean(token: str, naming_template: bool = True) -> None:
"""
把上一次的演示数据清空并"挪开",好让本轮能重新用干净的任务名。
⚠️ **已发布的任务既删不掉也改不了名**(`40001`,产品口径:只有草稿能删/改)——
所以这里只能:
· 用 SQL 清掉它的**提交单 / 审核记录 / 材料文件 / 名册行**(测试夹具性质的操作,
与 E2E 用 SQL 造夹具同口径);
· 把任务名改成 `【演示·旧】审核台 {id}`,让 `find_demo_tasks` 不再选中它。
为什么非要清不可:演示数据必须每次都能**从零重建**,否则第二次跑就会与上一次
的数据纠缠("同一个学生有两份提交单"),而那看起来像功能坏了。
"""
tasks = find_demo_tasks(token)
if not tasks:
print(" 没有找到演示任务,无需清理")
return
for t in tasks:
tid = t["id"]
sql(f"delete from collect_audit_record where submission_id in "
f"(select id from collect_submission where task_id = {tid});")
sql(f"delete from collect_file where submission_id in "
f"(select id from collect_submission where task_id = {tid});")
sql(f"delete from collect_submission where task_id = {tid};")
sql(f"delete from collect_roster where task_id = {tid};")
if naming_template:
sql(f"update collect_task set task_name = '【演示·旧】审核台 {tid}' where id = {tid};")
print(f" 旧的演示任务 {tid} 已清空{'并改名' if naming_template else ''}")
def sql(stmt: str) -> None:
"""开发库上的夹具操作(与 E2E 脚本同一口径:`docker exec psql`)。"""
R.sql(stmt)
def _redis_del(key: str) -> None:
"""删一个 Redis 键(限流计数用)。
容器名/端口与 `deploy/docker-compose.yml` 一致:`xt-redis`(宿主 16379)。
⚠️ 本机还有别的项目的 redis(6379),所以这里**显式指定容器**,不用默认端口。
⚠️ **必须带密码**:`xt-redis` 开了 requirepass(dev 是 `redis_dev_pwd`,
见 `application-dev.yml`)。不传密码时 redis-cli 回 `NOAUTH ...`,
而我第一版用 `capture_output=True` 又**不看返回码** —— 于是表现为
"清了但没清掉",最后还是撞 40071,排查时完全看不出是密码问题。
"""
import subprocess
r = subprocess.run(["docker", "exec", "xt-redis", "redis-cli",
"-a", "redis_dev_pwd", "--no-auth-warning", "DEL", key],
capture_output=True, text=True, timeout=30)
out = (r.stdout or "").strip()
if r.returncode != 0 or "NOAUTH" in out or "WRONGPASS" in out:
print(f" ⚠️ 清 Redis 键失败({key}):{out or r.stderr.strip()[:160]}")
return None
def main() -> int:
ap = argparse.ArgumentParser(description="摆审核台演示数据")
ap.add_argument("--clean", action="store_true", help="只清理演示任务")
ap.add_argument("--reset", action="store_true", help="先清理再重建")
ap.add_argument("--reset-gate-limit", action="store_true",
help="先清掉演示学生的门禁限流计数(Redis)。"
"限流是按用户 1 小时 10 次,反复重跑会撞 40071。\n"
"⚠️ 只清演示学生的键,不动全局限流配置")
ap.add_argument("--approve-one", action="store_true",
help="再走一次真实终审,让演示丙变成「已通过」"
"(导出功能需要至少一个已通过的人;否则两种范围都导不出来)")
args = ap.parse_args()
token = login()
print(" 管理员登录 ok")
if args.clean or args.reset:
clean(token)
if args.clean:
return 0
if find_demo_tasks(token):
print(f" 演示任务已存在({TASK_NAME})—— 要重建请加 --reset")
return 0
# ── ① 建任务(草稿)────────────────────────────────
st, body = R.call("POST", f"{BASE}/api/v1/tasks", token, {
"taskName": TASK_NAME,
"batchNo": "202609",
"majorName": "数控加工",
"jobName": "车工",
"collectMode": 1,
"certType": 1,
"startTime": "2026-01-01",
"endTime": "2026-12-31"
})
if body.get("code") != 0:
raise SystemExit(f"建任务失败:{body}")
task_id = body["data"]["id"]
print(f" 建任务 id={task_id}")
# 班级:从组织树里找一个班级(演示租户已有)
st, body = R.call("GET", f"{BASE}/api/v1/orgs/tree", token)
classes = []
def walk(nodes):
for n in nodes or []:
if n.get("orgType") == 3:
classes.append(n)
walk(n.get("children"))
walk(body.get("data") or [])
if not classes:
raise SystemExit("演示租户里没有班级节点,请先在「组织架构」里建一个班")
org = classes[0]
print(f" 用班级:{org.get('orgName')}(id={org.get('id')})")
st, body = R.call("PUT", f"{BASE}/api/v1/tasks/{task_id}/orgs", token, {
"orgs": [{"orgId": org["id"], "expectedCount": len(STUDENTS)}]
})
if body.get("code") != 0:
raise SystemExit(f"配置班级失败:{body}")
# ── 门禁限流计数:只在显式要求时清 ──
#
# ⚠️ 门禁按用户 1 小时 10 次限流(`MpRosterService` 的 `xt:mp:gate:user:{uid}`),
# 而演示脚本每次都用同一批演示学生 uid —— 一小时内跑到第 4 次就撞 40071,
# 整脚本退出,而调用它的 E2E 只会报"演示数据没摆成",看不出是限流。
# 做成**显式开关**而不是每次自动清:限流是生产行为,不该被脚本顺手改掉。
if args.reset_gate_limit:
uids = [u for u, _, _, _ in STUDENTS]
for uid_ in uids:
_redis_del(f"xt:mp:gate:user:{uid_}")
print(f" 已清门禁限流计数:{uids}")
# ── 演示班主任:**必须显式指定初审人**,否则这一屏验不了 ──
#
# 初审的授权口径是"这个班的初审人 / 班级负责人"(`AuditScopeMapper`
# 的两级回退),而不是"有 audit:first 权限就行"。所以管理员自己**看不到初审待办** ——
# 这是设计(权限 ≠ 范围),但演示时必须有一个真实的班主任才能把初审那一屏走通。
teacher_id, teacher_name = ensure_teacher(token, org["id"])
st, body = R.call("PUT", f"{BASE}/api/v1/tasks/{task_id}/orgs", token, {
"orgs": [{
"orgId": org["id"],
"expectedCount": len(STUDENTS),
"headUserId": teacher_id
}]
})
if body.get("code") != 0:
raise SystemExit(f"指定初审人失败:{body}")
# ⚠️ PUT 返回的是 `ApiResult<Void>`(data 为空),**不能拿它回显校验** ——
# 必须 GET 回来确认真的落库了。否则"接口成功但字段没写进去"会被当成通过。
st, body = R.call("GET", f"{BASE}/api/v1/tasks/{task_id}/orgs", token)
got = ((body.get("data") or [{}])[0]).get("headUserId")
if got != teacher_id:
raise SystemExit(f"指定初审人没生效:期望 {teacher_id},实际 {got}")
print(f" 初审人:{teacher_name}(id={teacher_id})")
# 材料模板(身份证 / 学籍表 / 一寸照片 都在 TPL_SCHOOL_ROLL 里)
st, body = R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/materials/apply-template", token,
{"templateCode": "TPL_SCHOOL_ROLL"})
if body.get("code") != 0:
raise SystemExit(f"应用材料模板失败:{body}")
print(f" 材料项:{len(body.get('data') or [])} 项")
# ── ② 导名册(走真实导入接口)──────────────────────
#
# ⚠️ **必须带 orgId**:官方模板里没有班级列,不指定的话名册行没有班级,
# 于是填报单也没有班级 → 初审的范围校验无从判定,直接 `40301
# 这份填报单没有班级信息,无法确认审核范围`(这正是 e2e-m5-audit 里那条断言的口径)。
_build_roster([(name, idc) for _, name, idc, _ in STUDENTS])
body = R.upload_roster(BASE, token, task_id, TMP_ROSTER,
extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]})
if body.get("code") != 0:
raise SystemExit(f"导入名册失败:{body}")
rep = body.get("data") or {}
print(f" 名册导入:created={rep.get('created')} failed={rep.get('failed')}")
for e in (rep.get("errors") or [])[:5]:
print(f" ✗ 第 {e.get('row')} 行 {e.get('col')}:{e.get('reason')}")
# ── ③ 发布(这样学生端才允许提交)──────────────────
st, body = R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/publish", token, {"force": True})
if body.get("code") != 0:
raise SystemExit(f"发布失败:{body}")
print(f" 任务已发布(状态={(body.get('data') or {}).get('status')})")
# ── ④ 逐个学生:门禁核验 → 填表 → 传图 → 提交 ──────
cfg = R.read_dev_config()
subs: dict[int, int] = {}
for uid, name, idc, _ in STUDENTS:
tok = R.mint_mp_token(cfg["jwt_secret"], cfg["issuer"], uid, 1, ["STUDENT"], 4)
# 门禁核验:服务端按姓名 + 证件号匹配名册并**原子绑定**(真实链路)
#
# ⚠️ 门禁按 IP 限流(40071)。连续跑两次种子脚本(全量回归里它被多个 E2E
# 各调一次)就会撞上,而表现是"演示数据没摆成",完全看不出是限流。
# 所以只对 40071 退避重试;**其它错误照旧立刻退出**,不掩盖真问题。
body = {}
for attempt in range(3):
st, body = R.call("POST", f"{BASE}/api/v1/mp/roster/match", tok,
{"taskId": task_id, "realName": name, "idCardNo": idc})
if body.get("code") != 40071:
break
wait = 5 * (attempt + 1)
print(f" · 门禁限流(40071),{wait} 秒后重试第 {attempt + 2} 次…")
time.sleep(wait)
if body.get("code") != 0:
raise SystemExit(f"门禁核验失败({name}):{body}")
roster_id = (body.get("data") or {}).get("rosterId")
st, body = R.call("POST", f"{BASE}/api/v1/mp/submissions", tok, {"taskId": task_id})
if body.get("code") != 0:
raise SystemExit(f"建草稿失败({name}):{body}")
sid = body["data"]["id"]
fill = {"work_unit": "无", "is_retake": False, "is_student": True,
"school": "示范职业学校", "major": "数控加工", "enroll_date": "2024-09-01"}
res = A.prepare_student(tok, sid, task_id, fill, dup_first_file=(uid == 993102))
if res.get("_upload_error"):
raise SystemExit(f"上传失败({name}):{res['_upload_error']}")
d = res.get("data") or {}
print(f" {name}:红={d.get('redCount')} 黄={d.get('yellowCount')} "
f"submitted={d.get('submitted')} rosterId={roster_id}")
subs[uid] = sid
# ── ⑤ 丙:由**班主任**先初审通过,给终审节点留一份 ──
#
# ⚠️ 这一步不能用管理员的 token:初审的范围校验按"这个班归不归我"判定,
# 管理员既不带头衔也不在初审人名单里 → `40301`。
# 这正是"权限 ≠ 范围"在真实链路上的体现(`e2e-m5-audit.py` 专门守着它)。
st, body = R.login(BASE, TEACHER_USERNAME, TEACHER_PASSWORD, "demo")
if body.get("code") != 0:
raise SystemExit(f"班主任登录失败:{body}")
teacher_tok = body["data"]["token"]
st, body = R.call("POST", f"{BASE}/api/v1/audit/submissions/{subs[993103]}/approve",
teacher_tok,
{"remark": "演示数据:初审通过,留一份待终审", "actionSource": "BUTTON"})
print(f" 丙 初审 → code={body.get('code')} status={(body.get('data') or {}).get('status')}"
f" msg={body.get('msg')}")
# ⑥ 可选:把丙推到**已通过**,让导出功能有东西可导。
# 必须走真实终审链路(管理员 token),不能直接改状态 ——
# 否则"导出的数据真的审过"这件事就没法再靠这套夹具验了。
if args.approve_one:
st, body = R.call("POST", f"{BASE}/api/v1/audit/submissions/{subs[993103]}/approve",
token,
{"finalOpinion": "演示数据:终审通过", "actionSource": "BUTTON"})
code = body.get("code")
status = (body.get("data") or {}).get("status")
print(f" 丙 终审 → code={code} status={status} msg={body.get('msg')}")
if code != 0 or status != 3:
print(" ⚠️ 终审没有把丙推到「已通过」—— 导出演示会拿不到候选数据")
if os.path.exists(TMP_ROSTER):
os.remove(TMP_ROSTER)
print("\n 完成。浏览器里:")
print(" · 收集任务 → 【演示】审核台 → 名单管理:3 人")
print(" · 审核 → 初审:演示甲(干净)/ 演示乙(黄色警告)")
print(" · 审核 → 终审:演示丙(已初审,待终审)" if not args.approve_one
else " · 导出中心:演示丙已终审通过,可直接导出(另有 2 人未审完会被 40035 拦)")
return 0
def ensure_teacher(token: str, org_id: int) -> tuple[int, str]:
"""
准备一个演示班主任(已存在就复用),返回 (id, 姓名)。
为什么必须有一个**真实的班主任账号**:
初审的范围校验是"这个班归不归我"(指定初审人 → 班级负责人),
管理员自己**看不到初审待办** —— 这是设计,不是缺陷。
所以演示/验收初审那一屏时,必须有一个真账号登进去看。
密码用 `resetPassword(mode=CUSTOM)` 设成固定值:新增账号返回的是**一次性**
初始密码,而演示脚本下次跑要能直接登录。
"""
st, body = R.call("GET", f"{BASE}/api/v1/users?keyword={TEACHER_USERNAME}&pageNum=1&pageSize=5",
token)
for u in (body.get("data") or {}).get("list") or []:
if u.get("username") == TEACHER_USERNAME:
st2, b2 = R.call("POST", f"{BASE}/api/v1/users/{u['id']}/reset-password", token,
{"mode": "CUSTOM", "newPassword": TEACHER_PASSWORD})
print(f" 复用已存在的演示班主任 id={u['id']}(重置密码 code={b2.get('code')})")
return u["id"], u.get("realName") or TEACHER_NAME
st, body = R.call("POST", f"{BASE}/api/v1/users", token, {
"username": TEACHER_USERNAME,
"realName": TEACHER_NAME,
"phone": TEACHER_PHONE,
"orgId": org_id,
"roleCodes": ["HEAD_TEACHER"],
"userType": 3
})
if body.get("code") != 0:
raise SystemExit(f"建演示班主任失败:{body}")
uid = body["data"]["id"]
st2, b2 = R.call("POST", f"{BASE}/api/v1/users/{uid}/reset-password", token,
{"mode": "CUSTOM", "newPassword": TEACHER_PASSWORD})
if b2.get("code") != 0:
raise SystemExit(f"设置演示班主任密码失败:{b2}")
return uid, TEACHER_NAME
def _build_roster(rows: list[tuple[str, str]]) -> None:
"""演示名册:表头 + 报考三项按任务声明(车工 / 数控车工 / 四级)填好。"""
from openpyxl import Workbook
wb = Workbook()
ws = wb.active
ws.title = "演示班"
header = ["序号", "姓名", "身份证号", "学历", "联系电话", "工作单位",
"报考职业", "报考工种", "报考等级", "是否补考",
"补考理论", "补考实操", "补考综合", "是否在校生",
"在读学校", "所学专业", "入学时间"]
for c, t in enumerate(header, start=1):
ws.cell(row=1, column=c, value=t)
for i, (name, idc) in enumerate(rows):
vals = [i + 1, name, idc, "中专", f"138001399{i}", "无",
"车工", "数控车工", "四级", "否", "", "", "", "是",
"示范职业学校", "数控加工", "2024-09-01"]
for c, v in enumerate(vals, start=1):
if v != "":
ws.cell(row=i + 2, column=c, value=v)
wb.save(TMP_ROSTER)
if __name__ == "__main__":
raise SystemExit(main())