feat(reports): DATEV-Monatsblatt-Export (Blanko-Vorlage) für Steuerberater

Neuer PDF-Export GET /reports/datev-monthly/export: ein Blatt pro Mitarbeiter/Monat im Layout der DATEV-Stundenaufzeichnungs-Vorlage (Beginn/Pause/Ende/Dauer + K/U/UU/F/SA/SU-Kürzel + Bemerkungen + Summe + Unterschriftsfelder). Führt time_entries, Absences und Feiertage pro Kalendertag zusammen. Außerdem: Mustervorlagen für neue Backend-Module (Model/Schema/Router/ Migration/Test) und eine Frontend-Page-Vorlage, abgeleitet vom hours_payouts-Modul als aktuellstem sauberen Muster. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CviFgc739S89xS97pvjszj
2026-08-06 22:19:54 +02:00
parent 90f45a4f85
commit 5cbeebe66d
10 changed files with 901 additions and 2 deletions
@@ -1,7 +1,7 @@
 from datetime import date, timedelta
 from uuid import UUID

-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi.responses import Response
 from sqlalchemy.ext.asyncio import AsyncSession

@@ -152,6 +152,28 @@ async def export_time_report(
                    headers={"Content-Disposition": f"attachment; filename={filename}.csv"})


+@router.get("/reports/datev-monthly/export")
+async def export_datev_monthly(
+    current_user: CurrentUser,
+    user_id: UUID,
+    year: int = Query(..., ge=2000, le=2100),
+    month: int = Query(..., ge=1, le=12),
+    db: AsyncSession = Depends(get_db),
+):
+    """DATEV-Monatsblatt (Blanko-Layout) als PDF für Steuerberater-Übergabe.
+    EMPLOYEE darf nur den eigenen Monat exportieren."""
+    if current_user.role not in _manager_roles and user_id != current_user.id:
+        raise HTTPException(403, "Nur eigenes Monatsblatt exportierbar")
+
+    sheet = await report_service.datev_monthly_report(
+        current_user.company_id, user_id, year, month, db
+    )
+    content = report_service.datev_monthly_report_to_pdf(sheet)
+    filename = f"datev_stundenaufzeichnung_{sheet.user_name.replace(' ', '_')}_{year}_{month:02d}.pdf"
+    return Response(content=content, media_type="application/pdf",
+                    headers={"Content-Disposition": f"attachment; filename={filename}"})
+
+
@router.get("/reports/absences/export")
 async def export_absence_report(
    current_user: CurrentUser,
@@ -188,3 +188,29 @@ class OvertimeReportDetailed(BaseModel):
    total_employees: int
    total_overtime: float
    rows: list[OvertimeReportRowDetailed]
+
+
+# ── DATEV-Monatsblatt ──────────────────────────────────────────────────────────
+# Layout entspricht "Stundenaufzeichnungen Muster blanko DATEV ab 2015":
+# ein Blatt pro Mitarbeiter/Monat, Zeile pro Kalendertag.
+
+class DatevDayRow(BaseModel):
+    day: int                              # Kalendertag 1-31
+    weekday_label: str                    # "Mo".."So" für Anzeige
+    start_time: time | None = None
+    break_minutes: int | None = None
+    end_time: time | None = None
+    duration_hours: float | None = None
+    code: str | None = None               # K/U/UU/F/SA/SU – siehe DATEV_CODES
+    recorded_on: date | None = None       # "aufgezeichnet am" (created_at des Eintrags)
+    note: str | None = None               # Bemerkungen (Absence-Typ-Name o.ä.)
+
+
+class DatevMonthlySheet(BaseModel):
+    company_name: str
+    user_name: str
+    personnel_number: str | None
+    year: int
+    month: int
+    rows: list[DatevDayRow]
+    total_hours: float
@@ -1,3 +1,4 @@
+import calendar
 import csv
 import io
 from collections import defaultdict
@@ -9,7 +10,7 @@ from sqlalchemy import distinct, func, select
 from sqlalchemy.ext.asyncio import AsyncSession

 from app.models.absence import Absence, AbsenceStatus
-from app.models.absence_type import AbsenceType
+from app.models.absence_type import AbsenceCategory, AbsenceType
 from app.models.company import Company
 from app.models.department import Department
 from app.models.overtime_balance import OvertimeBalance
@@ -21,6 +22,8 @@ from app.schemas.report import (
    AbsenceReport,
    AbsenceReportRow,
    CompanyDashboard,
+    DatevDayRow,
+    DatevMonthlySheet,
    DayEntry,
    EmployeeDashboard,
    HoursBreakdown,
@@ -675,6 +678,195 @@ class ReportService:
            total_rows=len(rows), total_hours=round(total_hours, 2), rows=rows,
        )

+    # ── DATEV-Monatsblatt ────────────────────────────────────────────────────
+    # Kürzel-Mapping laut Vorlage "Stundenaufzeichnungen Muster blanko DATEV ab 2015":
+    # K=Krank, U=Urlaub, UU=unbezahlter Urlaub, F=Feiertag, SA=Stundenweise abwesend,
+    # SU=Stundenweise Urlaub.
+
+    _WEEKDAY_LABELS = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
+
+    @staticmethod
+    def _datev_code_for_absence(absence_type: AbsenceType, is_half_day: bool) -> str:
+        if absence_type.category == AbsenceCategory.SICK:
+            return "K"
+        if absence_type.category == AbsenceCategory.VACATION:
+            if is_half_day:
+                return "SU"
+            return "U" if absence_type.is_paid else "UU"
+        # Alle übrigen Kategorien (overtime_comp, training, business_trip, other)
+        # haben kein eigenes DATEV-Kürzel -> nur Bemerkungen-Spalte
+        return "SA" if is_half_day else ""
+
+    async def datev_monthly_report(
+        self,
+        company_id: UUID,
+        user_id: UUID,
+        year: int,
+        month: int,
+        db: AsyncSession,
+    ) -> DatevMonthlySheet:
+        """Ein-Seite-pro-Mitarbeiter-Monatsblatt im DATEV-Blanko-Layout."""
+        user = await db.get(User, user_id)
+        if user is None or user.company_id != company_id:
+            raise ValueError("Mitarbeiter nicht gefunden")
+        company = await db.get(Company, company_id)
+
+        days_in_month = calendar.monthrange(year, month)[1]
+        date_from = date(year, month, 1)
+        date_to = date(year, month, days_in_month)
+
+        # Zeiteinträge des Monats
+        entries_stmt = select(TimeEntry).where(
+            TimeEntry.user_id == user_id,
+            TimeEntry.date >= date_from,
+            TimeEntry.date <= date_to,
+        )
+        entries_by_day: dict[int, TimeEntry] = {
+            e.date.day: e for e in (await db.scalars(entries_stmt)).all()
+        }
+
+        # Genehmigte Abwesenheiten, die den Monat überlappen
+        absences_stmt = (
+            select(Absence, AbsenceType)
+            .join(AbsenceType, Absence.absence_type_id == AbsenceType.id)
+            .where(
+                Absence.user_id == user_id,
+                Absence.status.in_([AbsenceStatus.APPROVED, AbsenceStatus.FIRST_APPROVED]),
+                Absence.start_date <= date_to,
+                Absence.end_date >= date_from,
+            )
+        )
+        absences = list((await db.execute(absences_stmt)).all())
+
+        # Feiertage (nur wenn Bundesland konfiguriert)
+        holidays: dict[date, tuple[str, bool]] = {}
+        if company and company.state:
+            holidays = await get_holidays_set(date_from, date_to, company.state, db)
+
+        rows: list[DatevDayRow] = []
+        total_hours = 0.0
+
+        for day in range(1, days_in_month + 1):
+            d = date(year, month, day)
+            weekday_label = self._WEEKDAY_LABELS[d.weekday()]
+
+            row = DatevDayRow(day=day, weekday_label=weekday_label)
+
+            entry = entries_by_day.get(day)
+            if entry is not None:
+                row.start_time = entry.start_time
+                row.end_time = entry.end_time
+                row.break_minutes = entry.break_minutes
+                row.duration_hours = entry.worked_hours
+                row.recorded_on = entry.created_at.date() if entry.created_at else None
+                if entry.worked_hours:
+                    total_hours += entry.worked_hours
+
+            # Feiertag hat Vorrang vor Abwesenheit (analog _categorize_hours-Logik)
+            if d in holidays:
+                row.code = "F"
+                row.note = holidays[d][0]
+            else:
+                for absence, absence_type in absences:
+                    if absence.start_date <= d <= absence.end_date:
+                        is_half_day = (
+                            (d == absence.start_date and absence.half_day_start)
+                            or (d == absence.end_date and absence.half_day_end)
+                        )
+                        row.code = self._datev_code_for_absence(absence_type, is_half_day)
+                        row.note = absence_type.name
+                        break
+
+            rows.append(row)
+
+        return DatevMonthlySheet(
+            company_name=company.name if company else "",
+            user_name=f"{user.first_name} {user.last_name}",
+            personnel_number=user.personnel_number,
+            year=year, month=month,
+            rows=rows, total_hours=round(total_hours, 2),
+        )
+
+    def datev_monthly_report_to_pdf(self, sheet: DatevMonthlySheet) -> bytes:
+        def fmt_t(t: time | None) -> str:
+            return t.strftime("%H:%M") if t else ""
+
+        def fmt_h(h: float | None) -> str:
+            return f"{h:.2f}".replace(".", ",") if h is not None else ""
+
+        rows_html = ""
+        for r in sheet.rows:
+            break_str = f"{r.break_minutes} min" if r.break_minutes else ""
+            recorded = r.recorded_on.strftime("%d.%m.%Y") if r.recorded_on else ""
+            rows_html += f"""<tr>
+  <td>{r.day}. ({r.weekday_label})</td>
+  <td>{fmt_t(r.start_time)}</td>
+  <td>{break_str}</td>
+  <td>{fmt_t(r.end_time)}</td>
+  <td class="right">{fmt_h(r.duration_hours)}</td>
+  <td class="bold" style="text-align:center">{r.code or ""}</td>
+  <td>{recorded}</td>
+  <td>{r.note or ""}</td>
+</tr>"""
+
+        month_names = [
+            "Januar", "Februar", "März", "April", "Mai", "Juni",
+            "Juli", "August", "September", "Oktober", "November", "Dezember",
+        ]
+        period = f"{month_names[sheet.month - 1]} {sheet.year}"
+
+        html = f"""<!DOCTYPE html>
+<html lang="de">
+<head>
+<meta charset="utf-8">
+<style>
+  @page {{ margin: 1.2cm 1.5cm; }}
+  body {{ font-family: Arial, Helvetica, sans-serif; font-size: 9px; color: #111827; }}
+  h1 {{ font-size: 14px; margin: 0 0 10px; }}
+  .meta {{ display: flex; gap: 30px; margin-bottom: 12px; font-size: 10px; }}
+  .meta div span {{ display: block; color: #6b7280; font-size: 8px; text-transform: uppercase; }}
+  table {{ width: 100%; border-collapse: collapse; }}
+  thead th {{ background: #1e40af; color: white; padding: 4px 6px; text-align: left; font-size: 8px; text-transform: uppercase; }}
+  td {{ padding: 3px 6px; border-bottom: 1px solid #e5e7eb; }}
+  td.right {{ text-align: right; }}
+  td.bold {{ font-weight: 700; }}
+  tfoot td {{ background: #1e293b; color: white; font-weight: bold; padding: 5px 6px; }}
+  .signatures {{ display: flex; justify-content: space-between; margin-top: 40px; }}
+  .sig-line {{ width: 45%; border-top: 1px solid #111827; padding-top: 4px; font-size: 9px; text-align: center; }}
+  .codes {{ margin-top: 16px; font-size: 8px; color: #4b5563; }}
+</style>
+</head>
+<body>
+<h1>Stundenaufzeichnung – {period}</h1>
+<div class="meta">
+  <div><span>Firma</span>{sheet.company_name}</div>
+  <div><span>Name des Mitarbeiters</span>{sheet.user_name}</div>
+  <div><span>Pers.-Nr.</span>{sheet.personnel_number or "—"}</div>
+</div>
+<table>
+<thead><tr>
+  <th>Kalendertag</th><th>Beginn</th><th>Pause</th><th>Ende</th>
+  <th>Dauer</th><th>*</th><th>aufgezeichnet am</th><th>Bemerkungen</th>
+</tr></thead>
+<tbody>
+{rows_html}
+</tbody>
+<tfoot><tr>
+  <td colspan="4">Summe</td>
+  <td class="right">{fmt_h(sheet.total_hours)}</td>
+  <td colspan="3"></td>
+</tr></tfoot>
+</table>
+<div class="codes">* K=Krank · U=Urlaub · UU=unbezahlter Urlaub · F=Feiertag · SA=Stundenweise abwesend · SU=Stundenweise Urlaub</div>
+<div class="signatures">
+  <div class="sig-line">Datum, Unterschrift Arbeitnehmer</div>
+  <div class="sig-line">Datum, Unterschrift Arbeitgeber</div>
+</div>
+</body>
+</html>"""
+        from weasyprint import HTML
+        return HTML(string=html).write_pdf()
+
    # ── Absence Report ───────────────────────────────────────────────────────

    async def absence_report(