report-admin/scripts/era2/engines/trace_engine.py

"""
数据溯源引擎 — 生成单校从原始数据到最终得分的完整计算链路

设计原则：
  - 不侵入现有PCA引擎，独立读取数据并重建中间过程
  - 输出JSON结构，前端R3F组件直接消费
  - 支持缓存（同一学校的trace数据不会频繁变化）
"""
import json
import hashlib
import logging
import numpy as np
import pandas as pd
from pathlib import Path
from typing import Dict, List, Optional, Any
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler

import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from config_era2 import (
    PCA_MEAN, PCA_STD, DIMENSION_FRAMEWORK, LEVEL_THRESHOLDS,
    LEVEL_DESCRIPTIONS, SUBJECTS, SCHOOL_TYPE_MAP,
)

logger = logging.getLogger(__name__)

# 缓存目录
PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
TRACE_CACHE_DIR = PROJECT_ROOT / "output" / "trace_cache"


class TraceEngine:
    """数据溯源引擎：为单校生成完整的6阶段计算链路"""

    def __init__(self, data_engine, pca_engine, stats_engine,
                 sub_scores: pd.DataFrame, dim_scores: pd.DataFrame):
        self.de = data_engine
        self.pca = pca_engine
        self.stats = stats_engine
        self.sub_scores = sub_scores
        self.dim_scores = dim_scores

    def compute_trace(self, school: str, district_schools: List[str],
                      use_cache: bool = True) -> Dict:
        """
        主入口：生成单校的完整计算链路

        Returns:
            {school, school_info, stages: {raw_data, scoring, pca_detail, standardized, levels, dimensions, overall}}
        """
        # 缓存
        if use_cache:
            cached = self._load_cache(school)
            if cached is not None:
                return cached

        logger.info(f"[Trace] 生成 {school} 的计算链路...")

        all_schools = list(self.sub_scores.index)
        dist_schools = [s for s in district_schools if s in all_schools]

        # 阶段0：原始数据概览
        stage_raw = self._trace_raw_data(school)

        # 阶段1：赋分过程 + 阶段2: PCA细节
        stage_scoring, stage_pca = self._trace_scoring_and_pca(school, all_schools)

        # 阶段3：标准化后得分
        stage_std = self._trace_standardized(school, dist_schools)

        # 阶段4：水平判定
        stage_levels = self._trace_levels(school)

        # 阶段5：维度聚合 → 总分
        stage_dims, stage_overall = self._trace_dimensions(school, dist_schools)

        # 全市对比数据（用于标准化分布可视化）
        all_schools_overall = {}
        for s in dist_schools:
            if s in self.dim_scores.index and "总体得分" in self.dim_scores.columns:
                all_schools_overall[s] = round(float(self.dim_scores.loc[s, "总体得分"]), 2)

        result = {
            "school": school,
            "school_info": SCHOOL_TYPE_MAP.get(school, {}),
            "district_school_count": len(dist_schools),
            "city_school_count": len(all_schools),
            "stages": {
                "raw_data": stage_raw,
                "scoring": stage_scoring,
                "pca_detail": stage_pca,
                "standardized": stage_std,
                "levels": stage_levels,
                "dimensions": stage_dims,
                "overall": stage_overall,
            },
            "all_schools_overall": all_schools_overall,
        }

        self._save_cache(school, result)
        return result

    # ====================================================================
    # 阶段0：原始数据概览
    # ====================================================================

    def _trace_raw_data(self, school: str) -> Dict:
        """展示该校的原始数据概况"""
        # B表数据
        course_df = self._get_course(school)
        b_fields = len(course_df)
        b_sample = []
        if len(course_df) > 0:
            sample_rows = course_df.head(12)
            for _, r in sample_rows.iterrows():
                b_sample.append({
                    "name": str(r.get("字段名称", "")),
                    "value": _safe_value(r.get("字段取值", "")),
                    "type": _guess_type(r.get("字段取值", "")),
                })

        # C表数据
        c_total = 0
        c_subjects = []
        for subj in SUBJECTS:
            sub_df = self._get_subject(school, subj)
            count = len(sub_df)
            c_total += count
            if count > 0:
                c_subjects.append({"subject": subj, "field_count": count})

        c_sample = []
        # 取第一个有数据的学科的前几行
        for subj in SUBJECTS:
            sub_df = self._get_subject(school, subj)
            if len(sub_df) > 0:
                for _, r in sub_df.head(8).iterrows():
                    c_sample.append({
                        "subject": subj,
                        "name": str(r.get("字段名称", "")),
                        "value": _safe_value(r.get("字段取值", "")),
                        "type": _guess_type(r.get("字段取值", "")),
                    })
                break

        return {
            "b_table": {
                "field_count": b_fields,
                "sample_fields": b_sample,
            },
            "c_table": {
                "field_count": c_total,
                "subject_count": len(c_subjects),
                "subjects": c_subjects,
                "sample_fields": c_sample,
            },
            "total_fields": b_fields + c_total,
        }

    # ====================================================================
    # 阶段1 & 2：赋分 + PCA
    # ====================================================================

    def _trace_scoring_and_pca(self, school: str, all_schools: List[str]) -> tuple:
        """
        为每个子维度重建赋分过程和PCA细节。
        
        策略：对每个子维度，分别计算该子维度的赋分矩阵，
        记录该校的具体输入值、赋分规则和PCA参数。
        """
        scoring = {}
        pca_detail = {}

        # 按维度框架遍历每个子维度
        sub_dim_methods = {
            "国家标准遵循": self._trace_national_standard,
            "课程结构建设": self._trace_course_structure,
            "课程规范落实": self._trace_school_level_generic,
            "教学方式变革": self._trace_subject_level_generic,
            "作业设计与管理变革": self._trace_subject_level_generic,
            "学科发展的个性化辅导": self._trace_subject_level_generic,
            "学生生涯发展指导": self._trace_school_level_generic,
            "培训支持": self._trace_subject_level_generic,
            "教研支持": self._trace_subject_level_generic,
            "项目支持": self._trace_subject_level_generic,
            "科学评价观": self._trace_subject_level_generic,
            "学业质量评估": self._trace_subject_level_generic,
            "综合素质评估": self._trace_school_level_generic,
            "实践活动评估": self._trace_subject_level_generic,
            "区域推进": self._trace_school_level_generic,
            "环境支持": self._trace_school_level_generic,
            "资源支持": self._trace_school_level_generic,
            "教学方式创新": self._trace_subject_level_generic,
            "评价精准化与个性化": self._trace_school_level_generic,
            "课程迭代优化": self._trace_school_level_generic,
        }

        for dim_name, info in DIMENSION_FRAMEWORK.items():
            for sub_dim in info["sub_dimensions"]:
                method = sub_dim_methods.get(sub_dim, self._trace_generic_fallback)
                try:
                    s_info, p_info = method(sub_dim, school, all_schools)
                except Exception as e:
                    logger.warning(f"[Trace] {sub_dim} trace failed: {e}")
                    s_info = {"method": "unknown", "error": str(e)}
                    p_info = {}

                # 追加最终得分
                final_score = float(self.sub_scores.loc[school, sub_dim]) if sub_dim in self.sub_scores.columns else None
                s_info["final_score"] = round(final_score, 2) if final_score is not None else None
                s_info["parent_dimension"] = dim_name

                scoring[sub_dim] = s_info
                pca_detail[sub_dim] = p_info

        return scoring, pca_detail

    def _trace_national_standard(self, sub_dim: str, school: str,
                                  all_schools: List[str]) -> tuple:
        """国家标准遵循的特殊trace"""
        BXIU_STD = {"语文": 8, "数学": 8, "英语": 6, "思想政治": 6, "历史": 4, "地理": 4,
                    "物理": 6, "化学": 4, "生命科学": 4, "体育": 12, "技术": 6, "艺术": 6}
        MERGE_MAP = {
            "信息技术": "技术", "通用技术": "技术", "劳动技术": "技术",
            "音乐": "艺术", "美术": "艺术", "生物学": "生命科学", "体育与健康": "体育",
        }
        SPSS_12 = list(BXIU_STD.keys())

        # 获取该校课时
        hours_df = self.pca._get_weekly_hours(school)
        merged = {s: {"必修": 0, "选必": 0, "选修": 0} for s in SPSS_12}
        if len(hours_df) > 0:
            for _, r in hours_df.iterrows():
                subj = r.get("学科", "")
                mapped = MERGE_MAP.get(subj, subj)
                if mapped not in merged:
                    continue
                field = r["字段名称"]
                val = r["字段取值"]
                if pd.isna(val):
                    continue
                if "必修课周课时" in field and "选择性" not in field:
                    merged[mapped]["必修"] += val
                elif "选择性必修" in field:
                    merged[mapped]["选必"] += val
                elif "选修课周课时" in field:
                    merged[mapped]["选修"] += val

        # 必修分档评分
        inputs = []
        for s in SPSS_12:
            actual = merged[s]["必修"]
            std = BXIU_STD[s]
            if actual == 0:
                score = 0.0
                rule = "低于标准->0"
            elif abs(actual - std) <= 1.0:
                score = 2.0
                rule = "一致->2"
            elif actual > std:
                score = 1.0
                rule = "高于标准->1"
            else:
                score = 0.0
                rule = "低于标准->0"
            inputs.append({
                "name": f"{s}必修课时",
                "raw": actual,
                "standard": std,
                "score": score,
                "rule": rule,
            })

        total_xb = sum(merged[s]["选必"] for s in SPSS_12)
        total_xx = sum(merged[s]["选修"] for s in SPSS_12)

        scoring_info = {
            "method": "PCA(12学科必修分档) + Z(选必达标) + Z(选修达标) -> 均值",
            "inputs": inputs,
            "sub_factors": [
                {"name": "必修PCA", "input_count": 12, "type": "PCA"},
                {"name": "选必达标", "raw": total_xb, "threshold": 42,
                 "met": total_xb >= 42, "type": "Z-score"},
                {"name": "选修达标", "raw": total_xx, "threshold": 14,
                 "met": total_xx >= 14, "type": "Z-score"},
            ],
        }

        # PCA细节：构建全市必修矩阵
        bx_rows = {}
        for s in all_schools:
            h_df = self.pca._get_weekly_hours(s)
            m = {subj: {"必修": 0} for subj in SPSS_12}
            if len(h_df) > 0:
                for _, r in h_df.iterrows():
                    subj = r.get("学科", "")
                    mapped = MERGE_MAP.get(subj, subj)
                    if mapped not in m:
                        continue
                    field = r["字段名称"]
                    val = r["字段取值"]
                    if pd.isna(val):
                        continue
                    if "必修课周课时" in field and "选择性" not in field:
                        m[mapped]["必修"] += val
            row = {}
            for subj in SPSS_12:
                actual = m[subj]["必修"]
                std_val = BXIU_STD[subj]
                if actual == 0:
                    row[subj] = 0.0
                elif abs(actual - std_val) <= 1.0:
                    row[subj] = 2.0
                elif actual > std_val:
                    row[subj] = 1.0
                else:
                    row[subj] = 0.0
            bx_rows[s] = row

        matrix = pd.DataFrame(bx_rows).T
        pca_info = self._extract_pca_details(matrix, school, "必修课PCA")

        return scoring_info, pca_info

    def _trace_course_structure(self, sub_dim: str, school: str,
                                 all_schools: List[str]) -> tuple:
        """课程结构建设的trace（3组PCA）"""
        scoring_info = {
            "method": "PCA(学科课程结构) + PCA(校本特色) + PCA(综合实践) -> 均值",
            "sub_factors": [
                {"name": "学科类课程结构PCA", "type": "PCA",
                 "description": "必修/选必/选修课时比例偏离度"},
                {"name": "校本特色课程PCA", "type": "PCA",
                 "description": "选修课数量+时长"},
                {"name": "综合实践PCA", "type": "PCA",
                 "description": "党团次数+社考个数+志愿时长+劳动"},
            ],
        }
        # 简化的PCA info
        pca_info = {
            "type": "multi_factor",
            "factor_count": 3,
            "school_count": len(all_schools),
            "note": "三组因子各自做PCA后取均值",
        }
        return scoring_info, pca_info

    def _trace_subject_level_generic(self, sub_dim: str, school: str,
                                      all_schools: List[str]) -> tuple:
        """C表学科级子维度的通用trace"""
        # 获取该校在该子维度的最终得分
        scoring_info = {
            "method": "学科级赋分 -> 全市Z标准化 -> PCA(第一主成分) -> x10+50 -> 学科均值 -> 学校均值",
            "data_source": "C表（学科课程实施情况表）",
            "subject_count": len(SUBJECTS),
        }

        # 尝试获取该学科的一些原始数据作为示例
        sample_inputs = []
        for subj in SUBJECTS[:3]:  # 取前3个学科作示例
            sub_df = self._get_subject(school, subj)
            if len(sub_df) > 0:
                for _, r in sub_df.head(3).iterrows():
                    sample_inputs.append({
                        "subject": subj,
                        "name": str(r.get("字段名称", "")),
                        "value": _safe_value(r.get("字段取值", "")),
                    })
        scoring_info["sample_inputs"] = sample_inputs

        pca_info = {
            "type": "subject_level",
            "school_count": len(all_schools),
            "subject_count": len(SUBJECTS),
            "pipeline": "题目赋分 -> Z标准化 -> PCA -> x10+50",
        }
        return scoring_info, pca_info

    def _trace_school_level_generic(self, sub_dim: str, school: str,
                                     all_schools: List[str]) -> tuple:
        """B表学校级子维度的通用trace"""
        scoring_info = {
            "method": "学校级赋分 -> 全市Z标准化 -> PCA(第一主成分) -> x10+50",
            "data_source": "B表（课程实施情况表）",
        }

        # 取一些原始数据作示例
        course_df = self._get_course(school)
        sample_inputs = []
        if len(course_df) > 0:
            for _, r in course_df.head(6).iterrows():
                sample_inputs.append({
                    "name": str(r.get("字段名称", "")),
                    "value": _safe_value(r.get("字段取值", "")),
                })
        scoring_info["sample_inputs"] = sample_inputs

        pca_info = {
            "type": "school_level",
            "school_count": len(all_schools),
            "pipeline": "赋分 -> Z标准化 -> PCA -> x10+50",
        }
        return scoring_info, pca_info

    def _trace_generic_fallback(self, sub_dim: str, school: str,
                                 all_schools: List[str]) -> tuple:
        """回退方法"""
        return {"method": "unknown"}, {}

    # ====================================================================
    # PCA细节提取
    # ====================================================================

    def _extract_pca_details(self, matrix: pd.DataFrame, school: str,
                              label: str = "PCA") -> Dict:
        """从赋分矩阵中提取PCA的详细参数"""
        matrix = matrix.dropna(axis=1, how="all")
        if matrix.shape[1] == 0:
            return {"label": label, "error": "empty_matrix"}

        filled = matrix.copy().infer_objects(copy=False)
        for col in filled.columns:
            col_mean = filled[col].mean()
            if np.isnan(col_mean):
                col_mean = 0.0
            filled[col] = filled[col].fillna(col_mean)

        if filled.shape[1] == 1:
            vals = filled.iloc[:, 0].values.astype(float)
            school_idx = list(filled.index).index(school) if school in filled.index else -1
            return {
                "label": label,
                "type": "single_variable_z",
                "variable": str(filled.columns[0]),
                "school_value": round(float(vals[school_idx]), 4) if school_idx >= 0 else None,
                "mean": round(float(np.nanmean(vals)), 4),
                "std": round(float(np.nanstd(vals, ddof=1)), 4),
                "n_schools": len(vals),
            }

        # Z标准化
        scaler = StandardScaler()
        try:
            scaled = scaler.fit_transform(filled.values.astype(float))
        except ValueError:
            return {"label": label, "error": "scaling_failed"}

        # PCA
        pca = PCA(n_components=min(1, filled.shape[1], filled.shape[0]))
        scores = pca.fit_transform(scaled)[:, 0]
        loadings = pca.components_[0]

        if np.sum(loadings) < 0:
            scores = -scores
            loadings = -loadings

        mean = np.mean(scores)
        std = np.std(scores, ddof=1)

        school_idx = list(filled.index).index(school) if school in filled.index else -1
        school_raw_score = scores[school_idx] if school_idx >= 0 else None
        school_std_score = ((school_raw_score - mean) / std * PCA_STD + PCA_MEAN) if (school_raw_score is not None and std > 0) else None

        # Loadings详情
        loading_details = []
        for i, col in enumerate(filled.columns):
            loading_details.append({
                "variable": str(col),
                "loading": round(float(loadings[i]), 4),
            })
        loading_details.sort(key=lambda x: abs(x["loading"]), reverse=True)

        return {
            "label": label,
            "type": "pca",
            "n_schools": int(filled.shape[0]),
            "n_variables": int(filled.shape[1]),
            "explained_variance_ratio": round(float(pca.explained_variance_ratio_[0]), 4),
            "loadings": loading_details,
            "school_pca_score": round(float(school_raw_score), 4) if school_raw_score is not None else None,
            "school_standardized": round(float(school_std_score), 2) if school_std_score is not None else None,
            "pca_mean": round(float(mean), 4),
            "pca_std": round(float(std), 4),
        }

    # ====================================================================
    # 阶段3：标准化后得分
    # ====================================================================

    def _trace_standardized(self, school: str, dist_schools: List[str]) -> Dict:
        """标准化后的20个子维度得分"""
        result = {}
        for dim_name, info in DIMENSION_FRAMEWORK.items():
            for sub_dim in info["sub_dimensions"]:
                if sub_dim not in self.sub_scores.columns:
                    continue
                score = float(self.sub_scores.loc[school, sub_dim])
                # 区内均值和全市均值
                dist_vals = self.sub_scores.loc[
                    [s for s in dist_schools if s in self.sub_scores.index], sub_dim
                ]
                all_vals = self.sub_scores[sub_dim]

                result[sub_dim] = {
                    "score": round(score, 2),
                    "district_avg": round(float(dist_vals.mean()), 2),
                    "city_avg": round(float(all_vals.mean()), 2),
                    "city_std": round(float(all_vals.std()), 2),
                    "diff_district": round(score - float(dist_vals.mean()), 2),
                    "diff_city": round(score - float(all_vals.mean()), 2),
                    "parent_dimension": dim_name,
                }
        return result

    # ====================================================================
    # 阶段4：水平判定
    # ====================================================================

    def _trace_levels(self, school: str) -> Dict:
        result = {}
        for dim_name, info in DIMENSION_FRAMEWORK.items():
            for sub_dim in info["sub_dimensions"]:
                if sub_dim not in self.sub_scores.columns:
                    continue
                score = float(self.sub_scores.loc[school, sub_dim])
                thresholds = LEVEL_THRESHOLDS.get(sub_dim, {})
                level = self.stats.determine_level(score, sub_dim)
                desc = LEVEL_DESCRIPTIONS.get(sub_dim, {}).get(level, "")

                result[sub_dim] = {
                    "score": round(score, 2),
                    "thresholds": {
                        "level2": thresholds.get("level2", 43),
                        "level3": thresholds.get("level3", 50),
                        "level4": thresholds.get("level4", 57),
                    },
                    "level": level,
                    "description": desc,
                    "parent_dimension": dim_name,
                }
        return result

    # ====================================================================
    # 阶段5：维度聚合 → 总分
    # ====================================================================

    def _trace_dimensions(self, school: str, dist_schools: List[str]) -> tuple:
        dims = {}
        for dim_name, info in DIMENSION_FRAMEWORK.items():
            sub_dims = info["sub_dimensions"]
            sub_scores = {}
            for sd in sub_dims:
                if sd in self.sub_scores.columns:
                    sub_scores[sd] = round(float(self.sub_scores.loc[school, sd]), 2)

            dim_score = float(self.dim_scores.loc[school, dim_name]) if dim_name in self.dim_scores.columns else None
            dist_dim_vals = self.dim_scores.loc[
                [s for s in dist_schools if s in self.dim_scores.index], dim_name
            ] if dim_name in self.dim_scores.columns else pd.Series()

            dims[dim_name] = {
                "sub_scores": sub_scores,
                "score": round(dim_score, 2) if dim_score is not None else None,
                "method": "mean(子维度标准化分)",
                "district_avg": round(float(dist_dim_vals.mean()), 2) if len(dist_dim_vals) > 0 else None,
            }

        # 总分
        overall_score = float(self.dim_scores.loc[school, "总体得分"]) if "总体得分" in self.dim_scores.columns else None
        dist_overall = self.dim_scores.loc[
            [s for s in dist_schools if s in self.dim_scores.index], "总体得分"
        ] if "总体得分" in self.dim_scores.columns else pd.Series()

        rank = int((dist_overall >= overall_score).sum()) if overall_score is not None and len(dist_overall) > 0 else None

        overall = {
            "score": round(overall_score, 2) if overall_score is not None else None,
            "method": "mean(7个维度分)",
            "district_avg": round(float(dist_overall.mean()), 2) if len(dist_overall) > 0 else None,
            "rank": rank,
            "total_schools": len(dist_schools),
        }

        return dims, overall

    # ====================================================================
    # 数据访问代理
    # ====================================================================

    def _get_course(self, school: str) -> pd.DataFrame:
        return self.pca._get_course(school)

    def _get_subject(self, school: str, subject: str) -> pd.DataFrame:
        return self.pca._get_subject(school, subject)

    # ====================================================================
    # 缓存
    # ====================================================================

    def _load_cache(self, school: str) -> Optional[Dict]:
        TRACE_CACHE_DIR.mkdir(parents=True, exist_ok=True)
        cache_file = TRACE_CACHE_DIR / f"{school}_trace.json"
        if cache_file.exists():
            try:
                return json.loads(cache_file.read_text("utf-8"))
            except Exception:
                return None
        return None

    def _save_cache(self, school: str, data: Dict):
        TRACE_CACHE_DIR.mkdir(parents=True, exist_ok=True)
        cache_file = TRACE_CACHE_DIR / f"{school}_trace.json"
        try:
            cache_file.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=_json_default), "utf-8")
        except Exception as e:
            logger.warning(f"[Trace] Cache write failed for {school}: {e}")


# ====================================================================
# 工具函数
# ====================================================================

def _safe_value(v) -> Any:
    """将pandas值转为JSON安全类型"""
    if pd.isna(v):
        return None
    if isinstance(v, (np.integer,)):
        return int(v)
    if isinstance(v, (np.floating,)):
        return round(float(v), 4)
    return str(v)


def _guess_type(v) -> str:
    """猜测字段类型"""
    if pd.isna(v):
        return "null"
    s = str(v).strip()
    try:
        float(s)
        return "number"
    except ValueError:
        pass
    if s in ("0", "1", "有", "无", "是", "否"):
        return "binary"
    if s in ("已经建成并使用", "已经建成但未使用", "尚未建成", "已建成并使用", "已建成但未使用"):
        return "ordinal"
    return "text"


def _json_default(obj):
    """JSON序列化兜底"""
    if isinstance(obj, (np.integer,)):
        return int(obj)
    if isinstance(obj, (np.floating,)):
        return round(float(obj), 4)
    if isinstance(obj, np.ndarray):
        return obj.tolist()
    if isinstance(obj, pd.Timestamp):
        return str(obj)
    if isinstance(obj, float) and (np.isnan(obj) or np.isinf(obj)):
        return None
    return str(obj)