Files
xiaotutongxue/scripts/check-wiring.py
T
lovife 9e78e1d239 feat(notify): 通知接线检查 + 修 F-101~F-111(模板对自动通知生效 / 催报真的会发 / 该发的通知都发了)
起点:F-1/F-31/F-50/F-52 是同一个形状 —— 每处单看都对,编译/lint/review 全不红,只有从入口走一遍才暴露。先把这一类固化成命令,再按输出逐条修。

新增 scripts/check-wiring.py:权限码有没有地方校验、错误码有没有被引用、站内信模板有没有发送点、bizType 有没有被用过、菜单键 ↔ PC 端 XtMenu 双向核对。有意不接线的写进脚本上方例外表(带理由)。进 verify-all(第 5 项)与 CI 的 check 阶段。检查器自身修了三次假阳性:只数 @RequirePerm 漏掉服务层校验、按标识符数引用被同名常量骗过、注释里的提及被当成接线(改为先去 Java 注释再数)。

F-101 自动通知改走可配置模板(FR-1002 原来只兑现了手工发送那一半):新增 NotifyTextResolver(单向依赖避开循环注入),库里改过的 > 内置;模板停用 = 这类通知不发且如实回 notifySent=false。F-102 加入申请三处通知(提交→审批人、通过/驳回→学生)。F-103 导出完成/失败通知发起人,正文不放签名 URL(模板里原本写着 {下载链接},一并改掉)。F-104 自动催报从来没发出过任何一条(end_time = date::date 与库里的 23:59:59 永不相等,测试夹具用 SQL 摆的日期型恰好能匹配)+ 补 FR-1003 的「截止当日」。F-105 驳回达上限通知管理员(契约与注释都写了,代码只有一行日志)。F-106 字典/通知记录两个菜单没有任何角色持有(手册还写着去那里办)。F-107 6 个权限码挂到自己的接口。F-109 Map.of 不接受 null → 通知被吞成一行 warn。F-110/F-111 测试自身的两个坑。

验证:新增 e2e-m5-notify-wiring.py 43 条(改模板→真驳回→断言学生收到改过的那句、停用→一条不发、申请/审批/导出/催报/升级/菜单/权限各一节);verify-all --with-e2e → 9 项全过 0 跳过,38 个脚本 / 1620 条断言 / 0 失败。
2026-10-08 14:46:11 +08:00

446 lines
22 KiB
Python
Raw 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 -*-
"""接线检查:找「定义好了、代码里也写了,但**从来没有被接上**」的东西。
## 为什么需要它(这一族错在本仓库反复出现)
盘点这个项目时抓到的 P0/P1 问题里,有一大类长着同一个样子:
| 例子 | 现象 |
|---|---|
| F-1 | `OperationLogMapper` 存在,但**全仓库只有它自己 import** → 敏感操作一个字都没留痕 |
| F-31 | `LoginLogMapper` 在写,但**微信登录那条链路不写** → 合规上查不到学生登录 |
| F-50 | `collect_validation_issue` 表 V1 就建好了,**一直没有代码用它** |
| F-24/35 | 实体要的审计列,建表时漏了 → 插入报错(四次) |
| F-52/67 | 契约里定义了接口,但没有"把链路接上"的那一步,功能等于装饰 |
它们的共同点:**每一处单看都是对的**(类在、表在、常量在、注释写得很清楚),
所以 code review、typecheck、编译、单元测试**全都不会红** —— 只有"从入口走一遍"
才会暴露。而人不会每次都走一遍全链路。
所以把它固化成命令:**列出"声明了但没有任何地方引用"的那些声明**,
逐条由人判断是"忘记接线"还是"有意留的"(有意留的要写进下面的已知例外表,
而不是靠下次再想一遍)。
## 查五件事
| # | 查什么 | 没接上的后果 |
|:--:|---|---|
| 1 | `PermissionDef` 的权限码没有任何 `@RequirePerm` 用它 | 权限码是死的(或接口忘了标权限) |
| 2 | `PermissionDef` 的权限码没有被任何内置角色分配 | **这个接口谁都调不了**(只有平台租户能绕过) |
| 3 | `ErrorCode` 的码零引用 | 文档里有、代码里没有对应分支 |
| 4 | `NotifyTemplates` 的站内信模板零发送点 | 模板页能改,但那条通知根本不发 / 不走模板 |
| 5 | `CollectMessage.BIZ_*` 零使用 | 消息中心那个分类永远是空的 |
| 6 | 内置角色声明的菜单键 ↔ PC 端 `XtMenu` 的 key | 菜单永远不显示(键拼错),或前端有菜单谁都看不到 |
## 它**不查**什么(免得被当成"全绿就没事")
- 不查"逻辑对不对":它只回答"有没有被引用过",一个被引用了但用错地方的常量
它照样算通过;
- 不查前端有没有把权限码用上(`v-permission` 之类)—— 那要跨语言解析模板;
- 不查服务层"备好了没人调"的公开方法(如本轮发现的
`NotifyTemplateService.templateOf`)—— 这类误报太多,靠人去读;
所以第 4 项查的是"**模板码有没有发送点**",那是它能可靠判定的那一层。
跑法:`python scripts/check-wiring.py`(`--quiet` 只打摘要,给 `verify-all.py` 用)
退出码:0 = 没有未接线的声明;1 = 有(每条都打出来)。
"""
import argparse
import glob
import io
import os
import re
import sys
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
os.chdir(ROOT)
PERM_DEF = 'backend/xt-server/src/main/java/cn/tianbaotong/fill/security/PermissionDef.java'
BUILTIN_ROLE = 'backend/xt-server/src/main/java/cn/tianbaotong/fill/security/BuiltinRole.java'
ERROR_CODE = 'backend/xt-common/src/main/java/cn/tianbaotong/fill/common/api/ErrorCode.java'
NOTIFY_TPL = 'backend/xt-server/src/main/java/cn/tianbaotong/fill/module/notify/service/NotifyTemplates.java'
MESSAGE_ENTITY = ('backend/xt-server/src/main/java/cn/tianbaotong/fill/module/notify/'
'entity/CollectMessage.java')
XT_ROUTES = 'frontend-pc/src/config/xt-routes.ts'
# ── 已知例外:**有意不接线的**,写在这里(连同理由),不要靠下次再想一遍 ──
# 权限码:没有任何地方校验它(既没有 @RequirePerm,也没有服务层的 permissionChecker)
PERM_NO_CHECK = {
'ROLE_MANAGE': '「角色与授权」页是二期功能(权限码先定下来,见 08 文档 6n):'
'一期不开放自定义角色,所以没有任何接口需要它',
'ROSTER_MANAGE_OWN': '「本班名册」不是独立接口,而是 roster 接口里的**数据范围**分支'
'(服务层按角色判 orgIds),接口级权限用 ROSTER_MANAGE / ROSTER_VIEW',
'PROGRESS_VIEW_OWN': '同上:进度看板的"本班/本人"是同一接口的范围分支(见 ProgressService)',
'SUBMISSION_OWN': '学生端接口靠**登录态 + 只能操作本人那一份**保证,'
'不走权限码(学生是 MP 端唯一角色,加一道权限码只会多一个出错点)',
'EXPORT_UNMASKED': '脱敏导出开关 `desensitize` 是**显式保留到二期**的(名册里本来就是'
'脱敏值,见 ExportDtos 的注释),所以这个码暂时没有对应接口',
}
# 权限码:被校验了,但没有被任何内置角色分配(= 谁都调不了)
PERM_NO_ROLE = {}
ERROR_CODE_UNUSED = {
'ID_CARD_CHECKSUM_FAILED': 'M4 识别与校验(等 OCR 选型,见 docs/01 D1)',
'ID_CARD_MISMATCH_ROSTER': '同上(M4)',
'SUSPECTED_WRONG_OWNER': '同上(M4,串档检测)',
'FILE_OWNERSHIP_DOUBT': '同上(M4,VAL-21 归属校验)',
'OCR_FAILED': '同上(M4,OCR 服务异常)',
'IMAGE_TOO_LARGE_CIRCUIT_BREAK': '同上(M4,图片熔断)',
'VLM_FAILED': '同上(M4,VLM 识别)',
'FILE_PROCESS_FAILED': '内部错误:统一由 GlobalExceptionHandler 兜成 50000 返回,'
'细分码留在文档里给排查用(客户端拿到的仍是"系统繁忙")',
'NOTIFY_FAILED': '同上(通知失败不影响主流程,只记日志)',
'STORAGE_FAILED': '同上(存储失败由 FileProcessService 转成 50001 口径的业务提示)',
'SIGNED_URL_EXPIRED': '签名链接过期由**存储层**返回(七牛 401 → StorageService 转成'
'可读提示),没有单独抛这个码;契约 §12.6 的语义已由提示覆盖',
'IDEMPOTENT_REPLAY': '契约 §9.4 的幂等键回放:一期用"重复提交返回既有结果"实现,'
'没有单独回这个码(要回它得先有幂等键,属二期)',
'REJECT_LIMIT_EXCEEDED': '驳回达上限**不是错误**:驳回本身成功了,'
'只是把 escalated 标出来(契约 §11.4「超限升级管理员」),'
'所以走的是响应字段而不是错误码 —— 见 docs/08 F-105',
}
NOTIFY_TPL_UNUSED = {
'TASK_DEADLINE': '临期提醒的自动发送走「自动催报」(BIZ 5 + URGE / URGE_DRAFT 模板),'
'所以这个码只在模板页里可编辑;要单独发"临期提醒"得先有"未绑定学生的账号",'
'见 docs/08 F-108 的记录',
'TASK_PUBLISH': '「发布即通知学生」在本系统的流程里**没有收件人**:'
'学生要先扫码核验才会绑定账号,而发布之前没有人会去扫 —— '
'所以这条通知发出去只会是 0 条(见 docs/08 F-108)',
}
BIZ_UNUSED = {
'BIZ_TASK_PUBLISH': '同上:bizType 1(任务发布)与 TASK_PUBLISH 模板同因,'
'在"发布时刻没有可发的收件人"这件事解决之前不发',
}
def read(path):
return io.open(path, encoding='utf-8').read()
def strip_java_comments(src):
"""把注释替换成空白(**保留换行**,行号才不会错位)。
⚠️ 必须先去注释再数引用:本仓库的注释极其啰嗦,而且**专门在解释"为什么不这么做"** ——
例如 `ExportController` 的类注释里写着"权限只有两个:export:download 与 export:unmasked",
而代码里从来没校验过 `EXPORT_UNMASKED`。不剥注释的话,这个检查会被自己的文档满足
(假阴性),而那正是《08》F-36 踩过的坑:**静态检查要能区分"代码"与"解释代码的文字"**。
"""
out = []
i, n = 0, len(src)
state = None # None / 'line' / 'block' / 'str' / 'char'
while i < n:
c = src[i]
nxt = src[i + 1] if i + 1 < n else ''
if state is None:
if c == '/' and nxt == '/':
state = 'line'
out.append(' ')
i += 2
continue
if c == '/' and nxt == '*':
state = 'block'
out.append(' ')
i += 2
continue
if c == '"':
state = 'str'
elif c == "'":
state = 'char'
out.append(c)
i += 1
continue
if state == 'line':
if c == '\n':
state = None
out.append('\n')
else:
out.append(' ')
i += 1
continue
if state == 'block':
if c == '*' and nxt == '/':
state = None
out.append(' ')
i += 2
continue
out.append('\n' if c == '\n' else ' ')
i += 1
continue
# 字符串 / 字符字面量里:原样保留(模板码就是字符串),只处理转义与结束符
if c == '\\':
out.append(src[i:i + 2])
i += 2
continue
if (state == 'str' and c == '"') or (state == 'char' and c == "'"):
state = None
out.append(c)
i += 1
return ''.join(out)
def java_sources():
out = {}
for p in glob.glob('backend/**/*.java', recursive=True):
try:
out[p] = strip_java_comments(read(p))
except Exception:
pass
return out
def refs(sources, name, skip_files):
"""名字作为**整个标识符**(不是子串)在其它文件里出现过几次。"""
pat = re.compile(r'\b' + re.escape(name) + r'\b')
hits = []
for p, src in sources.items():
if os.path.abspath(p) in skip_files:
continue
for i, line in enumerate(src.split('\n'), 1):
if pat.search(line):
hits.append((p, i, line.strip()))
return hits
def set_of_blocks(src):
"""把每个 `Set.of( ... )` 的参数原文取出来(括号配对,能吃下注释与换行)。
⚠️ 第一版用正则 `Set\\.of\\(\\s*((?:"[^"]*"\\s*,?\\s*)+)\\)` 匹配 ——
往菜单集合里加一行 `//` 注释之后它就**静默不匹配了**,于是 8 个菜单键
全被报成"没有任何角色声明"(假阳性)。静态检查里凡是"匹配不到就当没有"的写法,
都要先问一句"匹配不到时会发生什么"。
"""
out = []
for m in re.finditer(r'Set\.of\(', src):
i = m.end()
depth = 1
while i < len(src) and depth > 0:
if src[i] == '(':
depth += 1
elif src[i] == ')':
depth -= 1
i += 1
out.append(src[m.end():i - 1])
return out
def strip_line_comments(block):
return '\n'.join(re.sub(r'//.*$', '', ln) for ln in block.split('\n'))
def template_send_points(sources, code, tpl_file):
"""站内信模板的"发送点" = 代码里有人把这个码当成模板码用。
⚠️ 判据不能用"标识符出现次数":`JOIN_APPROVE` 同时是
`OperationLogService.ACTION_JOIN_APPROVE`(操作日志动作码)与
`NotificationService.TPL_JOIN_APPROVE`(微信订阅消息模板码)的值 ——
按标识符数会得出"3 处引用",而它当时**一个发送点都没有**。
所以认两件事:
① 字面量 `"CODE"` 出现在非"常量声明"的行上;
② 以 `"CODE"` 为值的常量被**在通知调用的上下文里**引用
(±6 行窗口里出现 `ofTemplate` / `NotifyRequest` / `notify(` / `subjectCode`)——
因为发送点常常写成 `SUBJECT_URGE` 这种具名常量,字面量根本不出现。
②是**启发式**:它可能被别的写法骗过(窗口里恰好有这些词)。它是烟雾报警器,不是证明。
"""
hits = []
aliases = {}
for p, s in sources.items():
for line in s.split('\n'):
m = re.search(r'static final String\s+(\w+)\s*=\s*"' + re.escape(code) + r'"', line)
if m:
aliases[m.group(1)] = p
lit = re.compile(r'"' + re.escape(code) + r'"')
for p, s in sources.items():
if os.path.abspath(p) == os.path.abspath(tpl_file):
continue
for i, line in enumerate(s.split('\n'), 1):
if lit.search(line) and not re.search(r'static final String\s+\w+\s*=', line):
hits.append((p, i, line.strip()))
ctx = re.compile(r'ofTemplate|NotifyRequest|notify\(|subjectCode')
for name in aliases:
pat = re.compile(r'\b' + name + r'\b')
for p, s in sources.items():
lines = s.split('\n')
for i, line in enumerate(lines, 1):
if not pat.search(line) or re.search(r'static final String\s+' + name + r'\b', line):
continue
window = '\n'.join(lines[max(0, i - 7):i + 5])
if ctx.search(window):
hits.append((p, i, line.strip()))
return hits
def menu_exception(key):
"""菜单键的已知例外(不是 PC 端菜单,或由基础路由提供)。"""
if key == 'dashboard':
return 'PC 工作台是**基础路由**(不在 XtMenu 里,所有登录用户都能看到)'
if key.startswith('mp-'):
return '这是**小程序**的 tabBar 键(学生端菜单),PC 端当然没有'
return None
def main():
ap = argparse.ArgumentParser(description='接线检查')
ap.add_argument('--quiet', action='store_true')
args = ap.parse_args()
def say(*a):
if not args.quiet:
print(*a)
src = java_sources()
problems = [] # (类别, 名字, 说明)
notes = [] # 已知例外,照样打出来(不能静默)
say('=' * 88)
say(' 接线检查 「声明了、但没有任何地方引用它」')
say('=' * 88)
# ── ① 权限码:有没有地方校验它 ────────────────────────────
perm_src = read(PERM_DEF)
perms = re.findall(r'^\s{4}([A-Z][A-Z0-9_]*)\(', perm_src, re.M)
# @RequirePerm 的用法(可能跨行、可能 `{A, B}` 多条)
require_perm = set()
for s in src.values():
for block in re.findall(r'@RequirePerm\s*\(([^)]*)\)', s, re.S):
require_perm.update(re.findall(r'PermissionDef\.([A-Z][A-Z0-9_]*)', block))
role_perm = set(re.findall(r'PermissionDef\.([A-Z][A-Z0-9_]*)', read(BUILTIN_ROLE)))
# 服务层校验(permissionChecker.has / require):**同样是校验点** ——
# ⚠️ 第一版只数 @RequirePerm,于是把 AUDIT_WITHDRAW 报成"没有接口用它",
# 而它明明在 AuditService 里用 permissionChecker.require 守了两处(F-36 的教训:
# 静态检查必须先确认"它数的那件事"就是"要判定的那件事")。
service_perm = set()
skip_perm_files = {os.path.abspath(PERM_DEF), os.path.abspath(BUILTIN_ROLE)}
for p, s in src.items():
if os.path.abspath(p) in skip_perm_files:
continue
service_perm.update(re.findall(r'PermissionDef\.([A-Z][A-Z0-9_]*)', s))
say('\n【①】权限码 %d 个(PermissionDef)' % len(perms))
for name in perms:
checked = name in service_perm or name in require_perm
in_role = name in role_perm
if checked and in_role:
where = '接口注解' if name in require_perm else '服务层'
say(' ✓ %-22s %s' % (name, where))
continue
# 有例外登记的照实打出来(**不能打 ✓**:那会让"未接线"看起来像已接线)
why = []
if not checked:
why.append(PERM_NO_CHECK.get(name) or '没有任何地方校验它(@RequirePerm 与服务层都没有)')
if not in_role:
why.append(PERM_NO_ROLE.get(name) or '没有被任何内置角色分配(这个接口谁都调不了)')
if (not checked and name in PERM_NO_CHECK) or (not in_role and name in PERM_NO_ROLE):
notes.append('权限码 %s —— 已知例外:%s' % (name, ';'.join(why)))
say(' — %-22s 已知例外:%s' % (name, ';'.join(why)[:60]))
else:
problems.append(('权限码', name, ';'.join(why)))
say(' ✗ %-22s %s' % (name, ';'.join(why)))
# ── ② 错误码:有没有被引用 ────────────────────────────────
ec_src = read(ERROR_CODE)
codes = re.findall(r'^\s{4}([A-Z][A-Z0-9_]*)\((\d+),', ec_src, re.M)
skip = {os.path.abspath(ERROR_CODE)}
say('\n【②】错误码 %d 个(ErrorCode)' % len(codes))
unused_codes = []
for name, num in codes:
if name == 'SUCCESS':
continue
if not refs(src, 'ErrorCode.' + name, skip):
unused_codes.append((name, num))
for name, num in unused_codes:
if name in ERROR_CODE_UNUSED:
notes.append('错误码 %s(%s) 零引用 —— 已知例外:%s' % (name, num, ERROR_CODE_UNUSED[name]))
else:
problems.append(('错误码', '%s(%s)' % (name, num), '全仓库零引用(除定义处)'))
say(' 零引用的码:%s' % ('、'.join('%s(%s)' % (n, c) for n, c in unused_codes) or '无'))
# ── ③ 站内信模板:有没有发送点 ────────────────────────────
tpl_src = read(NOTIFY_TPL)
tpls = re.findall(r'new Def\("([A-Z_]+)",\s*(\d+)', tpl_src)
say('\n【③】站内信模板 %d 个(NotifyTemplates)' % len(tpls))
for code, biz in tpls:
hits = template_send_points(src, code, NOTIFY_TPL)
if hits:
say(' ✓ %-14s biz=%s %d 处引用(%s)'
% (code, biz, len(hits), os.path.basename(hits[0][0])))
elif code in NOTIFY_TPL_UNUSED:
notes.append('站内信模板 %s 零发送点 —— 已知例外:%s' % (code, NOTIFY_TPL_UNUSED[code]))
say(' — %-14s biz=%s 零发送点(已知例外)' % (code, biz))
else:
problems.append(('站内信模板', code, '没有任何发送点(模板页能改,但这条通知不发)'))
say(' ✗ %-14s biz=%s **零发送点**' % (code, biz))
# 撞名:同一个字面量被两个命名空间当成"自己的码"
say('\n【③b】模板码撞名(不同命名空间、同一个字符串)')
for code, biz in tpls:
owners = []
for p, s in src.items():
for m in re.finditer(r'static final String\s+([A-Z_]+)\s*=\s*"%s"' % re.escape(code), s):
owners.append(m.group(1))
if len(owners) > 1:
say(' ! %-14s 同时是 %s 的值 —— 不是 bug,但下一个改代码的人会当成同一个东西'
% (code, '、'.join(sorted(set(owners)))))
# ── ④ 消息中心 bizType:有没有被用过 ──────────────────────
msg_src = read(MESSAGE_ENTITY)
bizs = re.findall(r'public static final int (BIZ_[A-Z_]+) = (\d+);', msg_src)
skip_msg = {os.path.abspath(MESSAGE_ENTITY)}
say('\n【④】消息中心 bizType %d 个(CollectMessage)' % len(bizs))
for name, num in bizs:
hits = refs(src, name, skip_msg)
if hits:
say(' ✓ %-18s = %s %d 处引用' % (name, num, len(hits)))
elif name in BIZ_UNUSED:
notes.append('bizType %s(%s) 零使用 —— 已知例外:%s' % (name, num, BIZ_UNUSED[name]))
else:
problems.append(('bizType', '%s(%s)' % (name, num),
'零使用(消息中心这个分类永远是空的)'))
say(' ✗ %-18s = %s **零使用**' % (name, num))
# ── ⑤ 内置角色菜单键 ↔ PC 端菜单 ──────────────────────────
role_src = read(BUILTIN_ROLE)
menu_keys = set()
for block in set_of_blocks(role_src):
cleaned = strip_line_comments(block)
# 权限集合里是 `PermissionDef.X`(没有引号)→ 只有"全是字符串字面量"的才是菜单
if re.search(r'PermissionDef\.', cleaned):
continue
menu_keys.update(re.findall(r'"([^"]+)"', cleaned))
pc_keys = set(re.findall(r"key:\s*'([a-zA-Z-]+)'", read(XT_ROUTES)))
say('\n【⑤】菜单键:内置角色声明 %d 个 / PC 端 XtMenu %d 个'
% (len(menu_keys), len(pc_keys)))
for k in sorted(menu_keys - pc_keys):
ex = menu_exception(k)
if ex:
notes.append('菜单键 %s —— 已知例外:%s' % (k, ex))
say(' — %-14s %s' % (k, ex))
else:
problems.append(('菜单键', k, '内置角色声明了它,但 PC 端 XtMenu 里没有这个 key'))
say(' ✗ %-14s 角色声明了,PC 端没有这个 key' % k)
for k in sorted(pc_keys - menu_keys):
problems.append(('菜单键', k, 'PC 端有菜单,但没有任何内置角色声明它(谁都看不到)'))
say(' ✗ %-14s PC 端有,但没有任何角色能看到' % k)
for k in sorted(menu_keys & pc_keys):
say(' ✓ %s' % k)
# ── 汇总 ────────────────────────────────────────────────
print('\n' + '=' * 88)
if notes:
print(' 已知例外 %d 条(有意不接线,写在本脚本上方的表里):' % len(notes))
for n in notes:
print(' — %s' % n)
print(' 未接线的声明 %d 条:' % len(problems))
for kind, name, why in problems:
print(' ✗ [%s] %s —— %s' % (kind, name, why))
if not problems:
print(' (无)')
print('=' * 88)
sys.stdout.flush()
return 1 if problems else 0
if __name__ == '__main__':
raise SystemExit(main())