scripte/timemaster
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Sondervertretungs-Faktoren (special_assignments)

Browse Source 
- Neues Model SpecialAssignment mit AssignmentMode (fza|payroll|both)
- CRUD-Endpunkte unter /users/{id}/special-assignments
- Payroll-Report: GET /reports/special-assignments/payroll?year=&month=
- Migration 0029: special_assignments Tabelle + btree_gist Overlap-Constraint
- _recalculate_overtime_balance berücksichtigt FZA-Faktoren
- Frontend: Sondervertretungs-Zeiträume im UsersPage Edit-Modal
- Frontend: ReportsPage neuer Tab 'Sondervertretungen' mit Payroll-Tabelle + CSV-Export

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
patrick
2026-05-25 00:55:47 +02:00
parent 1170e59e49
commit d60349df67
12 changed files with 837 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/main.py
+2
View File
@@ -15,6 +15,7 @@ from app.routers import import_kimai
from app.routers import kiosk
from app.routers import busylight
from app.routers import audit
from app.routers import special_assignments
@asynccontextmanager
@@ -77,6 +78,7 @@ app.include_router(import_kimai.router, prefix=API_PREFIX)
app.include_router(kiosk.router, prefix=API_PREFIX)
app.include_router(busylight.router, prefix=API_PREFIX)
app.include_router(audit.router, prefix=API_PREFIX)
app.include_router(special_assignments.router, prefix=API_PREFIX)
# ── Health ────────────────────────────────────────────────────────────────────
backend/app/models/__init__.py
+3
View File
@@ -14,6 +14,7 @@ from app.models.public_holiday import PublicHoliday
from app.models.smtp_config import SmtpConfig
from app.models.caldav_config import CaldavCompanyConfig, CaldavUserConfig
from app.models.kiosk_device import KioskDevice, KioskAuthMethod
from app.models.special_assignment import SpecialAssignment, AssignmentMode
__all__ = [
"Company",
@@ -35,4 +36,6 @@ __all__ = [
"PublicHoliday",
"KioskDevice",
"KioskAuthMethod",
"SpecialAssignment",
"AssignmentMode",
]
backend/app/models/special_assignment.py
+38
View File
@@ -0,0 +1,38 @@
"""Sondervertretungs-Faktoren: Per-User Zeitraum-Zuweisung mit Multiplikator."""
import enum
from datetime import date
from decimal import Decimal
from uuid import UUID, uuid4
from sqlalchemy import CheckConstraint, Date, Enum, ForeignKey, Numeric, String, Text
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
class AssignmentMode(str, enum.Enum):
fza = "fza" # Nur FZA-Stunden-Anrechnung
payroll = "payroll" # Nur Gehaltsabrechnung (Bericht)
both = "both" # Beides
class SpecialAssignment(Base):
"""Sondervertretungs-Zeitraum mit Faktor für einen Mitarbeiter."""
__tablename__ = "special_assignments"
__table_args__ = (
CheckConstraint("factor > 0 AND factor <= 10", name="ck_special_assignment_factor"),
CheckConstraint("date_from <= date_to", name="ck_special_assignment_dates"),
)
id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True, default=uuid4)
user_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
company_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), ForeignKey("companies.id", ondelete="CASCADE"), nullable=False, index=True)
date_from: Mapped[date] = mapped_column(Date, nullable=False)
date_to: Mapped[date] = mapped_column(Date, nullable=False)
factor: Mapped[Decimal] = mapped_column(Numeric(5, 3), nullable=False)
mode: Mapped[AssignmentMode] = mapped_column(Enum(AssignmentMode, name="assignment_mode"), nullable=False, default=AssignmentMode.both)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
label: Mapped[str | None] = mapped_column(String(100), nullable=True) # z.B. "Schichtleiter Vertretung"
backend/app/routers/special_assignments.py
+246
View File
@@ -0,0 +1,246 @@
"""CRUD für Sondervertretungs-Zeiträume (Special Assignments)."""
from datetime import date
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.dependencies import require_role
from app.models.special_assignment import AssignmentMode, SpecialAssignment
from app.models.time_entry import EntryStatus, TimeEntry
from app.models.user import User, UserRole
from app.schemas.special_assignment import (
PayrollAssignmentEntry,
PayrollAssignmentReport,
PayrollAssignmentRow,
SpecialAssignmentCreate,
SpecialAssignmentOut,
SpecialAssignmentUpdate,
)
router = APIRouter(tags=["Sondervertretungen"])
_manager_roles = (UserRole.MANAGER, UserRole.HR, UserRole.COMPANY_ADMIN, UserRole.SUPER_ADMIN)
async def _get_assignment_or_404(
assignment_id: UUID, company_id: UUID, db: AsyncSession
) -> SpecialAssignment:
sa = await db.get(SpecialAssignment, assignment_id)
if sa is None or sa.company_id != company_id:
raise HTTPException(404, "Zuweisung nicht gefunden")
return sa
async def _check_overlap(
user_id: UUID,
date_from: date,
date_to: date,
db: AsyncSession,
exclude_id: UUID | None = None,
) -> None:
"""Prüft ob sich der Zeitraum mit einer vorhandenen Zuweisung überschneidet."""
q = select(SpecialAssignment).where(
SpecialAssignment.user_id == user_id,
SpecialAssignment.date_from <= date_to,
SpecialAssignment.date_to >= date_from,
)
if exclude_id:
q = q.where(SpecialAssignment.id != exclude_id)
existing = await db.scalar(q)
if existing:
raise HTTPException(
409,
f"Zeitraum überschneidet sich mit vorhandener Zuweisung "
f"({existing.date_from}–{existing.date_to})",
)
# ── CRUD ──────────────────────────────────────────────────────────────────────
@router.get("/users/{user_id}/special-assignments", response_model=list[SpecialAssignmentOut])
async def list_assignments(
user_id: UUID,
current_user: User = Depends(require_role(*_manager_roles)),
db: AsyncSession = Depends(get_db),
):
target = await db.get(User, user_id)
if not target or target.company_id != current_user.company_id:
raise HTTPException(404, "Mitarbeiter nicht gefunden")
rows = list(await db.scalars(
select(SpecialAssignment)
.where(SpecialAssignment.user_id == user_id)
.order_by(SpecialAssignment.date_from)
))
return [SpecialAssignmentOut.model_validate(r) for r in rows]
@router.post("/users/{user_id}/special-assignments", response_model=SpecialAssignmentOut, status_code=201)
async def create_assignment(
user_id: UUID,
data: SpecialAssignmentCreate,
current_user: User = Depends(require_role(*_manager_roles)),
db: AsyncSession = Depends(get_db),
):
target = await db.get(User, user_id)
if not target or target.company_id != current_user.company_id:
raise HTTPException(404, "Mitarbeiter nicht gefunden")
await _check_overlap(user_id, data.date_from, data.date_to, db)
sa = SpecialAssignment(
user_id=user_id,
company_id=current_user.company_id,
**data.model_dump(),
)
db.add(sa)
await db.commit()
await db.refresh(sa)
return SpecialAssignmentOut.model_validate(sa)
@router.patch("/users/{user_id}/special-assignments/{assignment_id}", response_model=SpecialAssignmentOut)
async def update_assignment(
user_id: UUID,
assignment_id: UUID,
data: SpecialAssignmentUpdate,
current_user: User = Depends(require_role(*_manager_roles)),
db: AsyncSession = Depends(get_db),
):
sa = await _get_assignment_or_404(assignment_id, current_user.company_id, db)
if sa.user_id != user_id:
raise HTTPException(404, "Zuweisung nicht gefunden")
updates = data.model_dump(exclude_unset=True)
new_from = updates.get("date_from", sa.date_from)
new_to = updates.get("date_to", sa.date_to)
if new_from > new_to:
raise HTTPException(422, "date_from darf nicht nach date_to liegen")
await _check_overlap(user_id, new_from, new_to, db, exclude_id=assignment_id)
for field, value in updates.items():
setattr(sa, field, value)
await db.commit()
await db.refresh(sa)
return SpecialAssignmentOut.model_validate(sa)
@router.delete("/users/{user_id}/special-assignments/{assignment_id}", status_code=204)
async def delete_assignment(
user_id: UUID,
assignment_id: UUID,
current_user: User = Depends(require_role(*_manager_roles)),
db: AsyncSession = Depends(get_db),
):
sa = await _get_assignment_or_404(assignment_id, current_user.company_id, db)
if sa.user_id != user_id:
raise HTTPException(404, "Zuweisung nicht gefunden")
await db.delete(sa)
await db.commit()
# ── Payroll-Report ────────────────────────────────────────────────────────────
@router.get("/reports/special-assignments/payroll", response_model=PayrollAssignmentReport)
async def payroll_report(
year: int = Query(..., ge=2000, le=2100),
month: int = Query(..., ge=1, le=12),
current_user: User = Depends(require_role(*_manager_roles)),
db: AsyncSession = Depends(get_db),
):
"""Payroll-Report: Für jeden Mitarbeiter die Sondervertretungs-Stunden im Monat."""
from calendar import monthrange
first_day = date(year, month, 1)
last_day = date(year, month, monthrange(year, month)[1])
# Alle Zuweisungen der Firma, die den Monat überschneiden
assignments = list(await db.scalars(
select(SpecialAssignment).where(
SpecialAssignment.company_id == current_user.company_id,
SpecialAssignment.mode.in_([AssignmentMode.payroll, AssignmentMode.both]),
SpecialAssignment.date_from <= last_day,
SpecialAssignment.date_to >= first_day,
).order_by(SpecialAssignment.user_id, SpecialAssignment.date_from)
))
if not assignments:
return PayrollAssignmentReport(year=year, month=month, rows=[])
# User-IDs ermitteln
user_ids = list({a.user_id for a in assignments})
users_map: dict[UUID, User] = {}
for uid in user_ids:
u = await db.get(User, uid)
if u:
users_map[uid] = u
# Genehmigte Zeit-Einträge für betroffene User im Monat laden
entries = list(await db.scalars(
select(TimeEntry).where(
TimeEntry.user_id.in_(user_ids),
TimeEntry.date >= first_day,
TimeEntry.date <= last_day,
TimeEntry.status == EntryStatus.APPROVED,
TimeEntry.end_time.isnot(None),
)
))
# Gruppierung: user_id → list[TimeEntry]
entries_by_user: dict[UUID, list[TimeEntry]] = {}
for e in entries:
entries_by_user.setdefault(e.user_id, []).append(e)
rows: list[PayrollAssignmentRow] = []
# Groupby user_id
from itertools import groupby
sorted_assignments = sorted(assignments, key=lambda a: a.user_id)
for uid, user_assignments_iter in groupby(sorted_assignments, key=lambda a: a.user_id):
user_assignments = list(user_assignments_iter)
user = users_map.get(uid)
if not user:
continue
user_entries = entries_by_user.get(uid, [])
# Index: date → worked_hours
hours_by_date: dict[date, float] = {}
for e in user_entries:
hours_by_date[e.date] = hours_by_date.get(e.date, 0.0) + (e.worked_hours or 0.0)
assignment_entries: list[PayrollAssignmentEntry] = []
for sa in user_assignments:
# Effektiver Zeitraum innerhalb des Monats
eff_from = max(sa.date_from, first_day)
eff_to = min(sa.date_to, last_day)
# Stunden im Zeitraum summieren
normal_hours = sum(
hours for d, hours in hours_by_date.items()
if eff_from <= d <= eff_to
)
factor_hours = round(normal_hours * float(sa.factor), 2)
extra_hours = round(factor_hours - normal_hours, 2)
assignment_entries.append(PayrollAssignmentEntry(
assignment_id=sa.id,
label=sa.label,
date_from=eff_from,
date_to=eff_to,
factor=sa.factor,
normal_hours=round(normal_hours, 2),
factor_hours=factor_hours,
extra_hours=extra_hours,
))
total_normal = sum(e.normal_hours for e in assignment_entries)
total_factor = sum(e.factor_hours for e in assignment_entries)
rows.append(PayrollAssignmentRow(
user_id=uid,
user_name=f"{user.first_name} {user.last_name}",
personnel_number=getattr(user, "personnel_number", None),
assignments=assignment_entries,
total_normal_hours=round(total_normal, 2),
total_factor_hours=round(total_factor, 2),
total_extra_hours=round(total_factor - total_normal, 2),
))
return PayrollAssignmentReport(year=year, month=month, rows=rows)
backend/app/schemas/special_assignment.py
+76
View File
@@ -0,0 +1,76 @@
from datetime import date
from decimal import Decimal
from uuid import UUID
from pydantic import BaseModel, Field, model_validator
from app.models.special_assignment import AssignmentMode
class SpecialAssignmentCreate(BaseModel):
date_from: date
date_to: date
factor: Decimal = Field(gt=0, le=10, decimal_places=3)
mode: AssignmentMode = AssignmentMode.both
description: str | None = None
label: str | None = Field(None, max_length=100)
@model_validator(mode="after")
def dates_valid(self) -> "SpecialAssignmentCreate":
if self.date_from > self.date_to:
raise ValueError("date_from darf nicht nach date_to liegen")
return self
class SpecialAssignmentUpdate(BaseModel):
date_from: date | None = None
date_to: date | None = None
factor: Decimal | None = Field(None, gt=0, le=10)
mode: AssignmentMode | None = None
description: str | None = None
label: str | None = Field(None, max_length=100)
class SpecialAssignmentOut(BaseModel):
id: UUID
user_id: UUID
company_id: UUID
date_from: date
date_to: date
factor: Decimal
mode: AssignmentMode
description: str | None = None
label: str | None = None
model_config = {"from_attributes": True}
# ── Payroll-Report-Schemas ────────────────────────────────────────────────────
class PayrollAssignmentEntry(BaseModel):
"""Einzelner Zeitraum mit Faktor für den Payroll-Report."""
assignment_id: UUID
label: str | None
date_from: date
date_to: date
factor: Decimal
normal_hours: float # Stunden ohne Faktor
factor_hours: float # Stunden * Faktor (effektiv für Abrechnung)
extra_hours: float # factor_hours - normal_hours (Mehrwert)
class PayrollAssignmentRow(BaseModel):
"""Zusammenfassung pro Mitarbeiter."""
user_id: UUID
user_name: str
personnel_number: str | None
assignments: list[PayrollAssignmentEntry]
total_normal_hours: float
total_factor_hours: float
total_extra_hours: float
class PayrollAssignmentReport(BaseModel):
year: int
month: int
rows: list[PayrollAssignmentRow]
backend/app/services/report_service.py
+29 -5
View File
@@ -181,7 +181,14 @@ async def _get_or_create_overtime_balance(user: User, db: AsyncSession) -> Overt
async def _recalculate_overtime_balance(
user: User, schedule: WorkSchedule | None, db: AsyncSession
) -> OvertimeBalance:
"""Berechnet das Überstundenguthaben neu aus allen genehmigten Zeiteinträgen."""
"""Berechnet das Überstundenguthaben neu aus allen genehmigten Zeiteinträgen.
Sondervertretungs-Faktoren (mode=fza|both) werden berücksichtigt:
Für jeden Zeiteintrag wird geprüft ob der Tag in einem aktiven Sondervertretungs-
Zeitraum liegt; falls ja wird worked_hours mit dem Faktor multipliziert.
"""
from app.models.special_assignment import AssignmentMode, SpecialAssignment
entries = list(await db.scalars(
select(TimeEntry).where(
TimeEntry.user_id == user.id,
@@ -197,10 +204,27 @@ async def _recalculate_overtime_balance(
bal.last_calculated = datetime.utcnow()
return bal
date_from = min(e.date for e in entries)
date_to = max(e.date for e in entries)
expected = _expected_hours(schedule, date_from, date_to)
worked = sum(e.worked_hours or 0.0 for e in entries)
# Sondervertretungs-Zuweisungen laden (nur FZA-relevante)
date_from_all = min(e.date for e in entries)
date_to_all = max(e.date for e in entries)
special_assignments = list(await db.scalars(
select(SpecialAssignment).where(
SpecialAssignment.user_id == user.id,
SpecialAssignment.mode.in_([AssignmentMode.fza, AssignmentMode.both]),
SpecialAssignment.date_from <= date_to_all,
SpecialAssignment.date_to >= date_from_all,
)
))
def _fza_factor(entry_date: date) -> float:
"""Gibt den Faktor für einen Tag zurück (1.0 wenn keine Zuweisung aktiv)."""
for sa in special_assignments:
if sa.date_from <= entry_date <= sa.date_to:
return float(sa.factor)
return 1.0
expected = _expected_hours(schedule, date_from_all, date_to_all)
worked = sum((e.worked_hours or 0.0) * _fza_factor(e.date) for e in entries)
overtime = max(0.0, worked - expected)
bal.total_hours = Decimal(str(round(overtime, 2)))