src/data_loader.py - master - katherinesiv/study_material_recommender

src/data_loader.py
150 строк5 KB
Сиваева Екатерина
update src/data_loader.py
25 дек 2025, 11:23
25 дек 2025, 11:235745418
"""
Data loading and preprocessing module
"""

import logging
from typing import Any, Dict, Tuple

import pandas as pd


logger = logging.getLogger(__name__)


class DataLoader:
    """Loads and preprocesses study materials and ratings data"""

    def __init__(self, materials_path: str, ratings_path: str):
        """
        Initialize DataLoader with file paths.

        Args:
            materials_path: Path to materials CSV file
            ratings_path: Path to ratings CSV file
        """
        self.materials_path = materials_path
        self.ratings_path = ratings_path
        self.materials_df = None
        self.ratings_df = None
        self.user_item_matrix = None

    def load_data(self) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """
        Load data from CSV files.

        Returns:
            Tuple of (materials_df, ratings_df)
        """
        try:
            self.materials_df = pd.read_csv(self.materials_path)
            self.ratings_df = pd.read_csv(self.ratings_path)
            logger.info(
                f"Loaded {len(self.materials_df)} materials and "
                f"{len(self.ratings_df)} ratings"
            )
            return self.materials_df, self.ratings_df
        except FileNotFoundError as e:
            logger.error(f"File not found: {e}")
            raise
        except Exception as e:
            logger.error(f"Error loading  {e}")
            raise

    def preprocess_data(self) -> Dict[str, Any]:
        """
        Preprocess data for modeling.

        Returns:
            Dictionary containing preprocessed data
        """
        if self.materials_df is None or self.ratings_df is None:
            self.load_data()

        # Create user-item matrix for collaborative filtering
        self.user_item_matrix = self.ratings_df.pivot_table(
            index="user_id",
            columns="material_id",
            values="rating",
            fill_value=0
        )

        # Encode categorical features for content-based filtering
        materials_processed = self._encode_features(self.materials_df)

        # Calculate basic statistics
        num_users = len(self.ratings_df["user_id"].unique())
        num_materials = len(self.ratings_df["material_id"].unique())
        total_possible = num_users * num_materials
        rating_sparsity = 1 - (len(self.ratings_df) / total_possible)

        stats = {
            "num_users": num_users,
            "num_materials": num_materials,
            "avg_rating": self.ratings_df["rating"].mean(),
            "rating_sparsity": rating_sparsity
        }

        logger.info(f"Data statistics: {stats}")

        return {
            "materials_df": self.materials_df,
            "ratings_df": self.ratings_df,
            "user_item_matrix": self.user_item_matrix,
            "materials_processed": materials_processed,
            "stats": stats
        }

    def _encode_features(self, materials_df: pd.DataFrame) -> pd.DataFrame:
        """
        Encode categorical features for content-based filtering.

        Args:
            materials_df: Raw materials dataframe

        Returns:
            Encoded materials dataframe
        """
        df = materials_df.copy()

        # One-hot encode categorical columns if they exist
        categorical_cols = ["subject", "difficulty", "material_type"]
        existing_categorical = [col for col in categorical_cols if col in df.columns]

        if existing_categorical:
            df = pd.get_dummies(df, columns=existing_categorical, drop_first=True)

        # Normalize numerical columns if they exist
        numerical_cols = ["duration_minutes", "pages", "level"]
        existing_numerical = [col for col in numerical_cols if col in df.columns]

        if existing_numerical:
            for col in existing_numerical:
                if col in df.columns:
                    df[col] = (df[col] - df[col].mean()) / df[col].std()

        return df

    def get_user_history(self, user_id: int) -> pd.DataFrame:
        """
        Get learning history for a specific user.

        Args:
            user_id: User identifier

        Returns:
            DataFrame with user's rating history
        """
        if self.ratings_df is None:
            self.load_data()

        user_history = self.ratings_df[
            self.ratings_df["user_id"] == user_id
        ].copy()
        if len(user_history) > 0:
            user_history = user_history.merge(
                self.materials_df,
                on="material_id",
                how="left"
            )

        return user_history