src/data/log_parser.py - master - CW3/The_Project

src/data/log_parser.py
184 строки7 KB
CW3
Initial project setup with structure and CI/CD
28 дек 2025, 12:40
28 дек 2025, 12:405cacb7a
"""
Модуль для парсинга логов LMS (Moodle, Canvas, etc.)
"""
import pandas as pd
import numpy as np
from datetime import datetime
from typing import Dict, List, Optional
import re


class LMSLogParser:
    """Парсер логов системы управления обучением (LMS)"""
    
    # Типы событий в LMS
    EVENT_TYPES = {
        'login': ['login', 'logged in', 'authentication'],
        'logout': ['logout', 'logged out'],
        'course_view': ['course viewed', 'course view'],
        'resource_view': ['resource viewed', 'file viewed'],
        'assignment_submit': ['assignment submitted', 'submit assignment'],
        'quiz_attempt': ['quiz attempt', 'attempt submitted'],
        'forum_post': ['forum post', 'discussion post', 'forum reply'],
        'message_sent': ['message sent', 'message submitted'],
        'grade_view': ['grade viewed', 'grades viewed']
    }
    
    def __init__(self, log_file_path: str):
        """
        Инициализация парсера
        
        Args:
            log_file_path: Путь к файлу логов CSV
        """
        self.log_file_path = log_file_path
        self.df = None
        self.event_counts = None
        
    def load_logs(self) -> pd.DataFrame:
        """
        Загрузка логов из CSV файла
        
        Returns:
            DataFrame с логами
        """
        try:
            self.df = pd.read_csv(self.log_file_path, encoding='utf-8')
            print(f"Загружено {len(self.df)} записей логов")
            return self.df
        except Exception as e:
            print(f"Ошибка при загрузке файла: {e}")
            # Создаем пример данных, если файл не найден
            return self._create_sample_data()
    
    def _create_sample_data(self, num_records: int = 1000) -> pd.DataFrame:
        """
        Создание синтетических данных логов для тестирования
        
        Args:
            num_records: Количество записей
            
        Returns:
            DataFrame с синтетическими логами
        """
        np.random.seed(42)
        
        # Создаем временную шкалу
        start_date = datetime(2024, 9, 1)
        dates = [start_date + pd.Timedelta(hours=i) for i in range(num_records)]
        
        # Создаем события
        events = []
        event_categories = list(self.EVENT_TYPES.keys())
        
        for i in range(num_records):
            student_id = np.random.randint(1000, 1100)
            event_type = np.random.choice(event_categories, p=[0.1, 0.05, 0.2, 0.15, 0.1, 0.1, 0.15, 0.1, 0.05])
            course_id = np.random.choice(['CS101', 'MATH201', 'PHYS101', 'ENG101'])
            
            # Генерируем описание события
            description = self._generate_event_description(event_type, course_id)
            
            # Время события с некоторой случайностью
            event_time = dates[i] + pd.Timedelta(minutes=np.random.randint(0, 60))
            
            events.append({
                'timestamp': event_time,
                'student_id': student_id,
                'event_type': event_type,
                'course_id': course_id,
                'description': description,
                'duration_seconds': np.random.exponential(300)  # Время в системе
            })
        
        self.df = pd.DataFrame(events)
        print(f"Создано {len(self.df)} синтетических записей логов")
        return self.df
    
    def _generate_event_description(self, event_type: str, course_id: str) -> str:
        """Генерация описания события"""
        descriptions = {
            'login': f"Пользователь вошел в систему",
            'logout': f"Пользователь вышел из системы",
            'course_view': f"Просмотр курса {course_id}",
            'resource_view': f"Просмотр материала в курсе {course_id}",
            'assignment_submit': f"Сдача задания в курсе {course_id}",
            'quiz_attempt': f"Попытка теста в курсе {course_id}",
            'forum_post': f"Сообщение на форуме курса {course_id}",
            'message_sent': f"Отправлено сообщение в курсе {course_id}",
            'grade_view': f"Просмотр оценок курса {course_id}"
        }
        return descriptions.get(event_type, f"Событие в курсе {course_id}")
    
    def categorize_events(self) -> Dict[str, int]:
        """
        Категоризация событий по типам
        
        Returns:
            Словарь с количеством событий каждого типа
        """
        if self.df is None:
            self.load_logs()
        
        self.event_counts = self.df['event_type'].value_counts().to_dict()
        return self.event_counts
    
    def get_student_activity(self, student_id: int) -> pd.DataFrame:
        """
        Получение активности конкретного студента
        
        Args:
            student_id: ID студента
            
        Returns:
            DataFrame с активностью студента
        """
        if self.df is None:
            self.load_logs()
        
        return self.df[self.df['student_id'] == student_id].copy()
    
    def get_course_activity(self, course_id: str) -> pd.DataFrame:
        """
        Получение активности по курсу
        
        Args:
            course_id: ID курса
            
        Returns:
            DataFrame с активностью по курсу
        """
        if self.df is None:
            self.load_logs()
        
        return self.df[self.df['course_id'] == course_id].copy()
    
    def save_parsed_logs(self, output_path: str):
        """
        Сохранение обработанных логов
        
        Args:
            output_path: Путь для сохранения
        """
        if self.df is not None:
            self.df.to_csv(output_path, index=False, encoding='utf-8')
            print(f"Логи сохранены в {output_path}")


def generate_sample_logs_csv(output_path: str = 'data/sample_logs.csv', 
                           num_records: int = 500):
    """
    Генерация примерного CSV файла с логами LMS
    
    Args:
        output_path: Путь для сохранения
        num_records: Количество записей
    """
    parser = LMSLogParser('')
    df = parser._create_sample_data(num_records)
    
    # Сохраняем в CSV
    df.to_csv(output_path, index=False, encoding='utf-8')
    print(f"Пример логов сохранен в {output_path}")
    
    return df