# -*- coding: utf-8 -*- """iAOP-Template-Resin 模板资产 sanity 检查(无构建环境下的离线基本验证)。 检查项: 1. 点位字典 CSV:表头与内核 `core/edge-gateway/config/point_dict.example.csv` 对齐(9 列),且非空、无空行; 2. 所有 YAML 资产(RAG 配置 / 驾驶舱布局 / 版本清单)可被 yaml 解析; 3. version.yaml 声明的 assets 清单文件全部存在; 4. 驾驶舱布局 widget 类型在 PRD 5.5 schema 允许集合内。 用法:python _sanity_check.py """ import csv import os import sys import yaml # noqa: E402 HERE = os.path.dirname(os.path.abspath(__file__)) # 对齐 core/edge-gateway/config/point_dict.example.csv 表头 EXPECTED_COLUMNS = [ "device_id", "point_id", "name", "unit", "dataType", "sampleRate", "qualityCode", "opcNode", "protocol", ] # PRD 5.5 iAOP-cockpit-layout-v1 允许的 widget 类型 ALLOWED_WIDGETS = { "process_view", "trend", "kpi_card", "alarm_panel", "nl_query", } def main() -> int: failures: list[str] = [] # 1) 点位字典 CSV csv_path = os.path.join(HERE, "point-dict", "point_dict.resin.csv") with open(csv_path, "r", encoding="utf-8") as fh: rows = list(csv.reader(fh)) if not rows: failures.append("point_dict.resin.csv 为空") else: header = [c.strip() for c in rows[0]] if header != EXPECTED_COLUMNS: failures.append( f"CSV 表头与内核 schema 不一致:{header} != {EXPECTED_COLUMNS}") if len(rows) < 2: failures.append("CSV 缺少数据行") for i, row in enumerate(rows[1:], 2): if len(row) != len(EXPECTED_COLUMNS): failures.append(f"CSV 第 {i} 行列数异常:{len(row)}") # 2) YAML 资产可解析 for rel in ("rag-kb/kb.resin.template.yaml", "dashboard/cockpit.resin.yaml", "version.yaml"): path = os.path.join(HERE, rel) with open(path, "r", encoding="utf-8") as fh: yaml.safe_load(fh) # 3) version.yaml assets 清单存在 with open(os.path.join(HERE, "version.yaml"), "r", encoding="utf-8") as fh: version = yaml.safe_load(fh) for rel in version.get("assets", []): if not os.path.isfile(os.path.join(HERE, rel)): failures.append(f"version.yaml 声明资产缺失:{rel}") # 4) 驾驶舱布局 widget 类型合法 with open(os.path.join(HERE, "dashboard", "cockpit.resin.yaml"), "r", encoding="utf-8") as fh: cockpit = yaml.safe_load(fh) if cockpit.get("$schema") != "iAOP-cockpit-layout-v1": failures.append("驾驶舱布局缺少/错误 $schema") for w in cockpit.get("widgets", []): if w.get("type") not in ALLOWED_WIDGETS: failures.append(f"非法 widget 类型:{w.get('type')}") if failures: print("FAIL") for f in failures: print(" -", f) return 1 print(f"OK: 点位 {len(rows) - 1} 条,YAML/清单/widget 校验通过") return 0 if __name__ == "__main__": sys.exit(main())