diff --git a/skills/research-en/validate_json.py b/skills/research-en/validate_json.py new file mode 100644 index 0000000..c2e4ed0 --- /dev/null +++ b/skills/research-en/validate_json.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +JSON Field Validation Script +Validates whether JSON files completely cover all fields defined in fields.yaml +""" + +import json +import yaml +import sys +from pathlib import Path +from typing import Dict, List, Set, Any, Tuple + +# Category mapping (supports both Chinese and English keys) +CATEGORY_MAPPING = { + "basic_info": ["basic_info", "基本信息"], + "technical_features": ["technical_features", "technical_characteristics", "技术特性"], + "performance_metrics": ["performance_metrics", "performance", "性能指标"], + "milestone_significance": ["milestone_significance", "milestones", "里程碑意义"], + "business_info": ["business_info", "commercial_info", "商业信息"], + "competition_ecosystem": ["competition_ecosystem", "competition", "竞争与生态"], + "history": ["history", "历史沿革"], + "market_positioning": ["market_positioning", "market", "市场定位"], +} + + +def load_fields_yaml(fields_path: Path) -> Tuple[Set[str], Set[str], Dict[str, str]]: + """ + Load fields.yaml and return: + - all_fields: set of all field names + - required_fields: set of field names where required=true + - field_categories: mapping from field name to category + """ + with open(fields_path, 'r', encoding='utf-8') as f: + data = yaml.safe_load(f) + + all_fields = set() + required_fields = set() + field_categories = {} + + for category_info in data.get("field_categories", []): + category_name = category_info["category"] + for field in category_info.get("fields", []): + field_name = field["name"] + all_fields.add(field_name) + field_categories[field_name] = category_name + if field.get("required", False): + required_fields.add(field_name) + + return all_fields, required_fields, field_categories + + +def extract_json_fields(data: Dict, category_mapping: Dict = None) -> Set[str]: + """ + Extract all field names from JSON (supports both flat and nested structures) + """ + if category_mapping is None: + category_mapping = CATEGORY_MAPPING + + # Get all possible nested keys + nested_keys = set() + for keys in category_mapping.values(): + nested_keys.update(keys) + + fields = set() + + def collect_fields(d: Dict, is_top_level: bool = True): + for k, v in d.items(): + # Skip internal fields + if k in {"_source_file", "uncertain"}: + continue + # If it's a top-level nested key, recurse into it + if is_top_level and k in nested_keys: + if isinstance(v, dict): + collect_fields(v, is_top_level=False) + else: + fields.add(k) + if isinstance(v, dict): + collect_fields(v, is_top_level=False) + + collect_fields(data) + return fields + + +def validate_json(json_path: Path, all_fields: Set[str], required_fields: Set[str], + field_categories: Dict[str, str]) -> Dict: + """ + Validate a single JSON file + Returns validation result dictionary + """ + with open(json_path, 'r', encoding='utf-8') as f: + data = json.load(f) + + json_fields = extract_json_fields(data) + + # Calculate coverage + covered = all_fields & json_fields + missing = all_fields - json_fields + extra = json_fields - all_fields + + # Categorize missing fields + missing_required = missing & required_fields + missing_optional = missing - required_fields + + # Group missing fields by category + missing_by_category = {} + for field in missing: + cat = field_categories.get(field, "Unknown") + if cat not in missing_by_category: + missing_by_category[cat] = [] + missing_by_category[cat].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": list(missing_required), + "missing_optional": list(missing_optional), + "missing_by_category": missing_by_category, + "extra_fields": list(extra), + "valid": len(missing_required) == 0, # Valid if all required fields are covered + } + + +def print_result(result: Dict, verbose: bool = True): + """Print validation result""" + status = "PASS" if result["valid"] else "FAIL" + print(f"\n{'='*60}") + print(f"[{status}] {result['file']}") + print(f"{'='*60}") + print(f"Coverage: {result['coverage_rate']:.1f}% ({result['covered']}/{result['total_defined']})") + + if result["missing_required"]: + print(f"\n[ERROR] Missing required fields ({len(result['missing_required'])}):") + for field in result["missing_required"]: + print(f" - {field}") + + if verbose and result["missing_optional"]: + print(f"\n[WARN] Missing optional fields ({len(result['missing_optional'])}):") + for cat, fields in result["missing_by_category"].items(): + optional_fields = [f for f in fields if f not in result["missing_required"]] + if optional_fields: + print(f" [{cat}]: {', '.join(optional_fields)}") + + if verbose and result["extra_fields"]: + print(f"\n[INFO] Extra fields ({len(result['extra_fields'])}):") + print(f" {', '.join(result['extra_fields'][:10])}") + if len(result["extra_fields"]) > 10: + print(f" ... and {len(result['extra_fields']) - 10} more") + + +def main(): + """Main function""" + import argparse + parser = argparse.ArgumentParser(description="Validate whether JSON files cover all fields defined in fields.yaml") + parser.add_argument("--fields", "-f", type=str, help="Path to fields.yaml", default="fields.yaml") + parser.add_argument("--json", "-j", type=str, nargs="*", help="JSON file paths to validate") + parser.add_argument("--dir", "-d", type=str, help="Directory containing JSON files", default="results") + parser.add_argument("--quiet", "-q", action="store_true", help="Show summary only") + args = parser.parse_args() + + # Locate fields.yaml + fields_path = Path(args.fields) + if not fields_path.exists(): + # Try to find in current and parent directory + 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"[ERROR] fields.yaml not found: {fields_path}") + sys.exit(1) + + print(f"Field definition file: {fields_path}") + all_fields, required_fields, field_categories = load_fields_yaml(fields_path) + print(f"Total fields: {len(all_fields)} (required: {len(required_fields)}, optional: {len(all_fields) - len(required_fields)})") + + # Collect JSON files + json_files = [] + if args.json: + json_files = [Path(p) for p in args.json] + else: + json_dir = Path(args.dir) + if json_dir.exists(): + json_files = sorted(json_dir.glob("*.json")) + + if not json_files: + print(f"[WARN] No JSON files found") + sys.exit(0) + + # Validate each file + results = [] + for json_path in json_files: + if not json_path.exists(): + print(f"[WARN] File not found: {json_path}") + continue + result = validate_json(json_path, all_fields, required_fields, field_categories) + results.append(result) + print_result(result, verbose=not args.quiet) + + # Summary + print(f"\n{'='*60}") + print("Summary") + print(f"{'='*60}") + 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"Validation passed: {passed}/{len(results)}") + print(f"Average coverage: {avg_coverage:.1f}%") + + if passed < len(results): + sys.exit(1) + + +if __name__ == "__main__": + main()