""" 종합 진단 엔드포인트. GET /schools/{schul_code}/diagnose → 지속 가능성 진단 (예측 + SHAP + 점수 + 시뮬레이션) POST /schools/{schul_code}/simulate → 사용자 정의 정책 시뮬레이션 (overrides 적용 후 점수 재산출) """ from __future__ import annotations import logging import pandas as pd from fastapi import APIRouter, HTTPException, Query from src.analytics.diagnostics import DiagnosticsEngine from src.analytics.scorer import SustainabilityScorer from src.api.schemas import ( DiagnosticsResponse, ForecastResponse, ShapResultResponse, ShapValueItem, SimulationRequest, SustainabilityScoreResponse, SimulationResult, ) from src.ingestion.cleaner import DataCleaner from src.ingestion.data_go_kr import DataGoKrClient from src.ingestion.standardizer import DataStandardizer router = APIRouter(prefix="/schools", tags=["diagnostics"]) logger = logging.getLogger(__name__) _standardizer = DataStandardizer() _cleaner = DataCleaner() _engine = DiagnosticsEngine() _scorer = SustainabilityScorer() _HISTORY_YEARS = 7 async def _collect_school_data(schul_code: str, end_year: int) -> tuple[dict, pd.Series]: """ 학교 기본 정보 + 연도별 학생 수 시계열을 수집하여 반환합니다. Returns ------- (school_data dict, student timeseries pd.Series) """ school_data: dict = {"sd_schul_code": schul_code} yearly_counts: dict[int, float] = {} start_year = max(end_year - _HISTORY_YEARS + 1, 2000) async with DataGoKrClient() as client: # 학교 기본 정보 info_rows = await client.fetch_school_info(schul_code=schul_code) if info_rows: try: info_df = _standardizer.standardize_school_info(info_rows) if schul_code in info_df.index: school_data.update(info_df.loc[schul_code].to_dict()) # established_year 추출 est_date = str(school_data.get("established_date", "") or "") if est_date and len(est_date) >= 4: try: school_data["established_year"] = int(est_date[:4]) except ValueError: pass except Exception as exc: # noqa: BLE001 logger.warning("학교기본정보 표준화 실패: %s", exc) # 연도별 학생·교원 수 for year in range(start_year, end_year + 1): s_rows = await client.fetch_student_count(year=year, schul_code=schul_code) t_rows = await client.fetch_teacher_count(year=year, schul_code=schul_code) if not s_rows and not t_rows: continue stats = _standardizer.build_school_stats(s_rows, t_rows, year=year) if stats.empty or schul_code not in stats.index.get_level_values("sd_schul_code"): continue row = stats.xs(schul_code, level="sd_schul_code").iloc[0] if "student_count" in row and pd.notna(row["student_count"]): yearly_counts[year] = float(row["student_count"]) # 최신 연도의 교원 정보를 school_data 에 주입 if year == end_year: for col in ["teacher_count", "temp_teacher_count", "class_count"]: if col in row and pd.notna(row[col]): school_data[col] = float(row[col]) timeseries = pd.Series(yearly_counts).sort_index().astype(float) return school_data, timeseries @router.get( "/{schul_code}/diagnose", response_model=DiagnosticsResponse, summary="학교 종합 진단", description=( "시계열 예측, SHAP 기여도 분석, 지속 가능성 점수 산출, " "정책 시뮬레이션을 통합하여 학교 종합 진단 결과를 반환합니다." ), ) async def diagnose_school( schul_code: str, horizon_years: int = Query(default=5, ge=1, le=10, description="예측 기간 (년)"), end_year: int = Query(default=2023, ge=2000, le=2100, description="히스토리 마지막 학년도"), run_simulations: bool = Query(default=True, description="정책 시뮬레이션 실행 여부"), ) -> DiagnosticsResponse: """학교 종합 진단을 수행합니다.""" logger.info( "진단 요청: schul_code=%s horizon=%dy end_year=%d", schul_code, horizon_years, end_year, ) try: school_data, timeseries = await _collect_school_data(schul_code, end_year) except Exception as exc: logger.error("데이터 수집 실패: %s", exc) raise HTTPException(status_code=502, detail=f"데이터 수집 실패: {exc}") from exc if len(timeseries.dropna()) < 2: raise HTTPException( status_code=422, detail=f"예측에 필요한 시계열 데이터가 부족합니다 (수집 연도 수: {len(timeseries)}).", ) try: result = _engine.diagnose( schul_code=schul_code, school_data=school_data, timeseries=timeseries, horizon_years=horizon_years, run_simulations=run_simulations, ) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc except Exception as exc: logger.error("진단 엔진 오류: %s", exc) raise HTTPException(status_code=500, detail=f"진단 중 오류가 발생했습니다: {exc}") from exc raw = result.to_dict() # Pydantic 모델로 변환 forecast_resp = ForecastResponse(**raw["forecast"]) shap_resp = ShapResultResponse( **{ **raw["shap_result"], "top_factors": [ShapValueItem(**f) for f in raw["shap_result"]["top_factors"]], } ) score_resp = SustainabilityScoreResponse(**raw["score_detail"]) sim_resp = {k: SimulationResult(**v) for k, v in raw["simulations"].items()} return DiagnosticsResponse( schul_code=raw["schul_code"], school_name=raw["school_name"], status_label=raw["status_label"], status_code=raw["status_code"], sustainability_score=raw["sustainability_score"], forecast=forecast_resp, shap_result=shap_resp, score_detail=score_resp, simulations=sim_resp, metadata=raw["metadata"], ) @router.post( "/{schul_code}/simulate", response_model=SustainabilityScoreResponse, summary="사용자 정의 정책 시뮬레이션", description=( "school_data_overrides 에 변경할 지표를 지정하면 " "해당 값을 적용한 지속 가능성 점수를 반환합니다." ), ) async def simulate_policy( schul_code: str, request: SimulationRequest, end_year: int = Query(default=2023, ge=2000, le=2100, description="기준 학년도"), ) -> SustainabilityScoreResponse: """사용자 정의 정책 시뮬레이션을 수행합니다.""" logger.info( "시뮬레이션 요청: schul_code=%s overrides=%s", schul_code, request.school_data_overrides, ) try: school_data, _ = await _collect_school_data(schul_code, end_year) except Exception as exc: logger.error("데이터 수집 실패: %s", exc) raise HTTPException(status_code=502, detail=f"데이터 수집 실패: {exc}") from exc # 오버라이드 적용 modified = dict(school_data) for key, value in request.school_data_overrides.items(): modified[key] = value try: score = _scorer.compute(schul_code=schul_code, school_data=modified) except Exception as exc: logger.error("점수 산출 실패: %s", exc) raise HTTPException(status_code=500, detail=f"점수 산출 실패: {exc}") from exc return SustainabilityScoreResponse(**score.to_dict())