107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Business Rules – Validierungsregeln für Abwesenheiten.
|
||
|
||
Jede Regel ist eine Funktion die (db, AbsenceCreate) nimmt und
|
||
bei Verstoss eine RuleViolation raised.
|
||
"""
|
||
|
||
import datetime
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..core.schemas import AbsenceCreate
|
||
from ..core.enums import ABSENCE_META
|
||
from ..employees.repository import EmployeeRepository
|
||
from .repository import AbsenceRepository, BalanceRepository
|
||
from ..config import MIN_STAFF_COUNT
|
||
|
||
|
||
class RuleViolation(Exception):
|
||
"""Wird geworfen wenn eine Business Rule verletzt ist."""
|
||
def __init__(self, rule: str, message: str):
|
||
self.rule = rule
|
||
self.message = message
|
||
super().__init__(f"[{rule}] {message}")
|
||
|
||
|
||
def _business_days(start: datetime.date, end: datetime.date) -> float:
|
||
count = 0
|
||
d = start
|
||
while d <= end:
|
||
if d.weekday() < 5:
|
||
count += 1
|
||
d += datetime.timedelta(days=1)
|
||
return count
|
||
|
||
|
||
def check_no_overlap(db: Session, data: AbsenceCreate):
|
||
"""Keine überlappenden Abwesenheiten für denselben Mitarbeiter."""
|
||
repo = AbsenceRepository(db)
|
||
overlaps = repo.find_overlapping(data.employee_id, data.start_date, data.end_date)
|
||
if overlaps:
|
||
first = overlaps[0]
|
||
raise RuleViolation(
|
||
"NO_OVERLAP",
|
||
f"Überschneidung mit bestehender Abwesenheit "
|
||
f"({first.start_date} – {first.end_date}, {first.category.value})"
|
||
)
|
||
|
||
|
||
def check_balance(db: Session, data: AbsenceCreate):
|
||
"""Ferientage-Kontingent prüfen (nur für Kategorien mit deducts_balance)."""
|
||
meta = ABSENCE_META.get(data.category, {})
|
||
if not meta.get("deducts_balance"):
|
||
return
|
||
|
||
year = data.start_date.year
|
||
abs_repo = AbsenceRepository(db)
|
||
bal_repo = BalanceRepository(db)
|
||
|
||
balance = bal_repo.get_or_create(data.employee_id, year)
|
||
used = abs_repo.used_days(data.employee_id, year)
|
||
needed = _business_days(data.start_date, data.end_date)
|
||
remaining = balance.total_available - used
|
||
|
||
if needed > remaining:
|
||
raise RuleViolation(
|
||
"INSUFFICIENT_BALANCE",
|
||
f"Nicht genügend Ferientage. Benötigt: {needed:.0f}, "
|
||
f"Verfügbar: {remaining:.0f} (Anspruch: {balance.total_available:.0f}, "
|
||
f"Genommen: {used:.0f})"
|
||
)
|
||
|
||
|
||
def check_min_staffing(db: Session, data: AbsenceCreate):
|
||
"""Mindestbesetzung prüfen – nicht alle gleichzeitig abwesend."""
|
||
emp_repo = EmployeeRepository(db)
|
||
abs_repo = AbsenceRepository(db)
|
||
|
||
total_active = emp_repo.count_active()
|
||
if total_active <= MIN_STAFF_COUNT:
|
||
return
|
||
|
||
d = data.start_date
|
||
while d <= data.end_date:
|
||
if d.weekday() < 5:
|
||
absent_count = abs_repo.count_absent_on_date(d)
|
||
present = total_active - absent_count - 1
|
||
if present < MIN_STAFF_COUNT:
|
||
raise RuleViolation(
|
||
"MIN_STAFFING",
|
||
f"Am {d.strftime('%d.%m.%Y')} wären nur noch {present} "
|
||
f"Mitarbeiter anwesend (Minimum: {MIN_STAFF_COUNT})"
|
||
)
|
||
d += datetime.timedelta(days=1)
|
||
|
||
|
||
ALL_RULES = [check_no_overlap, check_balance, check_min_staffing]
|
||
|
||
|
||
def validate_absence(db: Session, data: AbsenceCreate, skip_rules: list[str] | None = None):
|
||
"""Alle Regeln durchlaufen. Wirft RuleViolation beim ersten Verstoss."""
|
||
skip = set(skip_rules or [])
|
||
for rule_fn in ALL_RULES:
|
||
if rule_fn.__name__ in skip:
|
||
continue
|
||
rule_fn(db, data)
|