21a844cb8a
A previous `git pull --rebase origin main` dropped 177 local commits,
losing 3400+ files across admin-v2, backend, studio-v2, website,
klausur-service, and many other services. The partial restore attempt
(660295e2) only recovered some files.
This commit restores all missing files from pre-rebase ref 98933f5e
while preserving post-rebase additions (night-scheduler, night-mode UI,
NightModeWidget dashboard integration).
Restored features include:
- AI Module Sidebar (FAB), OCR Labeling, OCR Compare
- GPU Dashboard, RAG Pipeline, Magic Help
- Klausur-Korrektur (8 files), Abitur-Archiv (5+ files)
- Companion, Zeugnisse-Crawler, Screen Flow
- Full backend, studio-v2, website, klausur-service
- All compliance SDKs, agent-core, voice-service
- CI/CD configs, documentation, scripts
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1706 lines
58 KiB
Python
1706 lines
58 KiB
Python
"""
|
|
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,
|
|
}
|