574 lines
18 KiB
Python
574 lines
18 KiB
Python
#!/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())
|