src/processor.py - master - NSir/student-progress-visualizer

src/processor.py
130 строк4 KB
NSir
Initial commit
23 май 2026, 19:41
23 май 2026, 19:419a83b4d
import pandas as pd


REQUIRED_COLUMNS = [
    "Фамилия",
    "Имя",
    "Индивидуальный номер",
    "Группа",
    "Итог",
]


def validate_columns(df: pd.DataFrame) -> None:
    """
    Проверяет наличие обязательных столбцов.
    """
    missing = [col for col in REQUIRED_COLUMNS if col not in df.columns]

    if missing:
        raise ValueError(
            "В данных отсутствуют обязательные столбцы: "
            + ", ".join(missing)
        )


def find_work_columns(df: pd.DataFrame) -> dict[str, list[str]]:
    """
    Находит столбцы с оценками по типам работ.
    """
    return {
        "Лекции": [col for col in df.columns if str(col).startswith("Лекция")],
        "Тесты": [col for col in df.columns if str(col).startswith("Тест")],
        "Задания": [col for col in df.columns if str(col).startswith("Задание")],
    }


def prepare_progress_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Добавляет агрегированные столбцы:
    - Баллы за лекции
    - Баллы за тесты
    - Баллы за задания
    - Итог
    """
    validate_columns(df)
    work_columns = find_work_columns(df)

    prepared = df.copy()

    for category, columns in work_columns.items():
        if columns:
            prepared[category] = prepared[columns].fillna(0).sum(axis=1)
        else:
            prepared[category] = 0

    prepared["Итог"] = prepared["Итог"].fillna(0)

    columns_to_keep = [
        "Дата",
        "Фамилия",
        "Имя",
        "Индивидуальный номер",
        "Группа",
        "Лекции",
        "Тесты",
        "Задания",
        "Итог",
    ]

    prepared = prepared[columns_to_keep]
    prepared = prepared.sort_values(["Дата", "Группа", "Фамилия", "Имя"])

    return prepared


def get_student_data(df: pd.DataFrame, student_id: str) -> pd.DataFrame:
    """
    Возвращает данные одного студента по индивидуальному номеру.
    """
    result = df[df["Индивидуальный номер"].astype(str) == str(student_id)]

    if result.empty:
        raise ValueError(f"Студент с индивидуальным номером {student_id} не найден")

    return result.sort_values("Дата")


def get_group_average_data(df: pd.DataFrame, group_id: str) -> pd.DataFrame:
    """
    Возвращает средние показатели по группе.
    """
    group_df = df[df["Группа"].astype(str) == str(group_id)]

    if group_df.empty:
        raise ValueError(f"Группа {group_id} не найдена")

    return (
        group_df
        .groupby("Дата", as_index=False)[["Лекции", "Тесты", "Задания", "Итог"]]
        .mean()
    )


def get_stream_average_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Возвращает средние показатели по всему потоку.
    """
    return (
        df
        .groupby("Дата", as_index=False)[["Лекции", "Тесты", "Задания", "Итог"]]
        .mean()
    )


def list_students(df: pd.DataFrame) -> pd.DataFrame:
    """
    Возвращает список студентов.
    """
    return (
        df[["Индивидуальный номер", "Фамилия", "Имя", "Группа"]]
        .drop_duplicates()
        .sort_values(["Группа", "Фамилия", "Имя"])
    )


def list_groups(df: pd.DataFrame) -> list:
    """
    Возвращает список групп.
    """
    return sorted(df["Группа"].dropna().astype(str).unique())