feat: enhance research workflow with user confirmations
- Add AskUserQuestion for Step 1 framework confirmation - Add TOC summary field selection in report generation - Add detail_level hierarchy (brief -> moderate -> detailed) - Add Chinese output requirement for JSON values - Add long text formatting rules 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jilinchen
co-authored by
Claude Opus 4.5
Claude Opus 4.5
parent
5eaf859274
commit
11a51a8d79
@@ -1,8 +1,8 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
JSON Field Validation Script
|
||||
Validate if JSON files completely cover all fields defined in fields.yaml
|
||||
JSON字段验证脚本
|
||||
验证JSON文件是否完整覆盖fields.yaml中定义的所有字段
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -11,25 +11,25 @@ import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Set, Any, Tuple
|
||||
|
||||
# Category mapping (supports both Chinese and English)
|
||||
# Category中英文映射
|
||||
CATEGORY_MAPPING = {
|
||||
"Basic Info": ["basic_info", "Basic Info"],
|
||||
"Technical Features": ["technical_features", "technical_characteristics", "Technical Features"],
|
||||
"Performance Metrics": ["performance_metrics", "performance", "Performance Metrics"],
|
||||
"Milestone Significance": ["milestone_significance", "milestones", "Milestone Significance"],
|
||||
"Business Info": ["business_info", "commercial_info", "Business Info"],
|
||||
"Competition & Ecosystem": ["competition_ecosystem", "competition", "Competition & Ecosystem"],
|
||||
"History": ["history", "History"],
|
||||
"Market Positioning": ["market_positioning", "market", "Market Positioning"],
|
||||
"基本信息": ["basic_info", "基本信息"],
|
||||
"技术特性": ["technical_features", "technical_characteristics", "技术特性"],
|
||||
"性能指标": ["performance_metrics", "performance", "性能指标"],
|
||||
"里程碑意义": ["milestone_significance", "milestones", "里程碑意义"],
|
||||
"商业信息": ["business_info", "commercial_info", "商业信息"],
|
||||
"竞争与生态": ["competition_ecosystem", "competition", "竞争与生态"],
|
||||
"历史沿革": ["history", "历史沿革"],
|
||||
"市场定位": ["market_positioning", "market", "市场定位"],
|
||||
}
|
||||
|
||||
|
||||
def load_fields_yaml(fields_path: Path) -> Tuple[Set[str], Set[str], Dict[str, str]]:
|
||||
"""
|
||||
Load fields.yaml, returns:
|
||||
- all_fields: Set of all field names
|
||||
- required_fields: Set of field names where required=true
|
||||
- field_categories: Mapping from field name to category
|
||||
加载fields.yaml，返回：
|
||||
- all_fields: 所有字段名集合
|
||||
- required_fields: required=true的字段名集合
|
||||
- field_categories: 字段名到类别的映射
|
||||
"""
|
||||
with open(fields_path, 'r', encoding='utf-8') as f:
|
||||
data = yaml.safe_load(f)
|
||||
@@ -52,12 +52,12 @@ def load_fields_yaml(fields_path: Path) -> Tuple[Set[str], Set[str], Dict[str, s
|
||||
|
||||
def extract_json_fields(data: Dict, category_mapping: Dict = None) -> Set[str]:
|
||||
"""
|
||||
Extract all field names from JSON (supports flat and nested structures)
|
||||
从JSON中提取所有字段名（支持扁平和嵌套结构）
|
||||
"""
|
||||
if category_mapping is None:
|
||||
category_mapping = CATEGORY_MAPPING
|
||||
|
||||
# Get all possible nested keys
|
||||
# 获取所有可能的嵌套key
|
||||
nested_keys = set()
|
||||
for keys in category_mapping.values():
|
||||
nested_keys.update(keys)
|
||||
@@ -66,10 +66,10 @@ def extract_json_fields(data: Dict, category_mapping: Dict = None) -> Set[str]:
|
||||
|
||||
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 nested structure top-level key, recurse into it
|
||||
# 如果是嵌套结构的顶级key，递归进入
|
||||
if is_top_level and k in nested_keys:
|
||||
if isinstance(v, dict):
|
||||
collect_fields(v, is_top_level=False)
|
||||
@@ -85,27 +85,27 @@ def extract_json_fields(data: Dict, category_mapping: Dict = None) -> Set[str]:
|
||||
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
|
||||
验证单个JSON文件
|
||||
返回验证结果字典
|
||||
"""
|
||||
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")
|
||||
cat = field_categories.get(field, "未知")
|
||||
if cat not in missing_by_category:
|
||||
missing_by_category[cat] = []
|
||||
missing_by_category[cat].append(field)
|
||||
@@ -121,65 +121,65 @@ def validate_json(json_path: Path, all_fields: Set[str], required_fields: Set[st
|
||||
"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
|
||||
"valid": len(missing_required) == 0, # required字段全覆盖则valid
|
||||
}
|
||||
|
||||
|
||||
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']})")
|
||||
print(f"覆盖率: {result['coverage_rate']:.1f}% ({result['covered']}/{result['total_defined']})")
|
||||
|
||||
if result["missing_required"]:
|
||||
print(f"\n[ERROR] Missing required fields ({len(result['missing_required'])}):")
|
||||
print(f"\n[ERROR] 缺失必需字段 ({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'])}):")
|
||||
print(f"\n[WARN] 缺失可选字段 ({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"\n[INFO] 额外字段 ({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 fields")
|
||||
print(f" ... 及 {len(result['extra_fields']) - 10} 个其他字段")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function"""
|
||||
"""主函数"""
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description="Validate if 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="JSON files directory", default="results")
|
||||
parser.add_argument("--quiet", "-q", action="store_true", help="Show summary only")
|
||||
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()
|
||||
|
||||
# Locate fields.yaml
|
||||
# 定位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}")
|
||||
print(f"[ERROR] fields.yaml不存在: {fields_path}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Field definitions file: {fields_path}")
|
||||
print(f"字段定义文件: {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)})")
|
||||
print(f"总字段数: {len(all_fields)} (必需: {len(required_fields)}, 可选: {len(all_fields) - len(required_fields)})")
|
||||
|
||||
# Collect JSON files
|
||||
# 收集JSON文件
|
||||
json_files = []
|
||||
if args.json:
|
||||
json_files = [Path(p) for p in args.json]
|
||||
@@ -189,27 +189,27 @@ def main():
|
||||
json_files = sorted(json_dir.glob("*.json"))
|
||||
|
||||
if not json_files:
|
||||
print(f"[WARN] No JSON files found")
|
||||
print(f"[WARN] 未找到JSON文件")
|
||||
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}")
|
||||
print(f"[WARN] 文件不存在: {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("汇总")
|
||||
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}%")
|
||||
print(f"验证通过: {passed}/{len(results)}")
|
||||
print(f"平均覆盖率: {avg_coverage:.1f}%")
|
||||
|
||||
if passed < len(results):
|
||||
sys.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user