lofyerandfactory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
71db82393a
Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
256 lines
9.8 KiB
Python
256 lines
9.8 KiB
Python
"""
|
|
数据引擎：Excel解析 + 数据清洗 + 赋分计算
|
|
负责将原始Excel数据转换为每所学校的结构化得分数据
|
|
"""
|
|
import pandas as pd
|
|
import numpy as np
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple
|
|
import logging
|
|
|
|
from ..config import (
|
|
EXCEL_BASIC_INFO, EXCEL_COURSE_IMPL, EXCEL_SUBJECT_IMPL,
|
|
SCHOOL_NAME_MAP, SUBJECTS, SCHOOL_TYPE_MAP,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DataEngine:
|
|
"""数据引擎：读取Excel、清洗、结构化"""
|
|
|
|
def __init__(self, data_dir: Optional[Path] = None):
|
|
self.data_dir = data_dir
|
|
self._basic_info: Optional[pd.DataFrame] = None
|
|
self._course_impl: Optional[pd.DataFrame] = None
|
|
self._subject_impl: Optional[pd.DataFrame] = None
|
|
|
|
def load_all(self) -> None:
|
|
"""加载所有Excel数据"""
|
|
logger.info("开始加载Excel数据...")
|
|
self._basic_info = self._load_basic_info()
|
|
self._course_impl = self._load_course_impl()
|
|
self._subject_impl = self._load_subject_impl()
|
|
logger.info(
|
|
f"数据加载完成: 基础信息={len(self._basic_info)}行, "
|
|
f"课程实施={len(self._course_impl)}行, "
|
|
f"学科课程={len(self._subject_impl)}行"
|
|
)
|
|
|
|
def _standardize_school_name(self, df: pd.DataFrame, col: str) -> pd.DataFrame:
|
|
"""标准化学校名称"""
|
|
df[col] = df[col].map(lambda x: SCHOOL_NAME_MAP.get(x, x))
|
|
return df
|
|
|
|
def _load_basic_info(self) -> pd.DataFrame:
|
|
"""加载基础信息表"""
|
|
df = pd.read_excel(EXCEL_BASIC_INFO)
|
|
df = self._standardize_school_name(df, "学校名称")
|
|
return df
|
|
|
|
def _load_course_impl(self) -> pd.DataFrame:
|
|
"""加载课程实施情况表"""
|
|
df = pd.read_excel(EXCEL_COURSE_IMPL)
|
|
df = self._standardize_school_name(df, "学校简称")
|
|
# 统一列名
|
|
df = df.rename(columns={"学校简称": "学校名称"})
|
|
return df
|
|
|
|
def _load_subject_impl(self) -> pd.DataFrame:
|
|
"""加载学科课程实施情况表"""
|
|
df = pd.read_excel(EXCEL_SUBJECT_IMPL)
|
|
df = self._standardize_school_name(df, "学校简称")
|
|
df = df.rename(columns={"学校简称": "学校名称"})
|
|
return df
|
|
|
|
@property
|
|
def schools(self) -> List[str]:
|
|
"""获取所有学校名称列表"""
|
|
if self._course_impl is None:
|
|
self.load_all()
|
|
return sorted(self._course_impl["学校名称"].unique().tolist())
|
|
|
|
def get_school_basic_info(self, school: str) -> Dict:
|
|
"""获取学校基本信息"""
|
|
if self._basic_info is None:
|
|
self.load_all()
|
|
df = self._basic_info[self._basic_info["学校名称"] == school]
|
|
info = SCHOOL_TYPE_MAP.get(school, {})
|
|
|
|
# 提取关键字段
|
|
def _get_field(field_name):
|
|
rows = df[df["字段名称"] == field_name]
|
|
if len(rows) > 0:
|
|
return rows.iloc[0]["字段值"]
|
|
return None
|
|
|
|
info.update({
|
|
"school_name": school,
|
|
"建校年份": _get_field("建校年份"),
|
|
"教师总数": _get_field("学校教师总数"),
|
|
"占地面积": _get_field("占地面积"),
|
|
"建筑面积": _get_field("建筑面积"),
|
|
})
|
|
return info
|
|
|
|
def get_school_course_data(self, school: str) -> pd.DataFrame:
|
|
"""获取某学校的课程实施情况数据"""
|
|
if self._course_impl is None:
|
|
self.load_all()
|
|
return self._course_impl[self._course_impl["学校名称"] == school].copy()
|
|
|
|
def get_school_subject_data(self, school: str, subject: Optional[str] = None) -> pd.DataFrame:
|
|
"""获取某学校的学科课程实施数据"""
|
|
if self._subject_impl is None:
|
|
self.load_all()
|
|
df = self._subject_impl[self._subject_impl["学校名称"] == school].copy()
|
|
if subject:
|
|
df = df[df["学科"] == subject]
|
|
return df
|
|
|
|
# ========== 课程领导力相关数据提取 ==========
|
|
|
|
def get_weekly_hours(self, school: str) -> pd.DataFrame:
|
|
"""获取学校各学科各年级各学期的周课时数据"""
|
|
df = self.get_school_course_data(school)
|
|
# 筛选周课时相关字段
|
|
hours_fields = ["学科必修课周课时", "学科选择性必修课周课时", "学科类选修课周课时"]
|
|
result = df[df["字段名称"].isin(hours_fields)].copy()
|
|
result["字段取值"] = pd.to_numeric(result["字段取值"], errors="coerce")
|
|
return result
|
|
|
|
def get_course_norms(self, school: str) -> Dict:
|
|
"""获取学校课程规范落实相关数据（建设规范、档案等）"""
|
|
df = self.get_school_course_data(school)
|
|
norm_keywords = ["建设规范文本", "档案", "已经建成并使用", "尚未建成"]
|
|
mask = df["字段名称"].apply(
|
|
lambda x: any(k in str(x) for k in norm_keywords) if pd.notna(x) else False
|
|
)
|
|
return df[mask][["字段名称", "字段取值"]].to_dict("records")
|
|
|
|
# ========== 教学变革力相关数据提取 ==========
|
|
|
|
def get_teaching_reform_data(self, school: str) -> Dict:
|
|
"""获取教学方式变革相关数据（认识程度、落实程度、实施方式）"""
|
|
df = self.get_school_subject_data(school)
|
|
reform_keywords = [
|
|
"认识程度", "落实程度", "认识", "落实",
|
|
"理解式学习", "自主性学习", "实践性学习", "跨学科学习",
|
|
"信息技术与教学融合", "信息融入教学",
|
|
]
|
|
mask = df["字段名称"].apply(
|
|
lambda x: any(k in str(x) for k in reform_keywords) if pd.notna(x) else False
|
|
)
|
|
return df[mask]
|
|
|
|
def get_homework_data(self, school: str) -> Dict:
|
|
"""获取作业设计与管理数据"""
|
|
df = self.get_school_subject_data(school)
|
|
hw_keywords = [
|
|
"作业", "实践类", "表现类", "跨学科", "团队合作",
|
|
"批改", "评价", "属性标注", "时长控制",
|
|
]
|
|
mask = df["字段名称"].apply(
|
|
lambda x: any(k in str(x) for k in hw_keywords) if pd.notna(x) else False
|
|
)
|
|
return df[mask]
|
|
|
|
# ========== 通用数据提取方法 ==========
|
|
|
|
def get_field_value(self, school: str, source: str, field_name: str,
|
|
subject: Optional[str] = None) -> Optional[str]:
|
|
"""通用字段值获取"""
|
|
if source == "basic":
|
|
df = self._basic_info[self._basic_info["学校名称"] == school]
|
|
col = "字段名称"
|
|
val_col = "字段值"
|
|
elif source == "course":
|
|
df = self.get_school_course_data(school)
|
|
col = "字段名称"
|
|
val_col = "字段取值"
|
|
elif source == "subject":
|
|
df = self.get_school_subject_data(school, subject)
|
|
col = "字段名称"
|
|
val_col = "字段取值"
|
|
else:
|
|
return None
|
|
|
|
rows = df[df[col] == field_name]
|
|
if len(rows) > 0:
|
|
return rows.iloc[0][val_col]
|
|
return None
|
|
|
|
def get_field_by_id(self, school: str, source: str, field_id: str,
|
|
subject: Optional[str] = None) -> List[Dict]:
|
|
"""通过字段ID获取数据"""
|
|
if source == "basic":
|
|
df = self._basic_info[self._basic_info["学校名称"] == school]
|
|
elif source == "course":
|
|
df = self.get_school_course_data(school)
|
|
elif source == "subject":
|
|
df = self.get_school_subject_data(school, subject)
|
|
else:
|
|
return []
|
|
|
|
rows = df[df["字段ID"] == field_id]
|
|
return rows.to_dict("records")
|
|
|
|
def build_score_matrix(self) -> pd.DataFrame:
|
|
"""
|
|
构建所有学校×所有字段的得分矩阵
|
|
这是赋分引擎的输入
|
|
"""
|
|
if self._course_impl is None:
|
|
self.load_all()
|
|
|
|
records = []
|
|
|
|
for school in self.schools:
|
|
# 课程实施表数据
|
|
course_df = self.get_school_course_data(school)
|
|
for _, row in course_df.iterrows():
|
|
records.append({
|
|
"学校": school,
|
|
"来源": "course",
|
|
"题号": row.get("题号"),
|
|
"字段ID": row.get("字段ID"),
|
|
"字段名称": row.get("字段名称"),
|
|
"字段取值": row.get("字段取值"),
|
|
"学科": row.get("学科", "不分学科"),
|
|
"年级": row.get("年级", "不分年级"),
|
|
"学期": row.get("学期", ""),
|
|
})
|
|
|
|
# 学科课程表数据
|
|
subject_df = self.get_school_subject_data(school)
|
|
for _, row in subject_df.iterrows():
|
|
records.append({
|
|
"学校": school,
|
|
"来源": "subject",
|
|
"题号": row.get("题号"),
|
|
"字段ID": row.get("字段ID"),
|
|
"字段名称": row.get("字段名称"),
|
|
"字段取值": row.get("字段取值"),
|
|
"学科": row.get("学科", "不分学科"),
|
|
"年级": "",
|
|
"学期": "",
|
|
})
|
|
|
|
matrix = pd.DataFrame(records)
|
|
logger.info(f"得分矩阵构建完成: {len(matrix)}行, {len(self.schools)}所学校")
|
|
return matrix
|
|
|
|
def summary(self) -> Dict:
|
|
"""数据摘要"""
|
|
if self._basic_info is None:
|
|
self.load_all()
|
|
return {
|
|
"schools": self.schools,
|
|
"school_count": len(self.schools),
|
|
"basic_info_rows": len(self._basic_info),
|
|
"course_impl_rows": len(self._course_impl),
|
|
"subject_impl_rows": len(self._subject_impl),
|
|
"subjects": SUBJECTS,
|
|
"school_types": {s: SCHOOL_TYPE_MAP[s]["type"] for s in self.schools},
|
|
}
|