#!/usr/bin/env python3 """ 覆盖率审计脚本 检查： 1. 二期ETL输出的字段名称，赋分引擎实际用到了多少 2. 赋分引擎里按关键词匹配的字段，在二期数据中能否命中 3. 每个维度的赋分数据源覆盖情况 """ import sys import re import inspect from pathlib import Path from collections import defaultdict import pandas as pd import numpy as np sys.path.insert(0, str(Path(__file__).parent)) sys.path.insert(0, str(Path(__file__).parent.parent.parent / "backend")) from data_engine_era2 import DataEngineEra2 from app.engines.scoring_engine import ScoringEngine from config_era2 import DIMENSION_FRAMEWORK, SUBJECTS def section(title): print(f"\n{'=' * 80}") print(f" {title}") print(f"{'=' * 80}") def main(): # ===== 加载二期全量数据（不分区，看全貌） ===== section("1. 加载二期全量数据") engine = DataEngineEra2(district_filter=None) engine.load_all() print(f" 学校数: {len(engine.schools)}") # 获取全量字段名称 course_df = engine._course_impl subject_df = engine._subject_impl basic_df = engine._basic_info course_fields = sorted(course_df["字段名称"].dropna().unique()) subject_fields = sorted(subject_df["字段名称"].dropna().unique()) basic_fields = sorted(basic_df["字段名称"].dropna().unique()) print(f" 课程实施表字段名称数: {len(course_fields)}") print(f" 学科课程表字段名称数: {len(subject_fields)}") print(f" 基础信息表字段名称数: {len(basic_fields)}") # ===== 2. 提取赋分引擎中所有用到的字段名称和关键词 ===== section("2. 赋分引擎字段引用分析") # 从 ScoringEngine 源码中提取所有字符串常量（字段名称/关键词） src = inspect.getsource(ScoringEngine) # 提取所有中文字符串（字段名称） # 匹配双引号和单引号中的中文字符串 field_refs = set() keyword_refs = set() # 精确字段引用：df["字段名称"] == "xxx" 或 字段名称 == field exact_patterns = re.findall(r'["\']([^"\']*[\u4e00-\u9fff][^"\']*)["\']', src) for p in exact_patterns: # 跳过注释性文字 if len(p) > 30 or '：' in p or '。' in p or '赋分' in p: continue field_refs.add(p) # 关键词引用：str.contains("xxx") contains_patterns = re.findall(r'\.str\.contains\(["\']([^"\']+)["\']', src) for p in contains_patterns: # 这些是用 | 分隔的关键词 for kw in p.split('|'): keyword_refs.add(kw.strip()) # 其他关键词引用（in循环中的列表） keyword_lists = re.findall(r'for (?:keyword|field|hw_type|resource) in \[([^\]]+)\]', src) for kl in keyword_lists: items = re.findall(r'["\']([^"\']+)["\']', kl) for item in items: if any('\u4e00' <= c <= '\u9fff' for c in item): field_refs.add(item) print(f"\n 赋分引擎中精确引用的字段名称: {len(field_refs)}个") print(f" 赋分引擎中关键词引用: {len(keyword_refs)}个") # ===== 3. 逐一检查精确字段在二期数据中的命中情况 ===== section("3. 精确字段匹配检查") all_data_fields = set(course_fields) | set(subject_fields) | set(basic_fields) matched = [] missing = [] for f in sorted(field_refs): if f in all_data_fields: matched.append(f) else: missing.append(f) print(f"\n ✅ 命中: {len(matched)}/{len(field_refs)}") print(f" ❌ 未命中: {len(missing)}/{len(field_refs)}") if missing: print(f"\n 未命中的字段（赋分引擎引用但二期数据中不存在）:") for f in missing: # 尝试模糊匹配 fuzzy = [df for df in all_data_fields if f.replace('_', '') in df.replace('_', '') or df.replace('_', '') in f.replace('_', '')] if fuzzy: print(f" ❌ {f}") print(f" → 可能对应: {fuzzy[:3]}") else: print(f" ❌ {f} (无近似匹配)") # ===== 4. 关键词匹配检查 ===== section("4. 关键词匹配检查") for kw in sorted(keyword_refs): course_hits = course_df[course_df["字段名称"].str.contains(kw, na=False)]["字段名称"].unique() subject_hits = subject_df[subject_df["字段名称"].str.contains(kw, na=False)]["字段名称"].unique() total = len(course_hits) + len(subject_hits) status = "✅" if total > 0 else "❌" print(f" {status} '{kw}': 课程表{len(course_hits)}个, 学科表{len(subject_hits)}个") if total == 0: # 看看有没有相近的 all_names = list(course_fields) + list(subject_fields) similar = [n for n in all_names if kw[:2] in n][:3] if similar: print(f" → 近似: {similar}") # ===== 5. 按维度逐一检查赋分数据覆盖 ===== section("5. 按维度检查赋分数据覆盖（抽样一所学校）") # 取一所数据较完整的学校 test_school = engine.schools[0] print(f" 测试学校: {test_school}") scoring = ScoringEngine(engine) scores = scoring._score_school(test_school) print(f"\n {'维度':<20} {'赋分项数':>8} {'均值':>8} {'是否有效':>8}") print(f" {'─' * 50}") for dim, vals in scores.items(): n = len(vals) avg = np.mean(vals) if vals else 0 # 判断是否有效：是否全是默认值 is_default = (n <= 1 and abs(avg - 1.0) < 0.01) or (n <= 1 and abs(avg - 1.5) < 0.01) or (n <= 1 and abs(avg - 2.0) < 0.01) or (n <= 1 and abs(avg - 0.5) < 0.01) status = "⚠️ 默认值" if is_default else "✅" print(f" {dim:<20} {n:>8} {avg:>8.2f} {status:>8}") # ===== 6. 多校抽样统计 ===== section("6. 多校统计：各维度赋分项数分布") # 取前20所学校统计 sample_schools = engine.schools[:20] dim_stats = defaultdict(list) for school in sample_schools: s = scoring._score_school(school) for dim, vals in s.items(): dim_stats[dim].append(len(vals)) print(f"\n 抽样学校数: {len(sample_schools)}") print(f"\n {'维度':<20} {'最小':>6} {'最大':>6} {'均值':>6} {'全为1':>8}") print(f" {'─' * 50}") for dim in DIMENSION_FRAMEWORK: for sub in DIMENSION_FRAMEWORK[dim]["sub_dimensions"]: vals = dim_stats.get(sub, [0]) min_v = min(vals) max_v = max(vals) avg_v = np.mean(vals) all_one = sum(1 for v in vals if v <= 1) warn = "⚠️" if all_one > len(vals) * 0.5 else "" print(f" {sub:<20} {min_v:>6} {max_v:>6} {avg_v:>6.1f} {all_one:>4}/{len(vals)} {warn}") # ===== 7. 二期新增但赋分引擎未使用的字段 ===== section("7. 二期数据中存在但赋分引擎未引用的高频字段（Top 30）") # 统计二期中每个字段名称出现的学校数 course_field_school_count = course_df.groupby("字段名称")["学校名称"].nunique().sort_values(ascending=False) subject_field_school_count = subject_df.groupby("字段名称")["学校名称"].nunique().sort_values(ascending=False) # 过滤掉已被赋分引擎引用的 unused_course = course_field_school_count[~course_field_school_count.index.isin(field_refs)] unused_subject = subject_field_school_count[~subject_field_school_count.index.isin(field_refs)] # 进一步过滤：去掉被关键词匹配可能命中的 def is_keyword_matched(field_name): for kw in keyword_refs: if kw in str(field_name): return True return False unused_course_strict = unused_course[~unused_course.index.map(is_keyword_matched)] unused_subject_strict = unused_subject[~unused_subject.index.map(is_keyword_matched)] print(f"\n [课程实施表] 未被引用的字段 (按学校覆盖率排序, Top 20):") for field, cnt in unused_course_strict.head(20).items(): print(f" {field}: {cnt}所学校有数据") print(f"\n [学科课程表] 未被引用的字段 (Top 20):") for field, cnt in unused_subject_strict.head(20).items(): print(f" {field}: {cnt}所学校有数据") print(f"\n✅ 覆盖率审计完成!") if __name__ == "__main__": main()