""" 应用全局状态:管理数据引擎、赋分引擎、统计引擎的单例 避免每次请求重新加载 Excel 数据 """ import logging import time from datetime import datetime from typing import Dict, List, Optional import pandas as pd from ..engines.data_engine import DataEngine from ..engines.scoring_engine import ScoringEngine from ..engines.stats_engine import StatsEngine from ..config import SCHOOL_TYPE_MAP, DIMENSION_FRAMEWORK logger = logging.getLogger(__name__) class AppState: """应用全局状态""" def __init__(self): self.data_engine: Optional[DataEngine] = None self.scoring_engine: Optional[ScoringEngine] = None self.stats_engine: Optional[StatsEngine] = None self.raw_scores: Optional[Dict] = None self.sub_scores: Optional[pd.DataFrame] = None self.dim_scores: Optional[pd.DataFrame] = None self.schools: List[str] = [] self._initialized = False def initialize(self): """初始化所有引擎(只在应用启动时调用一次)""" if self._initialized: return start = time.time() # 1. 加载数据 self.data_engine = DataEngine() self.data_engine.load_all() self.schools = self.data_engine.schools # 2. 赋分 self.scoring_engine = ScoringEngine(self.data_engine) self.raw_scores = self.scoring_engine.score_all_schools() # 3. 统计 self.stats_engine = StatsEngine() self.sub_scores = self.stats_engine.compute_dimension_scores(self.raw_scores) self.dim_scores = self.stats_engine.compute_dimension_aggregates(self.sub_scores) self._initialized = True elapsed = time.time() - start logger.info(f"全局状态初始化完成,耗时 {elapsed:.1f}s") def get_report_data(self, school: str) -> Dict: """获取某学校的报告数据包""" if not self._initialized: self.initialize() return self.stats_engine.compute_school_report_data( school, self.sub_scores, self.dim_scores ) def get_school_info(self, school: str) -> Dict: """获取学校基本信息""" info = SCHOOL_TYPE_MAP.get(school, {}) if not self.dim_scores is None and school in self.dim_scores.index: score = round(float(self.dim_scores.loc[school, "总体得分"]), 2) rank = int((self.dim_scores["总体得分"] >= self.dim_scores.loc[school, "总体得分"]).sum()) else: score = 0 rank = 0 return { "name": school, "type": info.get("type", ""), "code": info.get("code", ""), "nature": info.get("nature", ""), "feature": info.get("feature", ""), "score": score, "rank": rank, "total_schools": len(self.schools), } def get_all_schools_summary(self) -> List[Dict]: """获取所有学校的摘要信息""" if not self._initialized: self.initialize() summaries = [] # 排名 sorted_schools = self.dim_scores["总体得分"].sort_values(ascending=False) for rank, (school, score) in enumerate(sorted_schools.items(), 1): if school == "总体得分": continue info = SCHOOL_TYPE_MAP.get(school, {}) # 聚类 dim_cols = [c for c in self.dim_scores.columns if c in DIMENSION_FRAMEWORK] cluster_result = self.stats_engine.cluster_analysis(self.dim_scores[dim_cols]) cluster = cluster_result["school_clusters"].get(school, "") # 检查已生成的报告 from ..config import OUTPUT_DIR report_path = OUTPUT_DIR / f"{school}_报告.html" has_report = report_path.exists() report_size = report_path.stat().st_size if has_report else 0 report_generated_at = "" if has_report: mtime = report_path.stat().st_mtime report_generated_at = datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S") summaries.append({ "name": school, "type": info.get("type", ""), "code": info.get("code", ""), "nature": info.get("nature", ""), "score": round(float(score), 2), "rank": rank, "cluster": cluster, "has_report": has_report, "report_size": report_size, "report_generated_at": report_generated_at, }) return summaries # 全局单例 app_state = AppState()