#!/usr/bin/env python3 """ 二期 Pipeline 测试脚本 数据加载 → PCA赋分(SPSS对齐) → 标准化 → 水平判定 → 聚类 → 统计分析 不调用 LLM，不生成 HTML 报告 支持两种模式： --district 长宁区 只分析长宁区8所学校（与一期对比验证） --district all 分析全部13个区176所学校 """ import sys import time import argparse import json import logging from pathlib import Path import numpy as np # 添加当前目录到 path（加载 config_era2 和 data_engine_era2） sys.path.insert(0, str(Path(__file__).parent)) from data_engine_era2 import DataEngineEra2 from engines.pca_scoring_engine_era2 import PcaScoringEngineEra2 from engines.stats_engine_era2 import StatsEngineEra2 as StatsEngine from config_era2 import DIMENSION_FRAMEWORK, LEVEL_THRESHOLDS logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger(__name__) def clean_for_json(obj): """处理 numpy 类型""" if isinstance(obj, dict): return {k: clean_for_json(v) for k, v in obj.items()} elif isinstance(obj, list): return [clean_for_json(v) for v in obj] elif isinstance(obj, (np.integer,)): return int(obj) elif isinstance(obj, (np.floating,)): return round(float(obj), 4) elif isinstance(obj, np.ndarray): return obj.tolist() elif isinstance(obj, float): return round(obj, 4) return obj def main(): parser = argparse.ArgumentParser(description="二期数据Pipeline测试") parser.add_argument("--district", type=str, default="长宁区", help="区域筛选，'all'表示全部区域，默认'长宁区'") parser.add_argument("--school", type=str, default=None, help="指定一所学校查看详情") parser.add_argument("--list-schools", action="store_true", help="列出所有可用学校") args = parser.parse_args() start_time = time.time() print("=" * 70) print("📊 二期数据 Pipeline 测试") print("=" * 70) # ===== Step 1: 加载数据 ===== district = None if args.district == "all" else args.district logger.info(f"[1/5] 加载二期数据... (区域: {args.district})") data_engine = DataEngineEra2(district_filter=district) data_engine.load_all() summary = data_engine.summary() print(f"\n📋 数据摘要:") print(f" 学校数: {summary['school_count']}") print(f" 基础信息行数: {summary['basic_info_rows']}") print(f" 课程实施行数: {summary['course_impl_rows']}") print(f" 学科课程行数: {summary['subject_impl_rows']}") if args.list_schools: print(f"\n可用学校 ({len(data_engine.schools)}所):") for i, s in enumerate(data_engine.schools, 1): print(f" {i:3d}. {s}") return # ===== Step 2: PCA赋分（SPSS对齐） ===== logger.info(f"[2/5] PCA赋分计算 ({len(data_engine.schools)}所学校, SPSS对齐)...") pca_engine = PcaScoringEngineEra2(data_engine) pca_sub_scores = pca_engine.compute_all() # 打印PCA赋分摘要 print(f"\n📊 PCA赋分结果摘要 (前5所学校):") for school in list(data_engine.schools)[:5]: if school in pca_sub_scores.index: row = pca_sub_scores.loc[school] vals = [f"{c}:{row[c]:.1f}" for c in row.index[:4] if not np.isnan(row[c])] print(f" {school}: {', '.join(vals)}...") # ===== Step 3: 标准化 ===== logger.info("[3/5] 全市基准标准化...") stats_engine = StatsEngine() sub_scores = stats_engine.compute_dimension_scores_pca(pca_sub_scores) dim_scores = stats_engine.compute_dimension_aggregates(sub_scores) print(f"\n📊 三级维度标准化得分 (均值50, 标准差10):") print(f" 形状: {sub_scores.shape}") print(f" 均值: {sub_scores.mean().mean():.2f}") print(f" 标准差: {sub_scores.std().mean():.2f}") print(f"\n📊 二级维度得分 (前5所学校):") print(dim_scores.head().round(2).to_string()) # ===== Step 4: 水平判定 ===== logger.info("[4/5] 水平判定...") levels = stats_engine.compute_levels(sub_scores) print(f"\n📊 水平分布统计:") for dim in levels.columns: dist = levels[dim].value_counts().sort_index() dist_str = " ".join([f"水平{k}:{v}所" for k, v in dist.items()]) print(f" {dim}: {dist_str}") # ===== Step 5: 聚类 + 相关性 ===== logger.info("[5/5] 聚类分析 + 相关性...") dim_cols = [c for c in dim_scores.columns if c in DIMENSION_FRAMEWORK] clusters = stats_engine.cluster_analysis(dim_scores[dim_cols]) cluster_dist = {} for school, cluster in clusters["school_clusters"].items(): cluster_dist[cluster] = cluster_dist.get(cluster, 0) + 1 print(f"\n📊 聚类分布: {cluster_dist}") corr = stats_engine.correlation_analysis(dim_scores) print(f"\n📊 维度间相关性:") print(corr.round(3).to_string()) # ===== 单校详情 ===== test_school = args.school or (data_engine.schools[0] if data_engine.schools else None) if test_school and test_school in data_engine.schools: print(f"\n{'─' * 70}") print(f"🏫 {test_school} 详细报告数据") print(f"{'─' * 70}") report_data = stats_engine.compute_school_report_data(test_school, sub_scores, dim_scores) print(f" 总体得分: {report_data['overall']['score']}") print(f" 排名: {report_data['overall']['rank_in_district']}/{report_data['overall']['total_schools']}") print(f" 聚类类型: {report_data['overall']['cluster']}") print(f"\n 二级维度:") for dim, data in report_data['dimensions'].items(): print(f" {dim}: {data['score']} (区均{data['district_avg']}, 差{data['diff_district']:+.2f}) [{data['cluster']}]") print(f"\n 三级维度:") for sub_dim, data in report_data['sub_dimensions'].items(): print(f" {sub_dim}: {data['score']} 水平{data['level']} (区均{data['district_avg']})") # 保存报告数据 output_dir = Path(__file__).parent.parent.parent / "output" / "era2" output_dir.mkdir(parents=True, exist_ok=True) report_clean = clean_for_json(report_data) json_path = output_dir / f"{test_school}_report_data.json" with open(json_path, "w", encoding="utf-8") as f: json.dump(report_clean, f, ensure_ascii=False, indent=2) print(f"\n ✅ 报告数据 → {json_path}") # ===== 保存全量得分 ===== output_dir = Path(__file__).parent.parent.parent / "output" / "era2" output_dir.mkdir(parents=True, exist_ok=True) all_data = { "district_filter": args.district, "school_count": len(data_engine.schools), "sub_dimension_scores": clean_for_json(sub_scores.to_dict()), "dimension_scores": clean_for_json(dim_scores.to_dict()), "levels": clean_for_json(levels.to_dict()), "clusters": clean_for_json(clusters), } scores_path = output_dir / f"all_scores_{args.district}.json" with open(scores_path, "w", encoding="utf-8") as f: json.dump(all_data, f, ensure_ascii=False, indent=2) elapsed = time.time() - start_time print(f"\n{'=' * 70}") print(f"✅ Pipeline测试完成! 耗时 {elapsed:.1f}s") print(f" 区域: {args.district}") print(f" 学校数: {len(data_engine.schools)}") print(f" 得分文件: {scores_path}") print(f"{'=' * 70}") if __name__ == "__main__": main()