""" Session Repository - CRUD Operationen fuer Classroom Sessions (Feature f14). Abstraktion der Datenbank-Operationen fuer LessonSessions. """ from datetime import datetime from typing import Optional, List, Dict, Any from sqlalchemy.orm import Session as DBSession from .db_models import ( LessonSessionDB, PhaseHistoryDB, LessonTemplateDB, TeacherSettingsDB, LessonPhaseEnum, HomeworkDB, HomeworkStatusEnum, PhaseMaterialDB, MaterialTypeEnum, LessonReflectionDB, TeacherFeedbackDB, FeedbackTypeEnum, FeedbackStatusEnum, FeedbackPriorityEnum ) from .context_models import ( TeacherContextDB, SchoolyearEventDB, RecurringRoutineDB, MacroPhaseEnum, EventTypeEnum, EventStatusEnum, RoutineTypeEnum, RecurrencePatternEnum, FEDERAL_STATES, SCHOOL_TYPES ) from .models import ( LessonSession, LessonTemplate, LessonPhase, Homework, HomeworkStatus, PhaseMaterial, MaterialType, get_default_durations ) from .analytics import ( LessonReflection, SessionSummary, TeacherAnalytics, AnalyticsCalculator ) class SessionRepository: """Repository fuer LessonSession CRUD-Operationen.""" def __init__(self, db: DBSession): self.db = db # ==================== CREATE ==================== def create(self, session: LessonSession) -> LessonSessionDB: """ Erstellt eine neue Session in der Datenbank. Args: session: LessonSession Dataclass Returns: LessonSessionDB Model """ db_session = LessonSessionDB( id=session.session_id, teacher_id=session.teacher_id, class_id=session.class_id, subject=session.subject, topic=session.topic, current_phase=LessonPhaseEnum(session.current_phase.value), is_paused=session.is_paused, lesson_started_at=session.lesson_started_at, lesson_ended_at=session.lesson_ended_at, phase_started_at=session.phase_started_at, pause_started_at=session.pause_started_at, total_paused_seconds=session.total_paused_seconds, phase_durations=session.phase_durations, phase_history=session.phase_history, notes=session.notes, homework=session.homework, ) self.db.add(db_session) self.db.commit() self.db.refresh(db_session) return db_session # ==================== READ ==================== def get_by_id(self, session_id: str) -> Optional[LessonSessionDB]: """Holt eine Session nach ID.""" return self.db.query(LessonSessionDB).filter( LessonSessionDB.id == session_id ).first() def get_active_by_teacher(self, teacher_id: str) -> List[LessonSessionDB]: """Holt alle aktiven Sessions eines Lehrers.""" return self.db.query(LessonSessionDB).filter( LessonSessionDB.teacher_id == teacher_id, LessonSessionDB.current_phase != LessonPhaseEnum.ENDED ).all() def get_history_by_teacher( self, teacher_id: str, limit: int = 20, offset: int = 0 ) -> List[LessonSessionDB]: """Holt Session-History eines Lehrers (Feature f17).""" return self.db.query(LessonSessionDB).filter( LessonSessionDB.teacher_id == teacher_id, LessonSessionDB.current_phase == LessonPhaseEnum.ENDED ).order_by( LessonSessionDB.lesson_ended_at.desc() ).offset(offset).limit(limit).all() def get_by_class( self, class_id: str, limit: int = 20 ) -> List[LessonSessionDB]: """Holt Sessions einer Klasse.""" return self.db.query(LessonSessionDB).filter( LessonSessionDB.class_id == class_id ).order_by( LessonSessionDB.created_at.desc() ).limit(limit).all() # ==================== UPDATE ==================== def update(self, session: LessonSession) -> Optional[LessonSessionDB]: """ Aktualisiert eine bestehende Session. Args: session: LessonSession Dataclass mit aktualisierten Werten Returns: Aktualisierte LessonSessionDB oder None """ db_session = self.get_by_id(session.session_id) if not db_session: return None db_session.current_phase = LessonPhaseEnum(session.current_phase.value) db_session.is_paused = session.is_paused db_session.lesson_started_at = session.lesson_started_at db_session.lesson_ended_at = session.lesson_ended_at db_session.phase_started_at = session.phase_started_at db_session.pause_started_at = session.pause_started_at db_session.total_paused_seconds = session.total_paused_seconds db_session.phase_durations = session.phase_durations db_session.phase_history = session.phase_history db_session.notes = session.notes db_session.homework = session.homework self.db.commit() self.db.refresh(db_session) return db_session def update_notes( self, session_id: str, notes: str, homework: str ) -> Optional[LessonSessionDB]: """Aktualisiert nur Notizen und Hausaufgaben.""" db_session = self.get_by_id(session_id) if not db_session: return None db_session.notes = notes db_session.homework = homework self.db.commit() self.db.refresh(db_session) return db_session # ==================== DELETE ==================== def delete(self, session_id: str) -> bool: """Loescht eine Session.""" db_session = self.get_by_id(session_id) if not db_session: return False self.db.delete(db_session) self.db.commit() return True # ==================== CONVERSION ==================== def to_dataclass(self, db_session: LessonSessionDB) -> LessonSession: """ Konvertiert DB-Model zu Dataclass. Args: db_session: LessonSessionDB Model Returns: LessonSession Dataclass """ return LessonSession( session_id=db_session.id, teacher_id=db_session.teacher_id, class_id=db_session.class_id, subject=db_session.subject, topic=db_session.topic, current_phase=LessonPhase(db_session.current_phase.value), phase_started_at=db_session.phase_started_at, lesson_started_at=db_session.lesson_started_at, lesson_ended_at=db_session.lesson_ended_at, is_paused=db_session.is_paused, pause_started_at=db_session.pause_started_at, total_paused_seconds=db_session.total_paused_seconds or 0, phase_durations=db_session.phase_durations or get_default_durations(), phase_history=db_session.phase_history or [], notes=db_session.notes or "", homework=db_session.homework or "", ) class TeacherSettingsRepository: """Repository fuer Lehrer-Einstellungen (Feature f16).""" def __init__(self, db: DBSession): self.db = db def get_or_create(self, teacher_id: str) -> TeacherSettingsDB: """Holt oder erstellt Einstellungen fuer einen Lehrer.""" settings = self.db.query(TeacherSettingsDB).filter( TeacherSettingsDB.teacher_id == teacher_id ).first() if not settings: settings = TeacherSettingsDB( teacher_id=teacher_id, default_phase_durations=get_default_durations(), ) self.db.add(settings) self.db.commit() self.db.refresh(settings) return settings def update_phase_durations( self, teacher_id: str, durations: Dict[str, int] ) -> TeacherSettingsDB: """Aktualisiert die Standard-Phasendauern.""" settings = self.get_or_create(teacher_id) settings.default_phase_durations = durations self.db.commit() self.db.refresh(settings) return settings def update_preferences( self, teacher_id: str, audio_enabled: Optional[bool] = None, high_contrast: Optional[bool] = None, show_statistics: Optional[bool] = None ) -> TeacherSettingsDB: """Aktualisiert UI-Praeferenzen.""" settings = self.get_or_create(teacher_id) if audio_enabled is not None: settings.audio_enabled = audio_enabled if high_contrast is not None: settings.high_contrast = high_contrast if show_statistics is not None: settings.show_statistics = show_statistics self.db.commit() self.db.refresh(settings) return settings class TemplateRepository: """Repository fuer Stunden-Vorlagen (Feature f37).""" def __init__(self, db: DBSession): self.db = db # ==================== CREATE ==================== def create(self, template: LessonTemplate) -> LessonTemplateDB: """Erstellt eine neue Vorlage.""" db_template = LessonTemplateDB( id=template.template_id, teacher_id=template.teacher_id, name=template.name, description=template.description, subject=template.subject, grade_level=template.grade_level, phase_durations=template.phase_durations, default_topic=template.default_topic, default_notes=template.default_notes, is_public=template.is_public, usage_count=template.usage_count, ) self.db.add(db_template) self.db.commit() self.db.refresh(db_template) return db_template # ==================== READ ==================== def get_by_id(self, template_id: str) -> Optional[LessonTemplateDB]: """Holt eine Vorlage nach ID.""" return self.db.query(LessonTemplateDB).filter( LessonTemplateDB.id == template_id ).first() def get_by_teacher( self, teacher_id: str, include_public: bool = True ) -> List[LessonTemplateDB]: """ Holt alle Vorlagen eines Lehrers. Args: teacher_id: ID des Lehrers include_public: Auch oeffentliche Vorlagen anderer Lehrer einbeziehen """ if include_public: return self.db.query(LessonTemplateDB).filter( (LessonTemplateDB.teacher_id == teacher_id) | (LessonTemplateDB.is_public == True) ).order_by( LessonTemplateDB.usage_count.desc() ).all() else: return self.db.query(LessonTemplateDB).filter( LessonTemplateDB.teacher_id == teacher_id ).order_by( LessonTemplateDB.created_at.desc() ).all() def get_public_templates(self, limit: int = 20) -> List[LessonTemplateDB]: """Holt oeffentliche Vorlagen, sortiert nach Beliebtheit.""" return self.db.query(LessonTemplateDB).filter( LessonTemplateDB.is_public == True ).order_by( LessonTemplateDB.usage_count.desc() ).limit(limit).all() def get_by_subject( self, subject: str, teacher_id: Optional[str] = None ) -> List[LessonTemplateDB]: """Holt Vorlagen fuer ein bestimmtes Fach.""" query = self.db.query(LessonTemplateDB).filter( LessonTemplateDB.subject == subject ) if teacher_id: query = query.filter( (LessonTemplateDB.teacher_id == teacher_id) | (LessonTemplateDB.is_public == True) ) else: query = query.filter(LessonTemplateDB.is_public == True) return query.order_by( LessonTemplateDB.usage_count.desc() ).all() # ==================== UPDATE ==================== def update(self, template: LessonTemplate) -> Optional[LessonTemplateDB]: """Aktualisiert eine Vorlage.""" db_template = self.get_by_id(template.template_id) if not db_template: return None db_template.name = template.name db_template.description = template.description db_template.subject = template.subject db_template.grade_level = template.grade_level db_template.phase_durations = template.phase_durations db_template.default_topic = template.default_topic db_template.default_notes = template.default_notes db_template.is_public = template.is_public self.db.commit() self.db.refresh(db_template) return db_template def increment_usage(self, template_id: str) -> Optional[LessonTemplateDB]: """Erhoeht den Usage-Counter einer Vorlage.""" db_template = self.get_by_id(template_id) if not db_template: return None db_template.usage_count += 1 self.db.commit() self.db.refresh(db_template) return db_template # ==================== DELETE ==================== def delete(self, template_id: str) -> bool: """Loescht eine Vorlage.""" db_template = self.get_by_id(template_id) if not db_template: return False self.db.delete(db_template) self.db.commit() return True # ==================== CONVERSION ==================== def to_dataclass(self, db_template: LessonTemplateDB) -> LessonTemplate: """Konvertiert DB-Model zu Dataclass.""" return LessonTemplate( template_id=db_template.id, teacher_id=db_template.teacher_id, name=db_template.name, description=db_template.description or "", subject=db_template.subject or "", grade_level=db_template.grade_level or "", phase_durations=db_template.phase_durations or get_default_durations(), default_topic=db_template.default_topic or "", default_notes=db_template.default_notes or "", is_public=db_template.is_public, usage_count=db_template.usage_count, created_at=db_template.created_at, updated_at=db_template.updated_at, ) class HomeworkRepository: """Repository fuer Hausaufgaben-Tracking (Feature f20).""" def __init__(self, db: DBSession): self.db = db # ==================== CREATE ==================== def create(self, homework: Homework) -> HomeworkDB: """Erstellt eine neue Hausaufgabe.""" db_homework = HomeworkDB( id=homework.homework_id, teacher_id=homework.teacher_id, class_id=homework.class_id, subject=homework.subject, title=homework.title, description=homework.description, session_id=homework.session_id, due_date=homework.due_date, status=HomeworkStatusEnum(homework.status.value), ) self.db.add(db_homework) self.db.commit() self.db.refresh(db_homework) return db_homework # ==================== READ ==================== def get_by_id(self, homework_id: str) -> Optional[HomeworkDB]: """Holt eine Hausaufgabe nach ID.""" return self.db.query(HomeworkDB).filter( HomeworkDB.id == homework_id ).first() def get_by_teacher( self, teacher_id: str, status: Optional[str] = None, limit: int = 50 ) -> List[HomeworkDB]: """Holt alle Hausaufgaben eines Lehrers.""" query = self.db.query(HomeworkDB).filter( HomeworkDB.teacher_id == teacher_id ) if status: query = query.filter(HomeworkDB.status == HomeworkStatusEnum(status)) return query.order_by( HomeworkDB.due_date.asc().nullslast(), HomeworkDB.created_at.desc() ).limit(limit).all() def get_by_class( self, class_id: str, teacher_id: str, include_completed: bool = False, limit: int = 20 ) -> List[HomeworkDB]: """Holt alle Hausaufgaben einer Klasse.""" query = self.db.query(HomeworkDB).filter( HomeworkDB.class_id == class_id, HomeworkDB.teacher_id == teacher_id ) if not include_completed: query = query.filter(HomeworkDB.status != HomeworkStatusEnum.COMPLETED) return query.order_by( HomeworkDB.due_date.asc().nullslast(), HomeworkDB.created_at.desc() ).limit(limit).all() def get_by_session(self, session_id: str) -> List[HomeworkDB]: """Holt alle Hausaufgaben einer Session.""" return self.db.query(HomeworkDB).filter( HomeworkDB.session_id == session_id ).order_by(HomeworkDB.created_at.desc()).all() def get_pending( self, teacher_id: str, days_ahead: int = 7 ) -> List[HomeworkDB]: """Holt anstehende Hausaufgaben der naechsten X Tage.""" from datetime import timedelta cutoff = datetime.utcnow() + timedelta(days=days_ahead) return self.db.query(HomeworkDB).filter( HomeworkDB.teacher_id == teacher_id, HomeworkDB.status.in_([HomeworkStatusEnum.ASSIGNED, HomeworkStatusEnum.IN_PROGRESS]), HomeworkDB.due_date <= cutoff ).order_by(HomeworkDB.due_date.asc()).all() # ==================== UPDATE ==================== def update_status( self, homework_id: str, status: HomeworkStatus ) -> Optional[HomeworkDB]: """Aktualisiert den Status einer Hausaufgabe.""" db_homework = self.get_by_id(homework_id) if not db_homework: return None db_homework.status = HomeworkStatusEnum(status.value) self.db.commit() self.db.refresh(db_homework) return db_homework def update(self, homework: Homework) -> Optional[HomeworkDB]: """Aktualisiert eine Hausaufgabe.""" db_homework = self.get_by_id(homework.homework_id) if not db_homework: return None db_homework.title = homework.title db_homework.description = homework.description db_homework.due_date = homework.due_date db_homework.status = HomeworkStatusEnum(homework.status.value) self.db.commit() self.db.refresh(db_homework) return db_homework # ==================== DELETE ==================== def delete(self, homework_id: str) -> bool: """Loescht eine Hausaufgabe.""" db_homework = self.get_by_id(homework_id) if not db_homework: return False self.db.delete(db_homework) self.db.commit() return True # ==================== CONVERSION ==================== def to_dataclass(self, db_homework: HomeworkDB) -> Homework: """Konvertiert DB-Model zu Dataclass.""" return Homework( homework_id=db_homework.id, teacher_id=db_homework.teacher_id, class_id=db_homework.class_id, subject=db_homework.subject, title=db_homework.title, description=db_homework.description or "", session_id=db_homework.session_id, due_date=db_homework.due_date, status=HomeworkStatus(db_homework.status.value), created_at=db_homework.created_at, updated_at=db_homework.updated_at, ) class MaterialRepository: """Repository fuer Phasen-Materialien (Feature f19).""" def __init__(self, db: DBSession): self.db = db # ==================== CREATE ==================== def create(self, material: PhaseMaterial) -> PhaseMaterialDB: """Erstellt ein neues Material.""" db_material = PhaseMaterialDB( id=material.material_id, teacher_id=material.teacher_id, title=material.title, material_type=MaterialTypeEnum(material.material_type.value), url=material.url, description=material.description, phase=material.phase, subject=material.subject, grade_level=material.grade_level, tags=material.tags, is_public=material.is_public, usage_count=material.usage_count, session_id=material.session_id, ) self.db.add(db_material) self.db.commit() self.db.refresh(db_material) return db_material # ==================== READ ==================== def get_by_id(self, material_id: str) -> Optional[PhaseMaterialDB]: """Holt ein Material nach ID.""" return self.db.query(PhaseMaterialDB).filter( PhaseMaterialDB.id == material_id ).first() def get_by_teacher( self, teacher_id: str, phase: Optional[str] = None, subject: Optional[str] = None, limit: int = 50 ) -> List[PhaseMaterialDB]: """Holt alle Materialien eines Lehrers.""" query = self.db.query(PhaseMaterialDB).filter( PhaseMaterialDB.teacher_id == teacher_id ) if phase: query = query.filter(PhaseMaterialDB.phase == phase) if subject: query = query.filter(PhaseMaterialDB.subject == subject) return query.order_by( PhaseMaterialDB.usage_count.desc(), PhaseMaterialDB.created_at.desc() ).limit(limit).all() def get_by_phase( self, phase: str, teacher_id: str, include_public: bool = True ) -> List[PhaseMaterialDB]: """Holt alle Materialien fuer eine bestimmte Phase.""" if include_public: return self.db.query(PhaseMaterialDB).filter( PhaseMaterialDB.phase == phase, (PhaseMaterialDB.teacher_id == teacher_id) | (PhaseMaterialDB.is_public == True) ).order_by( PhaseMaterialDB.usage_count.desc() ).all() else: return self.db.query(PhaseMaterialDB).filter( PhaseMaterialDB.phase == phase, PhaseMaterialDB.teacher_id == teacher_id ).order_by( PhaseMaterialDB.created_at.desc() ).all() def get_by_session(self, session_id: str) -> List[PhaseMaterialDB]: """Holt alle Materialien einer Session.""" return self.db.query(PhaseMaterialDB).filter( PhaseMaterialDB.session_id == session_id ).order_by(PhaseMaterialDB.phase, PhaseMaterialDB.created_at).all() def get_public_materials( self, phase: Optional[str] = None, subject: Optional[str] = None, limit: int = 20 ) -> List[PhaseMaterialDB]: """Holt oeffentliche Materialien.""" query = self.db.query(PhaseMaterialDB).filter( PhaseMaterialDB.is_public == True ) if phase: query = query.filter(PhaseMaterialDB.phase == phase) if subject: query = query.filter(PhaseMaterialDB.subject == subject) return query.order_by( PhaseMaterialDB.usage_count.desc() ).limit(limit).all() def search_by_tags( self, tags: List[str], teacher_id: Optional[str] = None ) -> List[PhaseMaterialDB]: """Sucht Materialien nach Tags.""" # SQLite/PostgreSQL JSON contains query = self.db.query(PhaseMaterialDB) if teacher_id: query = query.filter( (PhaseMaterialDB.teacher_id == teacher_id) | (PhaseMaterialDB.is_public == True) ) else: query = query.filter(PhaseMaterialDB.is_public == True) # Filter by tags - vereinfachte Implementierung results = [] for material in query.all(): if material.tags and any(tag in material.tags for tag in tags): results.append(material) return results[:50] # ==================== UPDATE ==================== def update(self, material: PhaseMaterial) -> Optional[PhaseMaterialDB]: """Aktualisiert ein Material.""" db_material = self.get_by_id(material.material_id) if not db_material: return None db_material.title = material.title db_material.material_type = MaterialTypeEnum(material.material_type.value) db_material.url = material.url db_material.description = material.description db_material.phase = material.phase db_material.subject = material.subject db_material.grade_level = material.grade_level db_material.tags = material.tags db_material.is_public = material.is_public self.db.commit() self.db.refresh(db_material) return db_material def increment_usage(self, material_id: str) -> Optional[PhaseMaterialDB]: """Erhoeht den Usage-Counter eines Materials.""" db_material = self.get_by_id(material_id) if not db_material: return None db_material.usage_count += 1 self.db.commit() self.db.refresh(db_material) return db_material def attach_to_session( self, material_id: str, session_id: str ) -> Optional[PhaseMaterialDB]: """Verknuepft ein Material mit einer Session.""" db_material = self.get_by_id(material_id) if not db_material: return None db_material.session_id = session_id db_material.usage_count += 1 self.db.commit() self.db.refresh(db_material) return db_material # ==================== DELETE ==================== def delete(self, material_id: str) -> bool: """Loescht ein Material.""" db_material = self.get_by_id(material_id) if not db_material: return False self.db.delete(db_material) self.db.commit() return True # ==================== CONVERSION ==================== def to_dataclass(self, db_material: PhaseMaterialDB) -> PhaseMaterial: """Konvertiert DB-Model zu Dataclass.""" return PhaseMaterial( material_id=db_material.id, teacher_id=db_material.teacher_id, title=db_material.title, material_type=MaterialType(db_material.material_type.value), url=db_material.url, description=db_material.description or "", phase=db_material.phase, subject=db_material.subject or "", grade_level=db_material.grade_level or "", tags=db_material.tags or [], is_public=db_material.is_public, usage_count=db_material.usage_count, session_id=db_material.session_id, created_at=db_material.created_at, updated_at=db_material.updated_at, ) # ==================== REFLECTION REPOSITORY (Phase 5) ==================== class ReflectionRepository: """Repository fuer LessonReflection CRUD-Operationen.""" def __init__(self, db: DBSession): self.db = db # ==================== CREATE ==================== def create(self, reflection: LessonReflection) -> LessonReflectionDB: """Erstellt eine neue Reflection.""" db_reflection = LessonReflectionDB( id=reflection.reflection_id, session_id=reflection.session_id, teacher_id=reflection.teacher_id, notes=reflection.notes, overall_rating=reflection.overall_rating, what_worked=reflection.what_worked, improvements=reflection.improvements, notes_for_next_lesson=reflection.notes_for_next_lesson, ) self.db.add(db_reflection) self.db.commit() self.db.refresh(db_reflection) return db_reflection # ==================== READ ==================== def get_by_id(self, reflection_id: str) -> Optional[LessonReflectionDB]: """Holt eine Reflection nach ID.""" return self.db.query(LessonReflectionDB).filter( LessonReflectionDB.id == reflection_id ).first() def get_by_session(self, session_id: str) -> Optional[LessonReflectionDB]: """Holt die Reflection einer Session.""" return self.db.query(LessonReflectionDB).filter( LessonReflectionDB.session_id == session_id ).first() def get_by_teacher( self, teacher_id: str, limit: int = 20, offset: int = 0 ) -> List[LessonReflectionDB]: """Holt alle Reflections eines Lehrers.""" return self.db.query(LessonReflectionDB).filter( LessonReflectionDB.teacher_id == teacher_id ).order_by( LessonReflectionDB.created_at.desc() ).offset(offset).limit(limit).all() # ==================== UPDATE ==================== def update(self, reflection: LessonReflection) -> Optional[LessonReflectionDB]: """Aktualisiert eine Reflection.""" db_reflection = self.get_by_id(reflection.reflection_id) if not db_reflection: return None db_reflection.notes = reflection.notes db_reflection.overall_rating = reflection.overall_rating db_reflection.what_worked = reflection.what_worked db_reflection.improvements = reflection.improvements db_reflection.notes_for_next_lesson = reflection.notes_for_next_lesson self.db.commit() self.db.refresh(db_reflection) return db_reflection # ==================== DELETE ==================== def delete(self, reflection_id: str) -> bool: """Loescht eine Reflection.""" db_reflection = self.get_by_id(reflection_id) if not db_reflection: return False self.db.delete(db_reflection) self.db.commit() return True # ==================== CONVERSION ==================== def to_dataclass(self, db_reflection: LessonReflectionDB) -> LessonReflection: """Konvertiert DB-Model zu Dataclass.""" return LessonReflection( reflection_id=db_reflection.id, session_id=db_reflection.session_id, teacher_id=db_reflection.teacher_id, notes=db_reflection.notes or "", overall_rating=db_reflection.overall_rating, what_worked=db_reflection.what_worked or [], improvements=db_reflection.improvements or [], notes_for_next_lesson=db_reflection.notes_for_next_lesson or "", created_at=db_reflection.created_at, updated_at=db_reflection.updated_at, ) # ==================== ANALYTICS REPOSITORY (Phase 5) ==================== class AnalyticsRepository: """Repository fuer Analytics-Abfragen.""" def __init__(self, db: DBSession): self.db = db def get_session_summary(self, session_id: str) -> Optional[SessionSummary]: """ Berechnet die Summary einer abgeschlossenen Session. Args: session_id: ID der Session Returns: SessionSummary oder None wenn Session nicht gefunden """ db_session = self.db.query(LessonSessionDB).filter( LessonSessionDB.id == session_id ).first() if not db_session: return None # Session-Daten zusammenstellen session_data = { "session_id": db_session.id, "teacher_id": db_session.teacher_id, "class_id": db_session.class_id, "subject": db_session.subject, "topic": db_session.topic, "lesson_started_at": db_session.lesson_started_at, "lesson_ended_at": db_session.lesson_ended_at, "phase_durations": db_session.phase_durations or {}, } # Phase History aus DB oder JSON phase_history = db_session.phase_history or [] # Summary berechnen return AnalyticsCalculator.calculate_session_summary( session_data, phase_history ) def get_teacher_analytics( self, teacher_id: str, period_start: Optional[datetime] = None, period_end: Optional[datetime] = None ) -> TeacherAnalytics: """ Berechnet aggregierte Statistiken fuer einen Lehrer. Args: teacher_id: ID des Lehrers period_start: Beginn des Zeitraums (default: 30 Tage zurueck) period_end: Ende des Zeitraums (default: jetzt) Returns: TeacherAnalytics mit aggregierten Statistiken """ from datetime import timedelta if not period_end: period_end = datetime.utcnow() if not period_start: period_start = period_end - timedelta(days=30) # Sessions im Zeitraum abfragen sessions_query = self.db.query(LessonSessionDB).filter( LessonSessionDB.teacher_id == teacher_id, LessonSessionDB.lesson_started_at >= period_start, LessonSessionDB.lesson_started_at <= period_end ).all() # Sessions zu Dictionaries konvertieren sessions_data = [] for db_session in sessions_query: sessions_data.append({ "session_id": db_session.id, "teacher_id": db_session.teacher_id, "class_id": db_session.class_id, "subject": db_session.subject, "topic": db_session.topic, "lesson_started_at": db_session.lesson_started_at, "lesson_ended_at": db_session.lesson_ended_at, "phase_durations": db_session.phase_durations or {}, "phase_history": db_session.phase_history or [], }) return AnalyticsCalculator.calculate_teacher_analytics( sessions_data, period_start, period_end ) def get_phase_duration_trends( self, teacher_id: str, phase: str, limit: int = 20 ) -> List[Dict[str, Any]]: """ Gibt die Dauer-Trends fuer eine bestimmte Phase zurueck. Args: teacher_id: ID des Lehrers phase: Phasen-ID (einstieg, erarbeitung, etc.) limit: Max Anzahl der Datenpunkte Returns: Liste von Datenpunkten [{date, planned, actual, difference}] """ sessions = self.db.query(LessonSessionDB).filter( LessonSessionDB.teacher_id == teacher_id, LessonSessionDB.current_phase == LessonPhaseEnum.ENDED ).order_by( LessonSessionDB.lesson_ended_at.desc() ).limit(limit).all() trends = [] for db_session in sessions: history = db_session.phase_history or [] for entry in history: if entry.get("phase") == phase: planned = (db_session.phase_durations or {}).get(phase, 0) * 60 actual = entry.get("duration_seconds", 0) or 0 trends.append({ "date": db_session.lesson_started_at.isoformat() if db_session.lesson_started_at else None, "session_id": db_session.id, "subject": db_session.subject, "planned_seconds": planned, "actual_seconds": actual, "difference_seconds": actual - planned, }) break return list(reversed(trends)) # Chronologisch sortieren def get_overtime_analysis( self, teacher_id: str, limit: int = 30 ) -> Dict[str, Any]: """ Analysiert Overtime-Muster. Args: teacher_id: ID des Lehrers limit: Anzahl der zu analysierenden Sessions Returns: Dict mit Overtime-Statistiken pro Phase """ sessions = self.db.query(LessonSessionDB).filter( LessonSessionDB.teacher_id == teacher_id, LessonSessionDB.current_phase == LessonPhaseEnum.ENDED ).order_by( LessonSessionDB.lesson_ended_at.desc() ).limit(limit).all() phase_overtime: Dict[str, List[int]] = { "einstieg": [], "erarbeitung": [], "sicherung": [], "transfer": [], "reflexion": [], } for db_session in sessions: history = db_session.phase_history or [] phase_durations = db_session.phase_durations or {} for entry in history: phase = entry.get("phase", "") if phase in phase_overtime: planned = phase_durations.get(phase, 0) * 60 actual = entry.get("duration_seconds", 0) or 0 overtime = max(0, actual - planned) phase_overtime[phase].append(overtime) # Statistiken berechnen result = {} for phase, overtimes in phase_overtime.items(): if overtimes: result[phase] = { "count": len([o for o in overtimes if o > 0]), "total": len(overtimes), "avg_overtime_seconds": sum(overtimes) / len(overtimes), "max_overtime_seconds": max(overtimes), "overtime_percentage": len([o for o in overtimes if o > 0]) / len(overtimes) * 100, } else: result[phase] = { "count": 0, "total": 0, "avg_overtime_seconds": 0, "max_overtime_seconds": 0, "overtime_percentage": 0, } return result # ==================== TEACHER FEEDBACK REPOSITORY (Phase 7) ==================== class TeacherFeedbackRepository: """ Repository fuer Lehrer-Feedback CRUD-Operationen. Ermoeglicht Lehrern, Feedback (Bugs, Feature-Requests, Verbesserungen) direkt aus dem Lehrer-Frontend zu senden. """ def __init__(self, db: DBSession): self.db = db def create( self, teacher_id: str, title: str, description: str, feedback_type: str = "improvement", priority: str = "medium", teacher_name: str = "", teacher_email: str = "", context_url: str = "", context_phase: str = "", context_session_id: str = None, user_agent: str = "", related_feature: str = None, ) -> TeacherFeedbackDB: """Erstellt neues Feedback.""" import uuid db_feedback = TeacherFeedbackDB( id=str(uuid.uuid4()), teacher_id=teacher_id, teacher_name=teacher_name, teacher_email=teacher_email, title=title, description=description, feedback_type=FeedbackTypeEnum(feedback_type), priority=FeedbackPriorityEnum(priority), status=FeedbackStatusEnum.NEW, related_feature=related_feature, context_url=context_url, context_phase=context_phase, context_session_id=context_session_id, user_agent=user_agent, ) self.db.add(db_feedback) self.db.commit() self.db.refresh(db_feedback) return db_feedback def get_by_id(self, feedback_id: str) -> Optional[TeacherFeedbackDB]: """Holt Feedback nach ID.""" return self.db.query(TeacherFeedbackDB).filter( TeacherFeedbackDB.id == feedback_id ).first() def get_all( self, status: str = None, feedback_type: str = None, limit: int = 100, offset: int = 0 ) -> List[TeacherFeedbackDB]: """Holt alle Feedbacks mit optionalen Filtern.""" query = self.db.query(TeacherFeedbackDB) if status: query = query.filter(TeacherFeedbackDB.status == FeedbackStatusEnum(status)) if feedback_type: query = query.filter(TeacherFeedbackDB.feedback_type == FeedbackTypeEnum(feedback_type)) return query.order_by( TeacherFeedbackDB.created_at.desc() ).offset(offset).limit(limit).all() def get_by_teacher(self, teacher_id: str, limit: int = 50) -> List[TeacherFeedbackDB]: """Holt Feedback eines bestimmten Lehrers.""" return self.db.query(TeacherFeedbackDB).filter( TeacherFeedbackDB.teacher_id == teacher_id ).order_by( TeacherFeedbackDB.created_at.desc() ).limit(limit).all() def update_status( self, feedback_id: str, status: str, response: str = None, responded_by: str = None ) -> Optional[TeacherFeedbackDB]: """Aktualisiert den Status eines Feedbacks.""" db_feedback = self.get_by_id(feedback_id) if not db_feedback: return None db_feedback.status = FeedbackStatusEnum(status) if response: db_feedback.response = response db_feedback.responded_at = datetime.utcnow() db_feedback.responded_by = responded_by self.db.commit() self.db.refresh(db_feedback) return db_feedback def delete(self, feedback_id: str) -> bool: """Loescht ein Feedback.""" db_feedback = self.get_by_id(feedback_id) if not db_feedback: return False self.db.delete(db_feedback) self.db.commit() return True def get_stats(self) -> Dict[str, Any]: """Gibt Statistiken ueber alle Feedbacks zurueck.""" all_feedback = self.db.query(TeacherFeedbackDB).all() stats = { "total": len(all_feedback), "by_status": {}, "by_type": {}, "by_priority": {}, } for fb in all_feedback: # By Status status = fb.status.value stats["by_status"][status] = stats["by_status"].get(status, 0) + 1 # By Type fb_type = fb.feedback_type.value stats["by_type"][fb_type] = stats["by_type"].get(fb_type, 0) + 1 # By Priority priority = fb.priority.value stats["by_priority"][priority] = stats["by_priority"].get(priority, 0) + 1 return stats def to_dict(self, db_feedback: TeacherFeedbackDB) -> Dict[str, Any]: """Konvertiert DB-Model zu Dictionary.""" return { "id": db_feedback.id, "teacher_id": db_feedback.teacher_id, "teacher_name": db_feedback.teacher_name, "teacher_email": db_feedback.teacher_email, "title": db_feedback.title, "description": db_feedback.description, "feedback_type": db_feedback.feedback_type.value, "priority": db_feedback.priority.value, "status": db_feedback.status.value, "related_feature": db_feedback.related_feature, "context_url": db_feedback.context_url, "context_phase": db_feedback.context_phase, "context_session_id": db_feedback.context_session_id, "user_agent": db_feedback.user_agent, "response": db_feedback.response, "responded_at": db_feedback.responded_at.isoformat() if db_feedback.responded_at else None, "responded_by": db_feedback.responded_by, "created_at": db_feedback.created_at.isoformat() if db_feedback.created_at else None, "updated_at": db_feedback.updated_at.isoformat() if db_feedback.updated_at else None, } # ==================== Phase 8: Teacher Context Repository ==================== class TeacherContextRepository: """Repository fuer Lehrer-Kontext CRUD-Operationen (Phase 8).""" def __init__(self, db: DBSession): self.db = db # ==================== CREATE / GET-OR-CREATE ==================== def get_or_create(self, teacher_id: str) -> TeacherContextDB: """ Holt den Kontext eines Lehrers oder erstellt einen neuen. Args: teacher_id: ID des Lehrers Returns: TeacherContextDB Model """ context = self.get_by_teacher_id(teacher_id) if context: return context # Neuen Kontext erstellen from uuid import uuid4 context = TeacherContextDB( id=str(uuid4()), teacher_id=teacher_id, macro_phase=MacroPhaseEnum.ONBOARDING, ) self.db.add(context) self.db.commit() self.db.refresh(context) return context # ==================== READ ==================== def get_by_teacher_id(self, teacher_id: str) -> Optional[TeacherContextDB]: """Holt den Kontext eines Lehrers.""" return self.db.query(TeacherContextDB).filter( TeacherContextDB.teacher_id == teacher_id ).first() # ==================== UPDATE ==================== def update_context( self, teacher_id: str, federal_state: str = None, school_type: str = None, schoolyear: str = None, schoolyear_start: datetime = None, macro_phase: str = None, current_week: int = None, ) -> Optional[TeacherContextDB]: """Aktualisiert den Kontext eines Lehrers.""" context = self.get_or_create(teacher_id) if federal_state is not None: context.federal_state = federal_state if school_type is not None: context.school_type = school_type if schoolyear is not None: context.schoolyear = schoolyear if schoolyear_start is not None: context.schoolyear_start = schoolyear_start if macro_phase is not None: context.macro_phase = MacroPhaseEnum(macro_phase) if current_week is not None: context.current_week = current_week self.db.commit() self.db.refresh(context) return context def complete_onboarding(self, teacher_id: str) -> TeacherContextDB: """Markiert Onboarding als abgeschlossen.""" context = self.get_or_create(teacher_id) context.onboarding_completed = True context.macro_phase = MacroPhaseEnum.SCHULJAHRESSTART self.db.commit() self.db.refresh(context) return context def update_flags( self, teacher_id: str, has_classes: bool = None, has_schedule: bool = None, is_exam_period: bool = None, is_before_holidays: bool = None, ) -> TeacherContextDB: """Aktualisiert die Status-Flags eines Kontexts.""" context = self.get_or_create(teacher_id) if has_classes is not None: context.has_classes = has_classes if has_schedule is not None: context.has_schedule = has_schedule if is_exam_period is not None: context.is_exam_period = is_exam_period if is_before_holidays is not None: context.is_before_holidays = is_before_holidays self.db.commit() self.db.refresh(context) return context def to_dict(self, context: TeacherContextDB) -> Dict[str, Any]: """Konvertiert DB-Model zu Dictionary.""" return { "id": context.id, "teacher_id": context.teacher_id, "school": { "federal_state": context.federal_state, "federal_state_name": FEDERAL_STATES.get(context.federal_state, ""), "school_type": context.school_type, "school_type_name": SCHOOL_TYPES.get(context.school_type, ""), }, "school_year": { "id": context.schoolyear, "start": context.schoolyear_start.isoformat() if context.schoolyear_start else None, "current_week": context.current_week, }, "macro_phase": { "id": context.macro_phase.value, "label": self._get_phase_label(context.macro_phase), }, "flags": { "onboarding_completed": context.onboarding_completed, "has_classes": context.has_classes, "has_schedule": context.has_schedule, "is_exam_period": context.is_exam_period, "is_before_holidays": context.is_before_holidays, }, "created_at": context.created_at.isoformat() if context.created_at else None, "updated_at": context.updated_at.isoformat() if context.updated_at else None, } def _get_phase_label(self, phase: MacroPhaseEnum) -> str: """Gibt den Anzeigenamen einer Makro-Phase zurueck.""" labels = { MacroPhaseEnum.ONBOARDING: "Einrichtung", MacroPhaseEnum.SCHULJAHRESSTART: "Schuljahresstart", MacroPhaseEnum.UNTERRICHTSAUFBAU: "Unterrichtsaufbau", MacroPhaseEnum.LEISTUNGSPHASE_1: "Leistungsphase 1", MacroPhaseEnum.HALBJAHRESABSCHLUSS: "Halbjahresabschluss", MacroPhaseEnum.LEISTUNGSPHASE_2: "Leistungsphase 2", MacroPhaseEnum.JAHRESABSCHLUSS: "Jahresabschluss", } return labels.get(phase, phase.value) # ==================== Phase 8: Schoolyear Event Repository ==================== class SchoolyearEventRepository: """Repository fuer Schuljahr-Events (Phase 8).""" def __init__(self, db: DBSession): self.db = db def create( self, teacher_id: str, title: str, start_date: datetime, event_type: str = "other", end_date: datetime = None, class_id: str = None, subject: str = None, description: str = "", needs_preparation: bool = True, reminder_days_before: int = 7, extra_data: Dict[str, Any] = None, ) -> SchoolyearEventDB: """Erstellt ein neues Schuljahr-Event.""" from uuid import uuid4 event = SchoolyearEventDB( id=str(uuid4()), teacher_id=teacher_id, title=title, event_type=EventTypeEnum(event_type), start_date=start_date, end_date=end_date, class_id=class_id, subject=subject, description=description, needs_preparation=needs_preparation, reminder_days_before=reminder_days_before, extra_data=extra_data or {}, ) self.db.add(event) self.db.commit() self.db.refresh(event) return event def get_by_id(self, event_id: str) -> Optional[SchoolyearEventDB]: """Holt ein Event nach ID.""" return self.db.query(SchoolyearEventDB).filter( SchoolyearEventDB.id == event_id ).first() def get_by_teacher( self, teacher_id: str, status: str = None, event_type: str = None, limit: int = 50, ) -> List[SchoolyearEventDB]: """Holt Events eines Lehrers.""" query = self.db.query(SchoolyearEventDB).filter( SchoolyearEventDB.teacher_id == teacher_id ) if status: query = query.filter(SchoolyearEventDB.status == EventStatusEnum(status)) if event_type: query = query.filter(SchoolyearEventDB.event_type == EventTypeEnum(event_type)) return query.order_by(SchoolyearEventDB.start_date).limit(limit).all() def get_upcoming( self, teacher_id: str, days: int = 30, limit: int = 10, ) -> List[SchoolyearEventDB]: """Holt anstehende Events der naechsten X Tage.""" from datetime import timedelta now = datetime.utcnow() end = now + timedelta(days=days) return self.db.query(SchoolyearEventDB).filter( SchoolyearEventDB.teacher_id == teacher_id, SchoolyearEventDB.start_date >= now, SchoolyearEventDB.start_date <= end, SchoolyearEventDB.status != EventStatusEnum.CANCELLED, ).order_by(SchoolyearEventDB.start_date).limit(limit).all() def update_status( self, event_id: str, status: str, preparation_done: bool = None, ) -> Optional[SchoolyearEventDB]: """Aktualisiert den Status eines Events.""" event = self.get_by_id(event_id) if not event: return None event.status = EventStatusEnum(status) if preparation_done is not None: event.preparation_done = preparation_done self.db.commit() self.db.refresh(event) return event def delete(self, event_id: str) -> bool: """Loescht ein Event.""" event = self.get_by_id(event_id) if not event: return False self.db.delete(event) self.db.commit() return True def to_dict(self, event: SchoolyearEventDB) -> Dict[str, Any]: """Konvertiert DB-Model zu Dictionary.""" return { "id": event.id, "teacher_id": event.teacher_id, "event_type": event.event_type.value, "title": event.title, "description": event.description, "start_date": event.start_date.isoformat() if event.start_date else None, "end_date": event.end_date.isoformat() if event.end_date else None, "class_id": event.class_id, "subject": event.subject, "status": event.status.value, "needs_preparation": event.needs_preparation, "preparation_done": event.preparation_done, "reminder_days_before": event.reminder_days_before, "extra_data": event.extra_data, "created_at": event.created_at.isoformat() if event.created_at else None, } # ==================== Phase 8: Recurring Routine Repository ==================== class RecurringRoutineRepository: """Repository fuer wiederkehrende Routinen (Phase 8).""" def __init__(self, db: DBSession): self.db = db def create( self, teacher_id: str, title: str, routine_type: str = "other", recurrence_pattern: str = "weekly", day_of_week: int = None, day_of_month: int = None, time_of_day: str = None, # Format: "14:00" duration_minutes: int = 60, description: str = "", valid_from: datetime = None, valid_until: datetime = None, ) -> RecurringRoutineDB: """Erstellt eine neue wiederkehrende Routine.""" from uuid import uuid4 from datetime import time as dt_time time_obj = None if time_of_day: parts = time_of_day.split(":") time_obj = dt_time(int(parts[0]), int(parts[1])) routine = RecurringRoutineDB( id=str(uuid4()), teacher_id=teacher_id, title=title, routine_type=RoutineTypeEnum(routine_type), recurrence_pattern=RecurrencePatternEnum(recurrence_pattern), day_of_week=day_of_week, day_of_month=day_of_month, time_of_day=time_obj, duration_minutes=duration_minutes, description=description, valid_from=valid_from, valid_until=valid_until, ) self.db.add(routine) self.db.commit() self.db.refresh(routine) return routine def get_by_id(self, routine_id: str) -> Optional[RecurringRoutineDB]: """Holt eine Routine nach ID.""" return self.db.query(RecurringRoutineDB).filter( RecurringRoutineDB.id == routine_id ).first() def get_by_teacher( self, teacher_id: str, is_active: bool = True, routine_type: str = None, ) -> List[RecurringRoutineDB]: """Holt Routinen eines Lehrers.""" query = self.db.query(RecurringRoutineDB).filter( RecurringRoutineDB.teacher_id == teacher_id ) if is_active is not None: query = query.filter(RecurringRoutineDB.is_active == is_active) if routine_type: query = query.filter(RecurringRoutineDB.routine_type == RoutineTypeEnum(routine_type)) return query.all() def get_today(self, teacher_id: str) -> List[RecurringRoutineDB]: """Holt Routinen die heute stattfinden.""" today = datetime.utcnow() day_of_week = today.weekday() # 0 = Montag day_of_month = today.day routines = self.get_by_teacher(teacher_id, is_active=True) today_routines = [] for routine in routines: if routine.recurrence_pattern == RecurrencePatternEnum.DAILY: today_routines.append(routine) elif routine.recurrence_pattern == RecurrencePatternEnum.WEEKLY: if routine.day_of_week == day_of_week: today_routines.append(routine) elif routine.recurrence_pattern == RecurrencePatternEnum.BIWEEKLY: # Vereinfacht: Pruefen ob Tag passt (echte Logik braucht Startdatum) if routine.day_of_week == day_of_week: today_routines.append(routine) elif routine.recurrence_pattern == RecurrencePatternEnum.MONTHLY: if routine.day_of_month == day_of_month: today_routines.append(routine) return today_routines def update( self, routine_id: str, title: str = None, is_active: bool = None, day_of_week: int = None, time_of_day: str = None, ) -> Optional[RecurringRoutineDB]: """Aktualisiert eine Routine.""" routine = self.get_by_id(routine_id) if not routine: return None if title is not None: routine.title = title if is_active is not None: routine.is_active = is_active if day_of_week is not None: routine.day_of_week = day_of_week if time_of_day is not None: from datetime import time as dt_time parts = time_of_day.split(":") routine.time_of_day = dt_time(int(parts[0]), int(parts[1])) self.db.commit() self.db.refresh(routine) return routine def delete(self, routine_id: str) -> bool: """Loescht eine Routine.""" routine = self.get_by_id(routine_id) if not routine: return False self.db.delete(routine) self.db.commit() return True def to_dict(self, routine: RecurringRoutineDB) -> Dict[str, Any]: """Konvertiert DB-Model zu Dictionary.""" return { "id": routine.id, "teacher_id": routine.teacher_id, "routine_type": routine.routine_type.value, "title": routine.title, "description": routine.description, "recurrence_pattern": routine.recurrence_pattern.value, "day_of_week": routine.day_of_week, "day_of_month": routine.day_of_month, "time_of_day": routine.time_of_day.isoformat() if routine.time_of_day else None, "duration_minutes": routine.duration_minutes, "is_active": routine.is_active, "valid_from": routine.valid_from.isoformat() if routine.valid_from else None, "valid_until": routine.valid_until.isoformat() if routine.valid_until else None, "created_at": routine.created_at.isoformat() if routine.created_at else None, }