"""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 = 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 = 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 = 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 = 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 = 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)