feat(m0): 后端骨架 + 账号体系(M0 验收 36/36 通过)

技术栈(已在 4.1.1 上实测跑通,非纸面选型)
  Java 21 + Spring Boot 4.1.1 + MyBatis-Plus 3.5.17(boot4) + PostgreSQL 16.15
  + Redis 7 + Flyway 12.4 + springdoc 3.1.1;前端定 Vue 3 + Element Plus + uni-app

交付内容
  M0-01 基础设施     deploy/docker-compose.yml(PG + Redis + MinIO,容器名与端口避让本机既有项目)
  M0-02 后端骨架     xt-common + xt-server;统一响应 / 全局异常 / 58 个错误码 /
                     多租户拦截 / 软删除 / 字段自动填充 / AES-GCM / HMAC 索引 / RSA / JWT
  M0-05 租户        开通租户(一次完成建租户 + 发 6 个内置角色 + 建首个管理员)、列表、详情
  M0-06 组织        CRUD + 组织树 + ancestor_path 自动维护 + 占用校验
  M0-07 PC 登录     RSA 解密 + 图形验证码 + 失败锁定 + 等时比较防账号枚举 + 审计日志
  M0-08 登出/当前用户  Token 黑名单(存 SHA-256,TTL 到自然过期)
  M0-01 数据库      Flyway 2 个迁移 → 26 张表(20 业务 + 5 标准 + 1 迁移记录)

安全姿态
  · 租户拦截器【默认拒绝】:匿名请求触及租户表直接抛 40101,不做「不加条件放行」
  · 权限码定义在代码里(PermissionDef / BuiltinRole),不建 sys_permission 表,杜绝前后端漂移
  · 匿名接口只靠 @PublicApi 放行,不做路径白名单,避免两套机制各说各话
  · 账号不存在时也跑一次 BCrypt(假哈希),拉平响应时间防枚举
  · 登录密码 RSA 传输,防止网关/APM 把明文密码记进日志

脚本
  scripts/sync-ddl.py          文档 DDL → Flyway 迁移,带 --check 防漂移
  scripts/e2e-m0.py            M0 验收 36 项,可进 CI
  scripts/verify-docs.py       文档一致性(接口数【逐节】核对、错误码、跨文档数字、表格列数)
  scripts/dev-login.py         RSA 加密登录助手
  scripts/dev-clean-e2e.py     清理 E2E 测试数据

文档修正(实测暴露,非人工审查)
  · DDL 缺陷:sys_tenant.contact_phone / sys_user.phone / collect_roster.phone
    原 varchar(20) 按明文长度建,装不下 AES-GCM 密文(约 52 字符),改 varchar(128)(PRD V2.4)
  · 错误码 40007b/40007c 带字母后缀与「code 为数字」的约定冲突,规范化为 40071/40072
  · 07-开发实施计划 新增 §9 版本兼容踩坑记录(10 条实录)与 §10 M0 完成情况
This commit is contained in:
2026-09-28 17:28:21 +08:00
parent 63aec55329
commit 9d7a8ca845
88 changed files with 7564 additions and 32 deletions
+99
View File
@@ -0,0 +1,99 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
清理 e2e-m0.py 产生的测试数据。
E2E 每跑一次就新建一个租户(编码 e2e######)并给它建角色与账号,
反复跑会把开发库堆满。这个脚本把 e2e* 租户及其关联数据清掉。
用法
----
python scripts/dev-clean-e2e.py # 只看会删什么(默认 dry-run)
python scripts/dev-clean-e2e.py --apply # 真删
⚠️ 只在开发库上跑,不要对生产执行。
"""
from __future__ import annotations
import argparse
import subprocess
import sys
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
# 子表在前、主表在后(无外键约束,顺序只为可读性与排查方便)
STEPS = [
("角色授权", "sys_role_permission"),
("用户角色", "sys_user_role"),
("登录日志", "sys_login_log"),
("组织", "sys_org"),
("角色", "sys_role"),
("用户", "sys_user"),
]
TENANT_FILTER = "tenant_code LIKE 'e2e%'"
SUB_FILTER = f"tenant_id IN (SELECT id FROM sys_tenant WHERE {TENANT_FILTER})"
class Db:
def __init__(self, container: str, user: str, name: str):
self.container = container
self.user = user
self.name = name
def scalar_rows(self, sql: str) -> list[str]:
cmd = ["docker", "exec", self.container, "psql", "-U", self.user, "-d", self.name,
"-v", "ON_ERROR_STOP=1", "-t", "-A", "-F", "|", "-c", sql]
r = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8")
if r.returncode != 0:
print(r.stderr.strip(), file=sys.stderr)
sys.exit(1)
return [x for x in r.stdout.split("\n") if x.strip()]
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--apply", action="store_true", help="真正执行删除(默认只预览)")
ap.add_argument("--container", default="xt-pg")
ap.add_argument("--user", default="fill")
ap.add_argument("--db", default="filldb")
args = ap.parse_args()
db = Db(args.container, args.user, args.db)
tenants = db.scalar_rows(
f"SELECT id || '|' || tenant_code || '|' || tenant_name "
f"FROM sys_tenant WHERE {TENANT_FILTER} ORDER BY id")
if not tenants:
print("没有 e2e 测试租户,无需清理")
return 0
print(f"找到 {len(tenants)} 个 e2e 测试租户:")
for r in tenants[:15]:
p = r.split("|")
print(" id=%-5s code=%-14s %s" % (p[0], p[1], p[2] if len(p) > 2 else ""))
if len(tenants) > 15:
print(f" ... 另有 {len(tenants) - 15} 个")
if not args.apply:
print("\n这是预览。加 --apply 才会真正删除。")
return 0
print("\n开始清理:")
total = 0
for label, table in STEPS:
rows = db.scalar_rows(f"DELETE FROM {table} WHERE {SUB_FILTER} RETURNING id")
total += len(rows)
print(" %-8s %-20s 删除 %d 行" % (label, table, len(rows)))
rows = db.scalar_rows(f"DELETE FROM sys_tenant WHERE {TENANT_FILTER} RETURNING id")
total += len(rows)
print(" %-8s %-20s 删除 %d 行" % ("租户", "sys_tenant", len(rows)))
print(f"\n完成,共删除 {total} 行。")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+106
View File
@@ -0,0 +1,106 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
开发辅助:RSA 加密登录并打印 token。
为什么需要它
------------
登录接口的密码字段是 RSA 加密的(防止明文密码进网关日志),
所以用 curl / Postman 直接调会很别扭——得先算出一段密文。
这个脚本把它变成一行命令。
用法
----
python scripts/dev-login.py # 默认 admin / Admin@123
python scripts/dev-login.py platform Platform@123
python scripts/dev-login.py admin Admin@123 --base http://localhost:8080
依赖:cryptography (pip install cryptography)
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import sys
import urllib.request
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_PUB_KEY = os.path.join(ROOT, "deploy", "rsa-public-key-dev.pem")
def rsa_encrypt(plain: str, pub_key_path: str) -> str:
try:
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
except ImportError:
sys.exit("缺少依赖:pip install cryptography")
with open(pub_key_path, "rb") as f:
public_key = serialization.load_pem_public_key(f.read())
# 必须与服务端 RsaCipher 的 "RSA/ECB/PKCS1Padding" 一致。
# 换成 OAEP 会表现为「密码永远错」,两边都改才能换。
cipher = public_key.encrypt(plain.encode("utf-8"), padding.PKCS1v15())
return base64.b64encode(cipher).decode("ascii")
def post_json(url: str, payload: dict, timeout: int = 20) -> tuple[int, dict]:
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
url, data=data,
headers={"Content-Type": "application/json; charset=utf-8"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8")
try:
return e.code, json.loads(body)
except json.JSONDecodeError:
return e.code, {"raw": body}
def main() -> int:
ap = argparse.ArgumentParser(description="RSA 加密登录,打印 token")
ap.add_argument("username", nargs="?", default="admin")
ap.add_argument("password", nargs="?", default="Admin@123")
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
ap.add_argument("--tenant-code", default=None, help="账号在多个租户下重名时必填")
ap.add_argument("--pub-key", default=DEFAULT_PUB_KEY)
ap.add_argument("--json", action="store_true", help="打印完整响应而非仅 token")
args = ap.parse_args()
if not os.path.exists(args.pub_key):
sys.exit(f"找不到公钥:{args.pub_key}")
payload = {
"username": args.username,
"password": rsa_encrypt(args.password, args.pub_key),
}
if args.tenant_code:
payload["tenantCode"] = args.tenant_code
status, body = post_json(f"{args.base}/api/v1/auth/login", payload)
if args.json:
print(json.dumps(body, ensure_ascii=False, indent=2))
elif body.get("code") == 0:
d = body["data"]
print(d["token"])
else:
print(f"[失败] HTTP {status} code={body.get('code')} msg={body.get('msg')}",
file=sys.stderr)
return 1
return 0 if body.get("code") == 0 else 1
if __name__ == "__main__":
raise SystemExit(main())
+328
View File
@@ -0,0 +1,328 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
M0 验收测试。
把《05-开发任务拆解》里 M0 的验收标准变成可重复执行的程序:
[x] 能开通租户 → 建组织树 → 各角色登录看到不同菜单
[x] 跨租户访问被拦截
[x] schema-postgresql.sql 在测试环境零错误执行 ← 由 Flyway 迁移保证
用法
----
python scripts/e2e-m0.py
python scripts/e2e-m0.py --base http://localhost:8080
退出码:0 = 全部通过;1 = 有失败项(可直接用于 CI)。
依赖:cryptography
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import sys
import time
import urllib.error
import urllib.request
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PUB_KEY = os.path.join(ROOT, "deploy", "rsa-public-key-dev.pem")
PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []
# ── HTTP ──────────────────────────────────────────────
def http(method: str, url: str, body=None, token: str | None = None,
timeout: int = 30) -> tuple[int, dict]:
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
headers = {"Content-Type": "application/json; charset=utf-8"}
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8")
try:
return e.code, json.loads(raw)
except json.JSONDecodeError:
return e.code, {"code": -1, "msg": raw[:200]}
def rsa_encrypt(plain: str) -> str:
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
with open(PUB_KEY, "rb") as f:
pub = serialization.load_pem_public_key(f.read())
return base64.b64encode(pub.encrypt(plain.encode("utf-8"), padding.PKCS1v15())).decode()
def login(base: str, username: str, password: str, tenant_code: str | None = None) -> dict:
payload = {"username": username, "password": rsa_encrypt(password)}
if tenant_code:
payload["tenantCode"] = tenant_code
try:
_, body = http("POST", f"{base}/api/v1/auth/login", payload)
except Exception as e: # 网络层失败也要报成一条用例,而不是崩掉
return {"code": -1, "msg": f"请求失败: {e}"}
return body or {"code": -1, "msg": "空响应"}
# ── 断言 ──────────────────────────────────────────────
def data_of(body: dict) -> dict:
"""安全取 data:响应里 data 可能是 null。"""
if not isinstance(body, dict):
return {}
d = body.get("data")
return d if isinstance(d, dict) else {}
def list_of(body: dict) -> list:
d = body.get("data")
if isinstance(d, list):
return d
if isinstance(d, dict) and isinstance(d.get("list"), list):
return d["list"]
return []
def check(name: str, cond: bool, detail: str = "") -> bool:
if cond:
PASSED.append(name)
print(f" [PASS] {name}")
else:
FAILED.append((name, detail))
print(f" [FAIL] {name} {detail}")
return cond
def expect_code(name: str, body: dict, code: int) -> bool:
actual = body.get("code") if isinstance(body, dict) else None
return check(name, actual == code,
f"期望 code={code},实际 code={actual} msg={body.get('msg') if isinstance(body, dict) else body}")
def finish() -> int:
print("\n" + "=" * 66)
total = len(PASSED) + len(FAILED)
if FAILED:
print(f" 结果: {len(PASSED)}/{total} 通过 —— 有失败项")
print("\n 失败项:")
for name, detail in FAILED:
print(f" - {name}")
if detail:
print(f" {detail}")
else:
print(f" 结果: {len(PASSED)}/{total} 全部通过")
print("=" * 66)
return 0 if not FAILED else 1
# ── 用例 ──────────────────────────────────────────────
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
args = ap.parse_args()
base = args.base.rstrip("/")
stamp = str(int(time.time()))[-6:]
new_code = f"e2e{stamp}"
if not os.path.exists(PUB_KEY):
sys.exit(f"找不到 RSA 公钥:{PUB_KEY}")
print("=" * 66)
print(" M0 验收测试")
print(f" 目标: {base}")
print("=" * 66)
# ── A. 认证 ───────────────────────────────────────
print("\n[A] 认证与鉴权")
body = login(base, "platform", "Platform@123")
if not expect_code("A1 平台运营登录成功", body, 0):
return finish()
platform_token = data_of(body).get("token")
# ⚠️ admin 这个账号名在多个租户下都存在(E2E 每次会给新租户也建一个 admin),
# 所以必须带 tenantCode —— 这正是服务端「拒绝猜测、要求指明单位」的设计。
body = login(base, "admin", "Admin@123", tenant_code="demo")
expect_code("A2 学校管理员登录成功(带租户码)", body, 0)
demo_token = data_of(body).get("token")
demo_menus = set(data_of(body).get("menus") or [])
# 覆盖「账号跨租户重名」这条保护。
# 首次运行时 admin 只存在于 demo 一个租户(此时正常登录是对的);
# 反复运行后会出现多个同名 admin(此时必须要求指明单位)。
# 两种都算通过,唯一不允许的是「随便挑一个」。
body = login(base, "admin", "Admin@123")
code_n = body.get("code")
msg_n = str(body.get("msg", ""))
check("A2b 同名账号跨租户时:要么唯一可直接登录,要么要求指明单位——绝不猜",
code_n == 0 or (code_n == 40012 and "多个单位" in msg_n),
f"code={code_n} msg={msg_n}")
expect_code("A3 密码错误返回 40012", login(base, "admin", "WrongPassword!1", "demo"), 40012)
# 用每次唯一的账号名,避免上一轮的失败计数触发验证码要求
ghost = f"ghost_{stamp}"
expect_code("A4 账号不存在也返回 40012(不泄露账号是否存在)",
login(base, ghost, "Whatever@123"), 40012)
_, b = http("GET", f"{base}/api/v1/auth/me")
expect_code("A5 未带 token 访问受保护接口返回 40101", b, 40101)
_, b = http("GET", f"{base}/api/v1/auth/me", token=demo_token)
expect_code("A6 /auth/me 正常返回", b, 0)
check("A7 返回角色码 SCHOOL_ADMIN", data_of(b).get("roleCodes") == ["SCHOOL_ADMIN"],
str(data_of(b).get("roleCodes")))
check("A8 学校管理员菜单不含 tenant(无平台权限)", "tenant" not in demo_menus,
f"menus={sorted(demo_menus)}")
body = login(base, "platform", "Platform@123")
plat_menus = set(data_of(body).get("menus") or [])
check("A9 平台运营菜单含 tenant", "tenant" in plat_menus, f"menus={sorted(plat_menus)}")
check("A10 两种角色菜单不同(权限渲染生效)", demo_menus != plat_menus,
f"demo={sorted(demo_menus)} platform={sorted(plat_menus)}")
# ── B. 开通租户 ───────────────────────────────────
print("\n[B] 开通租户")
_, b = http("GET", f"{base}/api/v1/tenants", token=demo_token)
expect_code("B1 学校管理员无租户管理权限(40301)", b, 40301)
_, b = http("POST", f"{base}/api/v1/tenants", {
"tenantName": f"E2E 测试学校 {stamp}",
"tenantCode": new_code,
"tenantType": 1,
"contactName": "测试联系人",
"contactPhone": "13900139000",
"expireTime": "2027-12-31",
"adminUsername": "admin",
"adminRealName": "E2E 管理员",
"adminPhone": "13900139001",
}, token=platform_token)
expect_code("B2 平台运营开通租户成功", b, 0)
created = data_of(b)
new_tenant_id = created.get("id")
new_pwd = created.get("adminInitialPassword")
check("B3 返回初始密码(仅此一次)", bool(new_pwd), json.dumps(created, ensure_ascii=False)[:200])
check("B4 返回小程序码参数", created.get("miniProgramScene") == f"t{new_tenant_id}",
str(created.get("miniProgramScene")))
if new_tenant_id is None:
return finish()
body = login(base, "admin", new_pwd or "x", tenant_code=new_code)
expect_code("B5 新租户管理员用初始密码可登录(证明租户作用域切换生效)", body, 0)
new_token = data_of(body).get("token")
check("B6 新管理员归属租户正确",
data_of(body).get("tenant", {}).get("tenantCode") == new_code,
str(data_of(body).get("tenant")))
_, b = http("GET", f"{base}/api/v1/tenants?pageNum=1&pageSize=50", token=platform_token)
expect_code("B7 租户列表可查", b, 0)
ids = [t.get("id") for t in list_of(b)]
check("B8 列表含新建租户", new_tenant_id in ids, f"ids={ids}")
_, b = http("POST", f"{base}/api/v1/tenants", {
"tenantName": "重复编码", "tenantCode": new_code, "tenantType": 1,
"contactName": "x", "contactPhone": "13900139000", "expireTime": "2027-12-31",
"adminUsername": "admin", "adminRealName": "x", "adminPhone": "13900139002",
}, token=platform_token)
expect_code("B9 租户编码重复被拒(40001)", b, 40001)
# ── C. 组织树 ─────────────────────────────────────
print("\n[C] 组织架构")
def mk_org(token, parent_id, name, otype, code):
_, r = http("POST", f"{base}/api/v1/orgs", {
"parentId": parent_id, "orgName": name, "orgType": otype, "orgCode": code,
}, token=token)
return r
b = mk_org(demo_token, None, f"E2E 示范学校 {stamp}", 1, f"ROOT{stamp}")
expect_code("C1 建学校节点", b, 0)
school_id = b.get("data")
b = mk_org(demo_token, school_id, f"E2E 机电系 {stamp}", 2, f"DEPT{stamp}")
expect_code("C2 建院系节点", b, 0)
dept_id = b.get("data")
b = mk_org(demo_token, dept_id, f"E2E 24数控1班 {stamp}", 3, f"CLS{stamp}")
expect_code("C3 建班级节点", b, 0)
class_id = b.get("data")
expect_code("C4 同级重复组织编码被拒(40001)",
mk_org(demo_token, school_id, "重复编码", 2, f"DEPT{stamp}"), 40001)
expect_code("C5 组织类型层级颠倒被拒(40001)",
mk_org(demo_token, class_id, "层级颠倒", 1, f"BAD{stamp}"), 40001)
_, b = http("GET", f"{base}/api/v1/orgs/tree?withCount=true", token=demo_token)
expect_code("C6 组织树查询", b, 0)
tree = list_of(b)
def find(nodes, target):
for n in nodes:
if n.get("id") == target:
return n
hit = find(n.get("children") or [], target)
if hit:
return hit
return None
check("C7 三层结构正确(班级挂在院系下)", find(tree, class_id) is not None,
f"树中未见 id={class_id}")
root = find(tree, school_id)
check("C8 学校 → 院系 层级正确",
root is not None and dept_id in [c.get("id") for c in (root.get("children") or [])],
f"children={[c.get('id') for c in (root.get('children') or [])] if root else None}")
_, b = http("DELETE", f"{base}/api/v1/orgs/{dept_id}", token=demo_token)
expect_code("C9 有下级的组织不可删除(40015)", b, 40015)
# ── D. 跨租户隔离 ─────────────────────────────────
print("\n[D] 跨租户隔离")
_, b = http("GET", f"{base}/api/v1/orgs/tree", token=new_token)
expect_code("D1 新租户可查自己的组织树", b, 0)
dumped = json.dumps(list_of(b), ensure_ascii=False)
check("D2 新租户看不到 demo 租户的组织(租户条件自动注入)",
str(school_id) not in dumped, f"新租户树里出现了 demo 的组织 id={school_id}")
_, b = http("GET", f"{base}/api/v1/tenants", token=new_token)
expect_code("D3 新租户管理员调租户列表被拒(40301)", b, 40301)
# ── E. 异常与安全 ─────────────────────────────────
print("\n[E] 异常与安全")
_, b = http("POST", f"{base}/api/v1/auth/login", {"username": "admin", "password": "not-encrypted"})
check("E1 密码未加密(非法密文)也被拒,不泄露内部信息",
b.get("code") in (40012, 40001), f"code={b.get('code')} msg={b.get('msg')}")
_, b = http("GET", f"{base}/api/v1/orgs/tree", token="garbage.token.value")
expect_code("E2 伪造 token 返回 40101", b, 40101)
_, b = http("GET", f"{base}/api/v1/orgs/abc", token=demo_token)
check("E3 路径参数类型错误返回 40001", b.get("code") == 40001, f"code={b.get('code')}")
_, b = http("POST", f"{base}/api/v1/orgs", {"orgName": "", "orgType": 99, "orgCode": ""},
token=demo_token)
check("E4 参数校验失败返回 40001 且带字段级信息", b.get("code") == 40001,
f"code={b.get('code')}")
return finish()
if __name__ == "__main__":
raise SystemExit(main())
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
DDL 同步:docs/db/schema-postgresql.sql → backend/.../db/migration/V1__business_tables.sql
为什么要有这个脚本
------------------
文档里的 DDL 是**交付物**(已实测、写进验收标准),Flyway 迁移是**工程产物**。
两者必须字字相同,否则"文档说一套、代码跑另一套"——这是最难查的一类事故。
所以:迁移文件由脚本生成,禁止手改;改 DDL 一律先改文档,再跑本脚本。
用法
----
python scripts/sync-ddl.py # 生成/更新迁移文件
python scripts/sync-ddl.py --check # 只校验是否同步(CI 用,不同步则退出码 1)
"""
from __future__ import annotations
import io
import os
import sys
# Windows 控制台默认不是 UTF-8
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SOURCE = os.path.join(ROOT, "docs", "db", "schema-postgresql.sql")
TARGET = os.path.join(
ROOT, "backend", "xt-server", "src", "main", "resources",
"db", "migration", "V1__business_tables.sql",
)
HEADER = """\
-- ═══════════════════════════════════════════════════════════════════
-- V1 · 业务表(20 张)
--
-- ⚠️ 本文件由脚本生成,请勿手改!
-- 来源 :docs/db/schema-postgresql.sql
-- 重新生成 :python scripts/sync-ddl.py
-- 校验同步 :python scripts/sync-ddl.py --check
--
-- 源文件已在 PostgreSQL 16.15 实测零错误:
-- 20 表 / 177 列注释 / 65 索引 / 20 表注释
-- ═══════════════════════════════════════════════════════════════════
"""
def read(path: str) -> str:
with io.open(path, encoding="utf-8") as f:
return f.read()
def normalize(text: str) -> str:
"""统一行尾并去掉首尾空白,用于比对。"""
return text.replace("\r\n", "\n").replace("\r", "\n").strip()
def render(source_text: str) -> str:
return HEADER + normalize(source_text) + "\n"
def main() -> int:
check_only = "--check" in sys.argv
if not os.path.exists(SOURCE):
print(f"[X] 找不到源文件:{SOURCE}")
return 1
expected = render(read(SOURCE))
if check_only:
if not os.path.exists(TARGET):
print(f"[X] 迁移文件不存在:{TARGET}")
return 1
actual = io.open(TARGET, encoding="utf-8").read()
if normalize(actual) != normalize(expected):
print("[X] DDL 已漂移:文档与迁移文件不一致")
print(f" 文档 :{SOURCE}")
print(f" 迁移 :{TARGET}")
print(" 修复 :python scripts/sync-ddl.py")
return 1
print("[OK] DDL 已同步")
return 0
os.makedirs(os.path.dirname(TARGET), exist_ok=True)
with io.open(TARGET, "w", encoding="utf-8", newline="\n") as f:
f.write(expected)
tables = expected.count("CREATE TABLE ")
indexes = expected.count("CREATE INDEX ") + expected.count("CREATE UNIQUE INDEX ")
comments = expected.count("COMMENT ON COLUMN ")
print(f"[OK] 已生成 {os.path.relpath(TARGET, ROOT)}")
print(f" {tables} 表 / {indexes} 索引 / {comments} 列注释 / "
f"{len(expected.encode('utf-8'))} 字节")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+201
View File
@@ -0,0 +1,201 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
文档一致性校验。
被校的东西都是历史上真实出错过的:
1. 表格列数一致(错位会让 Markdown 渲染成一堆竖线)
2. 接口数:逐节「标题声称数」vs「实际表行数」——只看总数会漏掉
"这节多算 3 个、那节少算 3 个、总数恰好相等"这种情况
3. 错误码数量与文档声称一致
4. 关键数字跨文档一致(FR / VAL / 任务 / 人日 / 表数 / 接口)
5. 原型屏数与 figcaption 数一致
用法
----
python scripts/verify-docs.py # 校验,失败退出码 1
"""
from __future__ import annotations
import io
import os
import re
import sys
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
D = os.path.join(ROOT, "docs")
BT = chr(96)
ok_count = 0
bad: list[str] = []
def ok(msg: str) -> None:
global ok_count
ok_count += 1
print(f" [OK] {msg}")
def fail(msg: str) -> None:
bad.append(msg)
print(f" [FAIL] {msg}")
def read(*parts: str) -> str:
p = os.path.join(ROOT, *parts)
return io.open(p, encoding="utf-8").read() if os.path.exists(p) else ""
# ── 1. 表格列数 ────────────────────────────────────────
def check_tables(name: str, text: str) -> None:
lines = text.split("\n")
bad_rows = 0
i = 0
while i < len(lines):
if lines[i].strip().startswith("|") and i + 1 < len(lines) \
and re.match(r"^\|[\s:|-]+\|$", lines[i + 1].strip()):
n = lines[i].count("|")
j = i + 2
while j < len(lines) and lines[j].strip().startswith("|"):
if lines[j].count("|") != n:
bad_rows += 1
j += 1
i = j
else:
i += 1
if bad_rows:
fail(f"{name}:{bad_rows} 行表格列数不一致")
else:
ok(f"{name}:表格列数一致")
# ── 2. 接口数逐节核对 ──────────────────────────────────
def check_api_count() -> int:
s = read("docs", "03-接口设计.md")
if not s:
fail("找不到 03-接口设计.md")
return 0
sec = s[s.index("## §4"):s.index("## §5")]
parts = re.split(r"^(### 4\.\d+ .+)$", sec, flags=re.M)
claim_sum = actual_sum = 0
mismatches = []
i = 1
while i < len(parts) - 1:
title, body = parts[i].strip(), parts[i + 1]
i += 2
m = re.search(r"[((](\d+)[))]\s*$", title)
claim = int(m.group(1)) if m else None
# ⚠️ 允许 ⭐ 出现在方法之前(| ⭐ GET | ... |)
actual = len(re.findall(r"^\|\s*(?:⭐\s*)?(GET|POST|PUT|DELETE|PATCH)\s*\|", body, re.M))
# 索引节(交叉引用别处已计数的接口),不重复计入
if "索引" in title or "新增" in title:
continue
if claim is None:
continue
claim_sum += claim
actual_sum += actual
if claim != actual:
mismatches.append(f"{title[:30]} 声称{claim} 实际{actual}")
if mismatches:
fail(f"03 接口数逐节不一致:{'; '.join(mismatches)}")
elif claim_sum == actual_sum:
ok(f"03 接口数逐节一致 = {actual_sum}")
else:
fail(f"03 接口数:声称 {claim_sum} vs 实际 {actual_sum}")
return actual_sum
# ── 3. 错误码 ──────────────────────────────────────────
def check_error_codes() -> int:
s = read("docs", "03-接口设计.md")
sec = s[s.index("## §15"):]
codes = set(re.findall(
r"^\|\s*(?:⭐\s*)?" + re.escape(BT) + r"(\d{4,5})" + re.escape(BT) + r"\s*\|", sec, re.M))
total = len(codes) + 1 # + 成功码 0
claimed = re.search(r"本节共 \*\*(\d+) 个\*\*", s)
if claimed and int(claimed.group(1)) != total:
fail(f"03 错误码:实际 {total} vs 文档声称 {claimed.group(1)}")
else:
ok(f"03 错误码 = {total}")
return total
# ── 4. 跨文档关键数字 ──────────────────────────────────
def check_cross_doc(api: int, err: int) -> None:
docs = {
"PRD": read("docs", "数据信息填报系统PRD.md"),
"03": read("docs", "03-接口设计.md"),
"05": read("docs", "05-开发任务拆解.md"),
"README": read("README.md"),
"交接摘要": read("docs", "交接摘要.md"),
}
checks = [
("接口", str(api)),
("错误码", str(err)),
("任务", "100"),
("人日", "387"),
]
for label, value in checks:
hit = [n for n, t in docs.items() if value in t]
if len(hit) >= 3:
ok(f"{label} {value} 出现于 {len(hit)} 份文档")
else:
fail(f"{label} {value} 仅出现于 {hit}")
# ── 5. 原型屏数 ────────────────────────────────────────
def check_screens() -> None:
total = 0
for f in ["student-app.html", "teacher-app.html", "admin-web.html"]:
s = read("docs", "prototype", f)
total += len(re.findall(r"<figcaption", s, re.I))
claimed = re.search(r"合计屏数\s*:\s*(\d+)", read("docs", "交接摘要.md"))
if total:
ok(f"原型屏数(figcaption 计数)= {total}")
if claimed and int(claimed.group(1)) != total:
fail(f"原型屏数:实际 {total} vs 交接摘要声称 {claimed.group(1)}")
def main() -> int:
print("=" * 66)
print(" 文档一致性校验")
print("=" * 66)
print("\n[1] 表格列数")
for f in ["README.md"]:
check_tables(f, read(f))
for f in os.listdir(D):
if f.endswith(".md"):
check_tables(f, read("docs", f))
print("\n[2] 接口数(逐节核对)")
api = check_api_count()
print("\n[3] 错误码")
err = check_error_codes()
print("\n[4] 跨文档关键数字")
check_cross_doc(api, err)
print("\n[5] 原型屏数")
check_screens()
print("\n" + "=" * 66)
if bad:
print(f" 结果: {ok_count} 项通过,{len(bad)} 项失败")
for b in bad:
print(f" - {b}")
else:
print(f" 结果: {ok_count} 项全部通过")
print("=" * 66)
return 1 if bad else 0
if __name__ == "__main__":
raise SystemExit(main())