198 lines
8.2 KiB
Python
Executable File
198 lines
8.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import json
|
||
import sys
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
|
||
CATEGORY_MAPPING = {
|
||
"基本信息": ["basic_info", "基本信息"],
|
||
"技术特性": ["technical_features", "technical_characteristics", "技术特性"],
|
||
"性能指标": ["performance_metrics", "performance", "性能指标"],
|
||
"里程碑意义": ["milestone_significance", "milestones", "里程碑意义"],
|
||
"商业信息": ["business_info", "commercial_info", "商业信息"],
|
||
"竞争与生态": ["competition_ecosystem", "competition", "竞争与生态"],
|
||
"历史沿革": ["history", "历史沿革"],
|
||
"市场定位": ["market_positioning", "market", "市场定位"],
|
||
}
|
||
|
||
_SKIP_KEYS = {"_source_file", "uncertain"}
|
||
|
||
|
||
def load_fields_yaml(fields_path):
|
||
"""解析 fields.yaml，仅接受 research skill 实际生成的唯一 schema：
|
||
|
||
fields:
|
||
<category>:
|
||
- {name: ..., description: ..., detail_level: ...}
|
||
...
|
||
uncertain: []
|
||
|
||
只接受这一种格式。不符合的 fields.yaml 会直接报错退出，
|
||
而不是以零字段静默通过。
|
||
|
||
Required 语义（避免校验器空转通过/说谎）：
|
||
- 若任一字段显式带 `required:` 键 -> 沿用 opt-in 语义
|
||
- 否则（detail_level 式、无标记）-> 全部字段视为必填，因为脚本的目的就是完整覆盖。
|
||
"""
|
||
with fields_path.open(encoding="utf-8") as f:
|
||
data = yaml.safe_load(f) or {}
|
||
defs = [] # (name, category, required_or_None)
|
||
|
||
fn = data.get("fields")
|
||
if not isinstance(fn, dict):
|
||
print(f"[错误] fields.yaml 必须使用 `fields: {{<category>: [{{name, ...}}]}}` 格式；当前为 {type(fn).__name__ if fn is not None else 'None'}。")
|
||
sys.exit(1)
|
||
|
||
for cname, flist in fn.items():
|
||
if cname in _SKIP_KEYS:
|
||
continue
|
||
if not isinstance(flist, list):
|
||
print(f"[错误] 分类 `{cname}` 必须映射为字段字典列表；当前为 {type(flist).__name__}。")
|
||
sys.exit(1)
|
||
for field in flist:
|
||
if isinstance(field, dict) and "name" in field:
|
||
defs.append((str(field["name"]), str(cname), field.get("required", None)))
|
||
else:
|
||
print(f"[错误] `{cname}` 下的字段必须是带 `name` 键的字典；当前为 {field!r}。")
|
||
sys.exit(1)
|
||
|
||
if not defs:
|
||
print("[错误] fields.yaml 解析到零个字段。请确保至少有一个分类及其字段字典。")
|
||
sys.exit(1)
|
||
|
||
all_fields = {n for n, _, _ in defs}
|
||
if any(r is not None for _, _, r in defs):
|
||
required_fields = {n for n, _, r in defs if r}
|
||
else:
|
||
required_fields = set(all_fields)
|
||
field_categories = {n: c for n, c, _ in defs}
|
||
return all_fields, required_fields, field_categories
|
||
|
||
|
||
def extract_json_fields(data, category_mapping=None):
|
||
category_mapping = CATEGORY_MAPPING if category_mapping is None else category_mapping
|
||
nested_keys = {k for keys in category_mapping.values() for k in keys}
|
||
fields = set()
|
||
stack = [(data, True)]
|
||
while stack:
|
||
obj, is_category_level = stack.pop()
|
||
if isinstance(obj, dict):
|
||
for k, v in obj.items():
|
||
if k in _SKIP_KEYS:
|
||
continue
|
||
if is_category_level and k in nested_keys:
|
||
if isinstance(v, dict):
|
||
stack.append((v, True))
|
||
continue
|
||
fields.add(k)
|
||
elif isinstance(obj, list):
|
||
stack.extend((item, is_category_level) for item in obj if isinstance(item, dict))
|
||
return fields
|
||
|
||
|
||
def validate_json(json_path, all_fields, required_fields, field_categories):
|
||
with json_path.open(encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
json_fields = extract_json_fields(data)
|
||
covered = all_fields & json_fields
|
||
missing = all_fields - json_fields
|
||
extra = json_fields - all_fields
|
||
missing_required = missing & required_fields
|
||
missing_by_category = defaultdict(list)
|
||
for field in missing:
|
||
missing_by_category[field_categories.get(field, "未知")].append(field)
|
||
return {
|
||
"file": json_path.name,
|
||
"total_defined": len(all_fields),
|
||
"covered": len(covered),
|
||
"missing": len(missing),
|
||
"extra": len(extra),
|
||
"coverage_rate": len(covered) / len(all_fields) * 100 if all_fields else 100,
|
||
"missing_required": sorted(missing_required),
|
||
"missing_optional": sorted(missing - required_fields),
|
||
"missing_by_category": {k: sorted(v) for k, v in missing_by_category.items()},
|
||
"extra_fields": sorted(extra),
|
||
"valid": len(missing_required) == 0,
|
||
}
|
||
|
||
|
||
def print_result(result, verbose=True):
|
||
status = "通过" if result["valid"] else "失败"
|
||
line = "=" * 60
|
||
print(f"\n{line}")
|
||
print(f"[{status}] {result['file']}")
|
||
print(line)
|
||
print(f"覆盖率: {result['coverage_rate']:.1f}% ({result['covered']}/{result['total_defined']})")
|
||
if result["missing_required"]:
|
||
print(f"\n[错误] 缺少必填字段 ({len(result['missing_required'])}):")
|
||
print("\n".join(f" - {f}" for f in result["missing_required"]))
|
||
if verbose and result["missing_optional"]:
|
||
missing_required = set(result["missing_required"])
|
||
print(f"\n[警告] 缺少可选字段 ({len(result['missing_optional'])}):")
|
||
for cat in sorted(result["missing_by_category"]):
|
||
optional = [f for f in result["missing_by_category"][cat] if f not in missing_required]
|
||
if optional:
|
||
print(f" [{cat}]: {', '.join(optional)}")
|
||
if verbose and result["extra_fields"]:
|
||
extra = result["extra_fields"]
|
||
print(f"\n[信息] 额外字段 ({len(extra)}):")
|
||
print(f" {', '.join(extra[:10])}")
|
||
if len(extra) > 10:
|
||
print(f" ... 还有 {len(extra) - 10} 个")
|
||
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="验证JSON文件是否覆盖fields.yaml中定义的所有字段")
|
||
parser.add_argument("--fields", "-f", type=str, help="fields.yaml路径", default="fields.yaml")
|
||
parser.add_argument("--json", "-j", type=str, nargs="*", help="要验证的JSON文件路径")
|
||
parser.add_argument("--dir", "-d", type=str, help="包含JSON文件的目录", default="results")
|
||
parser.add_argument("--quiet", "-q", action="store_true", help="仅显示摘要")
|
||
args = parser.parse_args()
|
||
fields_path = Path(args.fields)
|
||
if not fields_path.exists():
|
||
for p in (Path.cwd() / "fields.yaml", Path.cwd().parent / "fields.yaml"):
|
||
if p.exists():
|
||
fields_path = p
|
||
break
|
||
if not fields_path.exists():
|
||
print(f"[错误] 找不到fields.yaml: {fields_path}")
|
||
sys.exit(1)
|
||
print(f"字段定义文件: {fields_path}")
|
||
all_fields, required_fields, field_categories = load_fields_yaml(fields_path)
|
||
print(f"总字段数: {len(all_fields)} (必填: {len(required_fields)}, 可选: {len(all_fields) - len(required_fields)})")
|
||
json_files = (
|
||
[Path(p) for p in args.json]
|
||
if args.json
|
||
else sorted(Path(args.dir).glob("*.json")) if Path(args.dir).exists() else []
|
||
)
|
||
if not json_files:
|
||
print("[警告] 未找到JSON文件")
|
||
sys.exit(0)
|
||
results = []
|
||
for json_path in json_files:
|
||
if not json_path.exists():
|
||
print(f"[警告] 文件不存在: {json_path}")
|
||
continue
|
||
result = validate_json(json_path, all_fields, required_fields, field_categories)
|
||
results.append(result)
|
||
print_result(result, verbose=not args.quiet)
|
||
line = "=" * 60
|
||
print(f"\n{line}")
|
||
print("汇总")
|
||
print(line)
|
||
passed = sum(1 for r in results if r["valid"])
|
||
avg_coverage = sum(r["coverage_rate"] for r in results) / len(results) if results else 0
|
||
print(f"验证通过: {passed}/{len(results)}")
|
||
print(f"平均覆盖率: {avg_coverage:.1f}%")
|
||
if passed < len(results):
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|