zkxw/backend/scripts/check_field_consistency.py

#!/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())