cykj/shanxia2.0
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

提交“

Browse Source
This commit is contained in:
zlwu committed 2026-05-30 14:03:11 +08:00
1 parent 69ee0f505c
commit 9b88581450
614 files changed
+90318 -689022
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/scripts/check_field_consistency.py
-573
View File
@@ -1,573 +0,0 @@
#!/usr/bin/env python3
"""字段一致性检查器 — 检测 CRUD → API Schema → 前端 formData 字段缺失问题。
用法:
python backend/scripts/check_field_consistency.py
python backend/scripts/check_field_consistency.py --entity consumable
python backend/scripts/check_field_consistency.py --json
退出码:
0 — 全部通过
1 — 存在警告（字段缺失）
2 — 存在错误（文件/类找不到等）
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from typing import Any, Dict, List, Optional, Set, Tuple
# ---------------------------------------------------------------------------
# 项目根目录（从脚本位置推断）
# ---------------------------------------------------------------------------
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
_PROJECT_ROOT = os.path.abspath(os.path.join(_SCRIPT_DIR, "..", ".."))
# ---------------------------------------------------------------------------
# 只读字段 — 这些字段出现在 _row_to_dict 但不应出现在 CreateBody/UpdateBody/formData
# ---------------------------------------------------------------------------
READ_ONLY_FIELDS: Set[str] = {
"id",
"createTime",
"updateTime",
"stockQty",
"tenantNo",
"attachmentFiles", # 计算字段，由后端从 URLs 生成
}
# ---------------------------------------------------------------------------
# 实体注册表
# ---------------------------------------------------------------------------
ENTITIES: List[Dict[str, str]] = [
{
"name": "consumable",
"crud_file": "backend/app/crud/consumable.py",
"schema_file": "backend/app/api/production.py",
"create_body": "ConsumableCreateBody",
"update_body": "ConsumableUpdateBody",
"frontend_file": "web-admin/src/views/production/consumable/index.vue",
},
{
"name": "reagent",
"crud_file": "backend/app/crud/reagent.py",
"schema_file": "backend/app/api/production.py",
"create_body": "ReagentCreateBody",
"update_body": "ReagentUpdateBody",
"frontend_file": "web-admin/src/views/production/reagent/index.vue",
},
{
"name": "material (market)",
"crud_file": "backend/app/crud/material.py",
"schema_file": "backend/app/api/material.py",
"create_body": "MaterialBody",
"update_body": "MaterialUpdateBody",
"frontend_file": "web-admin/src/views/market/material/index.vue",
},
{
"name": "qc_material",
"crud_file": "backend/app/crud/qc_material.py",
"schema_file": "backend/app/api/qc.py",
"create_body": "QcMaterialBody",
"update_body": "QcMaterialUpdateBody",
"frontend_file": "web-admin/src/views/quality/material/index.vue",
},
{
"name": "sop_document",
"crud_file": "backend/app/crud/sop_document.py",
"schema_file": "backend/app/api/production.py",
"create_body": "SopDocumentCreateBody",
"update_body": "SopDocumentUpdateBody",
"frontend_file": "web-admin/src/views/production/sop/index.vue",
},
]
# ===================================================================
# 终端颜色辅助
# ===================================================================
class _C:
"""ANSI 颜色常量"""
RESET = "\033[0m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
RED = "\033[31m"
CYAN = "\033[36m"
BOLD = "\033[1m"
DIM = "\033[2m"
def _supports_color() -> bool:
"""检测终端是否支持颜色"""
if os.environ.get("NO_COLOR"):
return False
if not hasattr(sys.stdout, "isatty"):
return False
return sys.stdout.isatty()
_COLOR = _supports_color()
def cprint(text: str, color: str = "") -> None:
if _COLOR and color:
print("{}{}{}".format(color, text, _C.RESET))
else:
print(text)
# ===================================================================
# 工具函数
# ===================================================================
def _read_file(rel_path: str) -> Optional[str]:
"""读取项目根目录下的相对路径文件"""
full = os.path.join(_PROJECT_ROOT, rel_path)
if not os.path.isfile(full):
return None
with open(full, "r", encoding="utf-8", errors="replace") as f:
return f.read()
def _snake_to_camel(name: str) -> str:
"""snake_case → camelCase: warning_qty → warningQty"""
parts = name.split("_")
return parts[0] + "".join(p.capitalize() for p in parts[1:])
# ===================================================================
# 提取逻辑 — _row_to_dict
# ===================================================================
_RE_DICT_KEY = re.compile(r'"([a-zA-Z][a-zA-Z0-9]*)"\s*:')
def extract_row_to_dict_fields(source: str) -> List[str]:
"""从 CRUD 文件提取 _row_to_dict 函数的字典键（camelCase）。
支持一个文件中有多个 _row_to_dict 的情况 — 只取第一个。
"""
# 定位函数起始
func_match = re.search(
r"def\s+_row_to_dict\s*\([^)]*\)\s*->\s*[^:]+:", source
)
if not func_match:
return []
# 从函数开始往后找到第一个 return { ... } 块
start = func_match.end()
# 找到 return {
ret_match = re.search(r"return\s*\{", source[start:])
if not ret_match:
return []
brace_start = start + ret_match.end() - 1 # 指向 {
# 匹配大括号
depth = 0
end = brace_start
for i in range(brace_start, len(source)):
ch = source[i]
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
end = i
break
block = source[brace_start : end + 1]
keys = _RE_DICT_KEY.findall(block)
# 去重但保持顺序
seen = set()
result = []
for k in keys:
if k not in seen:
seen.add(k)
result.append(k)
return result
# ===================================================================
# 提取逻辑 — Pydantic Schema 类
# ===================================================================
_RE_FIELD_ALIAS = re.compile(r'alias\s*=\s*"([^"]+)"')
_RE_FIELD_LINE = re.compile(
r"^\s+([a-z_][a-z0-9_]*)\s*:\s*"
)
def extract_schema_fields(source: str, class_name: str) -> Optional[List[str]]:
"""从 Pydantic 类提取字段名（返回 camelCase 列表）。
优先使用 alias，否则 snake_case 转 camelCase。
返回 None 表示类未找到。
"""
# 找到类定义
class_match = re.search(
r"class\s+{}\s*\([^)]*\)\s*:".format(re.escape(class_name)),
source,
)
if not class_match:
return None
class_start = class_match.end()
# 收集类体 — 到下一个顶层 class 或 def（不缩进）或文件末尾
body_lines = []
for line in source[class_start:].split("\n"):
stripped = line.lstrip()
if stripped and not stripped.startswith("#") and not line[0].isspace():
# 到达下一个顶层定义
break
if stripped.startswith("class ") or (stripped.startswith("def ") and not line.startswith(" ")):
break
body_lines.append(line)
body = "\n".join(body_lines)
fields = []
# 按行解析
for line in body.split("\n"):
field_match = _RE_FIELD_LINE.match(line)
if not field_match:
continue
field_name = field_match.group(1)
if field_name == "model_config":
continue
# 检查是否有 alias
alias_match = _RE_FIELD_ALIAS.search(line)
if alias_match:
fields.append(alias_match.group(1))
else:
fields.append(_snake_to_camel(field_name))
return fields
# ===================================================================
# 提取逻辑 — 前端 formData
# ===================================================================
_RE_FORMDATA_KEY = re.compile(r"^\s+([a-zA-Z][a-zA-Z0-9]*)\s*:")
def extract_frontend_formdata(source: str) -> List[str]:
"""从 Vue 文件提取 formData 的键名（camelCase）。"""
# 匹配多种写法: reactive({ ... }), reactive<SomeType>({ ... })
match = re.search(
r"(?:const|let|var)\s+formData\s*=\s*reactive(?:<[^>]*>)?\s*\(\s*\{",
source,
)
if not match:
return []
start = match.end() - 1 # 指向 {
depth = 0
end = start
for i in range(start, len(source)):
ch = source[i]
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
end = i
break
block = source[start : end + 1]
fields = []
for line in block.split("\n"):
m = _RE_FORMDATA_KEY.match(line)
if m:
fields.append(m.group(1))
return fields
# ===================================================================
# 单个实体检查
# ===================================================================
def check_entity(entity: Dict[str, str]) -> Dict[str, Any]:
"""检查单个实体的字段一致性，返回报告字典。"""
name = entity["name"]
report: Dict[str, Any] = {
"entity": name,
"status": "ok",
"crud_fields": [],
"create_body_fields": None,
"update_body_fields": None,
"frontend_fields": [],
"warnings": [],
"errors": [],
}
# 1. 读取 CRUD 文件
crud_src = _read_file(entity["crud_file"])
if crud_src is None:
report["status"] = "error"
report["errors"].append("CRUD file not found: {}".format(entity["crud_file"]))
return report
crud_fields = extract_row_to_dict_fields(crud_src)
report["crud_fields"] = crud_fields
if not crud_fields:
report["status"] = "error"
report["errors"].append(
"No _row_to_dict fields found in {}".format(entity["crud_file"])
)
return report
writable_fields = [f for f in crud_fields if f not in READ_ONLY_FIELDS]
# 2. 读取 Schema 文件
schema_src = _read_file(entity["schema_file"])
if schema_src is None:
report["status"] = "error"
report["errors"].append(
"Schema file not found: {}".format(entity["schema_file"])
)
return report
# 3. 提取 CreateBody
create_fields = extract_schema_fields(schema_src, entity["create_body"])
report["create_body_fields"] = create_fields
if create_fields is None:
report["status"] = "warning"
report["warnings"].append(
"Create body class '{}' not found in {}".format(
entity["create_body"], entity["schema_file"]
)
)
else:
missing = [f for f in writable_fields if f not in create_fields]
if missing:
report["status"] = "warning"
report["warnings"].append(
"CreateBody missing fields: {}".format(", ".join(sorted(missing)))
)
# 4. 提取 UpdateBody
update_fields = extract_schema_fields(schema_src, entity["update_body"])
report["update_body_fields"] = update_fields
if update_fields is None:
report["status"] = "warning"
report["warnings"].append(
"Update body class '{}' not found in {}".format(
entity["update_body"], entity["schema_file"]
)
)
else:
missing = [f for f in writable_fields if f not in update_fields]
if missing:
report["status"] = "warning"
report["warnings"].append(
"UpdateBody missing fields: {}".format(", ".join(sorted(missing)))
)
# 5. 读取前端文件
fe_src = _read_file(entity["frontend_file"])
if fe_src is None:
report["status"] = "warning"
report["warnings"].append(
"Frontend file not found: {}".format(entity["frontend_file"])
)
return report
fe_fields = extract_frontend_formdata(fe_src)
report["frontend_fields"] = fe_fields
if not fe_fields:
report["status"] = "warning"
report["warnings"].append(
"No formData found in {}".format(entity["frontend_file"])
)
return report
missing_fe = [f for f in writable_fields if f not in fe_fields]
if missing_fe:
report["status"] = "warning"
report["warnings"].append(
"Frontend formData missing fields: {}".format(
", ".join(sorted(missing_fe))
)
)
# 6. 反向检查: Schema 有但 _row_to_dict 没有的字段
if create_fields is not None:
extra_create = [f for f in create_fields if f not in crud_fields]
if extra_create:
report.setdefault("info", []).append(
"CreateBody has extra fields (not in _row_to_dict): {}".format(
", ".join(sorted(extra_create))
)
)
# 7. 反向检查: formData 有但 _row_to_dict 没有的字段
extra_fe = [f for f in fe_fields if f not in crud_fields]
if extra_fe:
report.setdefault("info", []).append(
"Frontend formData has extra fields (not in _row_to_dict): {}".format(
", ".join(sorted(extra_fe))
)
)
return report
# ===================================================================
# 报告输出
# ===================================================================
def print_report(report: Dict[str, Any]) -> None:
"""打印单个实体的检查报告（终端彩色输出）"""
name = report["entity"]
status = report["status"]
if status == "ok":
cprint(" [OK] {}".format(name), _C.GREEN)
elif status == "warning":
cprint(" [!!] {} — warnings".format(name), _C.YELLOW)
else:
cprint(" [XX] {} — errors".format(name), _C.RED)
# CRUD fields
crud_fields = report.get("crud_fields", [])
if crud_fields:
writable = [f for f in crud_fields if f not in READ_ONLY_FIELDS]
read_only = [f for f in crud_fields if f in READ_ONLY_FIELDS]
cprint(
" _row_to_dict ({} total, {} writable): {}".format(
len(crud_fields),
len(writable),
", ".join(writable),
),
_C.DIM,
)
if read_only:
cprint(
" read-only: {}".format(", ".join(sorted(read_only))),
_C.DIM,
)
# Schema fields
for body_label, body_key in [
("CreateBody", "create_body_fields"),
("UpdateBody", "update_body_fields"),
]:
fields = report.get(body_key)
if fields is not None:
cprint(
" {}: {}".format(body_label, ", ".join(fields)),
_C.DIM,
)
# Frontend fields
fe_fields = report.get("frontend_fields", [])
if fe_fields:
cprint(
" formData: {}".format(", ".join(fe_fields)),
_C.DIM,
)
# Warnings
for w in report.get("warnings", []):
cprint(" WARN: {}".format(w), _C.YELLOW)
# Errors
for e in report.get("errors", []):
cprint(" ERR: {}".format(e), _C.RED)
# Info
for info in report.get("info", []):
cprint(" INFO: {}".format(info), _C.CYAN)
print()
def print_json_report(reports: List[Dict[str, Any]]) -> None:
"""输出 JSON 格式报告"""
print(json.dumps(reports, indent=2, ensure_ascii=False))
# ===================================================================
# 主入口
# ===================================================================
def main() -> int:
parser = argparse.ArgumentParser(
description="字段一致性检查器 — CRUD ↔ Schema ↔ 前端 formData"
)
parser.add_argument(
"--entity",
type=str,
default=None,
help="只检查指定实体名称（如 consumable）",
)
parser.add_argument(
"--json",
action="store_true",
dest="json_output",
help="输出 JSON 格式",
)
args = parser.parse_args()
# 筛选实体
entities = ENTITIES
if args.entity:
entities = [e for e in ENTITIES if e["name"] == args.entity]
if not entities:
cprint(
"Entity '{}' not found. Available: {}".format(
args.entity, ", ".join(e["name"] for e in ENTITIES)
),
_C.RED,
)
return 2
if not args.json_output:
cprint(
"Field Consistency Checker — checking {} entities".format(
len(entities)
),
_C.BOLD,
)
cprint("Project root: {}".format(_PROJECT_ROOT), _C.DIM)
print()
reports = []
has_warnings = False
has_errors = False
for entity in entities:
report = check_entity(entity)
reports.append(report)
if report["status"] == "warning":
has_warnings = True
elif report["status"] == "error":
has_errors = True
if args.json_output:
print_json_report(reports)
else:
for report in reports:
print_report(report)
# 汇总
ok_count = sum(1 for r in reports if r["status"] == "ok")
warn_count = sum(1 for r in reports if r["status"] == "warning")
err_count = sum(1 for r in reports if r["status"] == "error")
cprint(
"Summary: {} OK, {} warnings, {} errors (total {})".format(
ok_count, warn_count, err_count, len(reports)
),
_C.BOLD,
)
if has_errors:
return 2
if has_warnings:
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
backend/scripts/clear_biz_orders.py
-48
View File
@@ -1,48 +0,0 @@
#!/usr/bin/env python3
"""清空订单及相关表（开发调试用）。用法：在 backend 目录执行 python scripts/clear_biz_orders.py"""
from __future__ import annotations
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from sqlalchemy import text
from app.core.database import engine
TABLES = [
"biz_order_status_log_his",
"biz_order_status_log",
"biz_order_item",
"biz_order_his",
"biz_order_adjust_his",
"biz_order_adjust",
"biz_wechat_pay",
"biz_refund_his",
"biz_refund",
"biz_settlement_his",
"biz_settlement",
"biz_invoice",
"biz_deliverer_review_his",
"biz_deliverer_review",
"biz_delivery_warning_his",
"biz_delivery_warning",
"biz_delivery_task_item",
"biz_delivery_task_his",
"biz_delivery_task",
"biz_order",
]
def main() -> None:
with engine.begin() as conn:
conn.execute(text("SET FOREIGN_KEY_CHECKS=0"))
for t in TABLES:
conn.execute(text(f"TRUNCATE TABLE `{t}`"))
conn.execute(text("SET FOREIGN_KEY_CHECKS=1"))
print("已清空订单及相关表:", ", ".join(TABLES))
if __name__ == "__main__":
main()
backend/scripts/migrate_minio_urls.py
-195
View File
@@ -1,195 +0,0 @@
#!/usr/bin/env python3
"""批量替换数据库历史 URL 前缀（MinIO 域名迁移脚本）。
用法（在 backend 目录执行）：
1) 先预览：
python scripts/migrate_minio_urls.py --dry-run
2) 正式执行：
python scripts/migrate_minio_urls.py
3) 自定义前缀：
python scripts/migrate_minio_urls.py \
--old-prefix "https://www.chengyantek.cn:20004" \
--new-prefix "https://www.chengyantek.cn"
"""
from __future__ import annotations
import argparse
import sys
from datetime import datetime
from pathlib import Path
from sqlalchemy import text
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.core.database import engine # noqa: E402
DEFAULT_OLD_PREFIX = "https://www.chengyantek.cn:20004"
DEFAULT_NEW_PREFIX = "https://www.chengyantek.cn"
TEXT_TYPES = (
"char",
"varchar",
"tinytext",
"text",
"mediumtext",
"longtext",
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="批量替换数据库中的 URL 前缀")
parser.add_argument("--old-prefix", default=DEFAULT_OLD_PREFIX, help="旧 URL 前缀")
parser.add_argument("--new-prefix", default=DEFAULT_NEW_PREFIX, help="新 URL 前缀")
parser.add_argument(
"--dry-run",
action="store_true",
help="仅预览受影响记录数，不执行更新",
)
return parser.parse_args()
def quote_ident(name: str) -> str:
"""安全引用 MySQL 标识符。"""
return f"`{name.replace('`', '``')}`"
def get_text_columns(conn) -> list[tuple[str, str]]:
sql = text(
"""
SELECT table_name, column_name
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND data_type IN :types
ORDER BY table_name, ordinal_position
"""
)
rows = conn.execute(sql, {"types": TEXT_TYPES}).fetchall()
return [(r[0], r[1]) for r in rows]
def count_affected(conn, table_name: str, column_name: str, old_prefix: str) -> int:
sql = text(
f"""
SELECT COUNT(*) AS cnt
FROM {quote_ident(table_name)}
WHERE {quote_ident(column_name)} LIKE :pattern
"""
)
return int(conn.execute(sql, {"pattern": f"%{old_prefix}%"}).scalar() or 0)
def update_column(
conn,
table_name: str,
column_name: str,
old_prefix: str,
new_prefix: str,
) -> int:
sql = text(
f"""
UPDATE {quote_ident(table_name)}
SET {quote_ident(column_name)} = REPLACE({quote_ident(column_name)}, :old_prefix, :new_prefix)
WHERE {quote_ident(column_name)} LIKE :pattern
"""
)
result = conn.execute(
sql,
{
"old_prefix": old_prefix,
"new_prefix": new_prefix,
"pattern": f"%{old_prefix}%",
},
)
return int(result.rowcount or 0)
def make_backup_table_name(table_name: str, suffix: str) -> str:
"""生成备份表名，控制在 MySQL 64 字符限制内。"""
max_table_len = 64
reserve = len("_bak_") + len(suffix)
max_base_len = max_table_len - reserve
base = table_name[:max_base_len]
return f"{base}_bak_{suffix}"
def backup_table(conn, table_name: str, backup_table_name: str) -> None:
"""备份单张表（结构+数据）。"""
conn.execute(
text(
f"CREATE TABLE {quote_ident(backup_table_name)} LIKE {quote_ident(table_name)}"
)
)
conn.execute(
text(
f"INSERT INTO {quote_ident(backup_table_name)} SELECT * FROM {quote_ident(table_name)}"
)
)
def main() -> None:
args = parse_args()
old_prefix = args.old_prefix.strip().rstrip("/")
new_prefix = args.new_prefix.strip().rstrip("/")
dry_run = args.dry_run
if not old_prefix or not new_prefix:
raise ValueError("old-prefix/new-prefix 不能为空")
if old_prefix == new_prefix:
raise ValueError("old-prefix 与 new-prefix 不能相同")
print(f"数据库 URL 迁移开始，模式：{'预览(dry-run)' if dry_run else '正式执行'}")
print(f"旧前缀: {old_prefix}")
print(f"新前缀: {new_prefix}")
total_rows = 0
hit_columns = 0
affected_items: list[tuple[str, str, int]] = []
with engine.begin() as conn:
columns = get_text_columns(conn)
for table_name, column_name in columns:
affected = count_affected(conn, table_name, column_name, old_prefix)
if affected <= 0:
continue
affected_items.append((table_name, column_name, affected))
if not affected_items:
print("未发现需要替换的数据，已结束。")
return
if not dry_run:
backup_suffix = datetime.now().strftime("%Y%m%d%H%M%S")
backup_tables: dict[str, str] = {}
for table_name, _, _ in affected_items:
if table_name in backup_tables:
continue
backup_name = make_backup_table_name(table_name, backup_suffix)
backup_table(conn, table_name, backup_name)
backup_tables[table_name] = backup_name
print(f"[BACKUP] {table_name} -> {backup_name}")
print(f"备份完成：共备份 {len(backup_tables)} 张表。")
for table_name, column_name, affected in affected_items:
hit_columns += 1
if dry_run:
print(f"[DRY-RUN] {table_name}.{column_name} -> {affected} 行")
total_rows += affected
continue
updated = update_column(conn, table_name, column_name, old_prefix, new_prefix)
print(f"[UPDATED] {table_name}.{column_name} -> {updated} 行")
total_rows += updated
if dry_run:
print(f"预览完成：共 {hit_columns} 个字段命中，预计影响 {total_rows} 行。")
else:
print(f"执行完成：共更新 {hit_columns} 个字段，影响 {total_rows} 行。")
if __name__ == "__main__":
main()
backend/scripts/reorder_web_menus.py
-245
View File
@@ -1,245 +0,0 @@
#!/usr/bin/env python3
"""重排 Web 端一级菜单顺序，并生成 sys_menu 初始化 SQL。"""
from __future__ import annotations
import ast
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
MENU_PY = ROOT / "backend/app/api/menu.py"
SQXP_TS = ROOT / "web-admin/src/router/modules/sqxp.ts"
SQL_OUT = ROOT / "docs/sql/sys_menu_web_init.sql"
BLOCK_ORDER = [2, 3, 4, 5, 7, 9, 10, 1, 6, 8, 11, 12, 13]
BLOCK_TITLES = {
1: "市场基础信息",
2: "细胞存储",
3: "科研捐赠",
4: "中试服务",
5: "临床应用",
6: "质检基础信息",
7: "质检管理",
8: "生产基础信息",
9: "生产管理",
10: "运营支撑",
11: "系统基础信息",
12: "工作流管理",
13: "系统管理",
}
SQXP_HEADER = """import type { AppRouteRecord } from '@/types/router'
/**
* 前端路由（13 个一级菜单：细胞存储、科研捐赠、中试服务、临床应用、
* 质检管理、生产管理、运营支撑、市场基础信息、质检基础信息、生产基础信息、
* 系统基础信息、工作流管理、系统管理）
* 与后端菜单对齐时请同步 `backend/app/api/menu.py`。
*/
export const sqxpAdminRoutes: AppRouteRecord[] = [
"""
MENU_TAIL = (
"\n]\n\n\n"
'@router.get("/list")\n'
"async def get_menu_list(_user: dict = Depends(require_admin_or_super_admin)):\n"
' """获取菜单列表（与当前前端路由一致）"""\n'
' return success_response({"menuList": STATIC_MENU_TREE})\n'
)
BLOCK_RE_TS = re.compile(
r" // ──+\n // (\d+)\. ([^\n]+)\n // ──+\n",
re.MULTILINE,
)
BLOCK_RE_PY = re.compile(
r" # ──+\n # (\d+)\. ([^\n]+)\n # ──+\n",
re.MULTILINE,
)
def split_menu_blocks(
content: str, array_start_marker: str, block_re: re.Pattern
) -> tuple[str, dict[int, str]]:
start = content.index(array_start_marker) + len(array_start_marker)
depth = 1
i = start
while i < len(content) and depth > 0:
if content[i] == "[":
depth += 1
elif content[i] == "]":
depth -= 1
i += 1
array_body = content[start : i - 1]
prefix = content[:start]
parts = block_re.split(array_body)
blocks: dict[int, str] = {}
idx = 1
while idx < len(parts):
num = int(parts[idx])
block = parts[idx + 2].rstrip()
if block.endswith(","):
block = block[:-1]
blocks[num] = block.rstrip()
idx += 3
return prefix, blocks
def reorder_blocks(blocks: dict[int, str]) -> list[tuple[int, str, str]]:
ordered = []
for new_idx, old_num in enumerate(BLOCK_ORDER, start=1):
ordered.append((new_idx, BLOCK_TITLES[old_num], blocks[old_num]))
return ordered
def assemble_blocks(ordered: list[tuple[int, str, str]], *, ts: bool) -> str:
chunks = []
for num, title, body in ordered:
if ts:
sep = " // " + "─" * 46
chunks.append(f"{sep}\n // {num}. {title}\n{sep}\n{body}")
else:
sep = " # " + "─" * 44
chunks.append(f"{sep}\n # {num}. {title}\n{sep}\n{body}")
return ",\n\n".join(chunks) + "\n"
def load_static_menu_tree() -> list:
content = MENU_PY.read_text(encoding="utf-8")
marker = "STATIC_MENU_TREE = "
start = content.index(marker) + len(marker)
depth = 0
i = start
while i < len(content):
ch = content[i]
if ch == "[":
depth += 1
elif ch == "]":
depth -= 1
if depth == 0:
i += 1
break
i += 1
literal = content[start:i]
return ast.literal_eval(literal)
def sql_escape(val: str | None) -> str:
if val is None:
return "NULL"
return "'" + str(val).replace("'", "''") + "'"
def walk_routes(nodes: list, parent_id: str, order_counter: dict, rows: list) -> None:
for node in nodes:
node_id = node["id"]
meta = node.get("meta") or {}
title = meta.get("title") or node.get("name") or ""
path = node.get("path") or ""
component = node.get("component") or ""
children = node.get("children") or []
is_hide = meta.get("isHide") is True
menu_code = meta.get("menuCode")
perms = menu_code if menu_code else None
if parent_id == "2":
order_counter["web"] += 1
order_num = order_counter["web"]
else:
order_counter.setdefault(parent_id, 0)
order_counter[parent_id] += 1
order_num = order_counter[parent_id]
menu_type = "M" if children else "C"
visible = "0" if is_hide else "1"
rows.append(
(
node_id,
title,
parent_id,
menu_type,
path,
component or None,
perms,
visible,
order_num,
)
)
if children:
walk_routes(children, node_id, order_counter, rows)
def generate_sql(tree: list) -> str:
rows: list = []
order_counter: dict = {"web": 0}
walk_routes(tree, "2", order_counter, rows)
lines = [
"-- Web 端 sys_menu 初始化（根节点 + 13 个一级业务菜单及子菜单）",
"-- 生成：backend/scripts/reorder_web_menus.py",
"-- 一级顺序：细胞存储 → 科研捐赠 → 中试服务 → 临床应用 → 质检管理 → 生产管理",
"-- → 运营支撑 → 市场基础信息 → 质检基础信息 → 生产基础信息",
"-- → 系统基础信息 → 工作流管理 → 系统管理",
"",
"SET NAMES utf8mb4;",
"",
"INSERT INTO `sys_menu` (`id`, `menu_name`, `parent_id`, `menu_type`, `path`, `component`, `perms`, `is_frame`, `visible`, `status`, `tenant_no`, `order_num`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`) VALUES",
"('1', '移动端菜单', '0', 'M', '/mini', NULL, NULL, 1, '1', '1', 'zkxw', 1, NULL, NULL, '', NOW(), NULL, NOW(), 0),",
"('2', 'Web端菜单', '0', 'M', '/web', NULL, NULL, 1, '1', '1', 'zkxw', 2, NULL, NULL, '', NOW(), NULL, NOW(), 0);",
"",
"INSERT INTO `sys_menu` (`id`, `menu_name`, `parent_id`, `menu_type`, `path`, `component`, `perms`, `is_frame`, `visible`, `status`, `tenant_no`, `order_num`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`) VALUES",
]
value_lines = []
for row in rows:
(
node_id,
title,
parent_id,
menu_type,
path,
component,
perms,
visible,
order_num,
) = row
value_lines.append(
f"({sql_escape(node_id)}, {sql_escape(title)}, {sql_escape(parent_id)}, "
f"{sql_escape(menu_type)}, {sql_escape(path)}, {sql_escape(component)}, "
f"{sql_escape(perms)}, 1, {sql_escape(visible)}, '1', 'zkxw', {order_num}, "
f"NULL, NULL, '', NOW(), NULL, NOW(), 0)"
)
lines.append(",\n".join(value_lines) + ";")
lines.append("")
return "\n".join(lines)
def main() -> None:
menu_content = MENU_PY.read_text(encoding="utf-8")
pre, blocks = split_menu_blocks(menu_content, "STATIC_MENU_TREE = [\n", BLOCK_RE_PY)
body = assemble_blocks(reorder_blocks(blocks), ts=False)
MENU_PY.write_text(pre + body + MENU_TAIL, encoding="utf-8")
print(f"updated {MENU_PY.relative_to(ROOT)}")
ts_content = SQXP_TS.read_text(encoding="utf-8")
_, blocks_ts = split_menu_blocks(
ts_content, "export const sqxpAdminRoutes: AppRouteRecord[] = [\n", BLOCK_RE_TS
)
body_ts = assemble_blocks(reorder_blocks(blocks_ts), ts=True)
SQXP_TS.write_text(SQXP_HEADER + body_ts + "]\n", encoding="utf-8")
print(f"updated {SQXP_TS.relative_to(ROOT)}")
tree = load_static_menu_tree()
sql = generate_sql(tree)
SQL_OUT.write_text(sql, encoding="utf-8")
row_count = sql.count("('") - 2 # minus root inserts
print(f"wrote {SQL_OUT.relative_to(ROOT)} (~{row_count} menu rows, {len(tree)} top-level menus)")
if __name__ == "__main__":
main()