feat: ICS-Kalender-Feed für Fristen (Issue #21)

Neuer Endpoint GET /api/fristen/kalender.ics:
- Alle offenen + überfälligen Fristen als ICS-Feed
- Ganztägige Events mit VALUE=DATE
- Überfällige Fristen mit ⚠️-Prefix und Tage-Angabe
- 7-Tage-Vorab-Alarm (VALARM TRIGGER:-P7D)
- Stabile UIDs (frist-{id}@antraege.toppyr.de)
- RFC 5545 konform: CRLF, Line Folding, Escaping
- Cache-Header: public, max-age=3600
- VTIMEZONE Europe/Berlin inkludiert
- Kein Auth erforderlich (öffentlicher Feed)
- Keine externen Dependencies

Closes #21
This commit is contained in:
Dotty Dotter 2026-04-02 23:16:53 +02:00
parent b601c0a366
commit a5cf173e5b

View File

@ -1,10 +1,11 @@
from __future__ import annotations from __future__ import annotations
"""API routes for Fristen-Tracking (Issue #17).""" """API routes for Fristen-Tracking (Issue #17)."""
from datetime import date, datetime from datetime import date, datetime, timedelta
from typing import Optional from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import Response
from pydantic import BaseModel from pydantic import BaseModel
from tracker.db.session import get_connection from tracker.db.session import get_connection
@ -171,6 +172,182 @@ def delete_frist(frist_id: int, conn=Depends(_db)):
return {"ok": True, "deleted": frist_id} return {"ok": True, "deleted": frist_id}
@router.get("/kalender.ics")
def fristen_kalender(conn=Depends(_db)):
"""ICS-Kalender-Feed für offene und überfällige Fristen."""
today = date.today()
rows = conn.execute(
"""SELECT f.id, f.typ, f.datum, f.beschreibung, f.status,
k.status AS kette_status,
v_ursprung.aktenzeichen AS kette_aktenzeichen,
v_ursprung.betreff AS kette_betreff
FROM fristen f
LEFT JOIN ketten k ON f.kette_id = k.id
LEFT JOIN vorlagen v_ursprung ON k.ursprung_id = v_ursprung.id
WHERE f.status IN ('offen', 'überfällig')
OR (f.status = 'offen' AND f.datum < date('now'))
ORDER BY f.datum ASC""",
).fetchall()
events = []
for r in rows:
row = dict(r)
frist_datum = date.fromisoformat(row["datum"]) if row["datum"] else None
if not frist_datum:
continue
frist_id = row["id"]
typ = row["typ"] or "Frist"
beschreibung = row["beschreibung"] or ""
aktenzeichen = row["kette_aktenzeichen"] or ""
kette_betreff = row["kette_betreff"] or ""
status = row["status"]
# Detect overdue
is_overdue = status == "überfällig" or (status == "offen" and frist_datum < today)
if is_overdue:
status = "überfällig"
# Build SUMMARY
summary_prefix = "\u26a0\ufe0f " if is_overdue else ""
summary = f"{summary_prefix}{typ}: {beschreibung} ({aktenzeichen})"
# Build DESCRIPTION
desc_parts = [beschreibung]
if is_overdue:
days_overdue = (today - frist_datum).days
desc_parts.append(f"\u00dcBERF\u00c4LLIG seit {days_overdue} Tagen")
desc_parts.append(f"Kette: {kette_betreff}")
desc_parts.append(f"Status: {status}")
desc_parts.append("")
desc_parts.append("https://antraege.toppyr.de/explorer")
description = "\\n".join(desc_parts)
# CATEGORIES + STATUS mapping
if is_overdue:
categories = "überfällig"
ics_status = "CONFIRMED"
else:
categories = "offen"
ics_status = "TENTATIVE"
dtstart = frist_datum.strftime("%Y%m%d")
dtend = (frist_datum + timedelta(days=1)).strftime("%Y%m%d")
dtstamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
event_lines = [
"BEGIN:VEVENT",
f"UID:frist-{frist_id}@antraege.toppyr.de",
f"DTSTAMP:{dtstamp}",
f"DTSTART;VALUE=DATE:{dtstart}",
f"DTEND;VALUE=DATE:{dtend}",
f"SUMMARY:{_ics_escape(summary)}",
f"DESCRIPTION:{_ics_escape(description)}",
"URL:https://antraege.toppyr.de/explorer",
f"CATEGORIES:{_ics_escape(categories)}",
f"STATUS:{ics_status}",
"BEGIN:VALARM",
"TRIGGER:-P7D",
"ACTION:DISPLAY",
f"DESCRIPTION:Frist in 7 Tagen: {_ics_escape(beschreibung)}",
"END:VALARM",
"END:VEVENT",
]
events.append("\r\n".join(event_lines))
cal_lines = [
"BEGIN:VCALENDAR",
"VERSION:2.0",
"PRODID:-//Antragstracker Hagen//Fristen//DE",
"CALSCALE:GREGORIAN",
"METHOD:PUBLISH",
"X-WR-CALNAME:Antragstracker Fristen",
"X-WR-TIMEZONE:Europe/Berlin",
"BEGIN:VTIMEZONE",
"TZID:Europe/Berlin",
"BEGIN:DAYLIGHT",
"TZNAME:CEST",
"TZOFFSETFROM:+0100",
"TZOFFSETTO:+0200",
"DTSTART:19700329T020000",
"RRULE:FREQ=YEARLY;BYMONTH=3;BYDAY=-1SU",
"END:DAYLIGHT",
"BEGIN:STANDARD",
"TZNAME:CET",
"TZOFFSETFROM:+0200",
"TZOFFSETTO:+0100",
"DTSTART:19701025T030000",
"RRULE:FREQ=YEARLY;BYMONTH=10;BYDAY=-1SU",
"END:STANDARD",
"END:VTIMEZONE",
]
for event in events:
cal_lines.append(event)
cal_lines.append("END:VCALENDAR")
ics_body = "\r\n".join(cal_lines) + "\r\n"
# Apply line folding (max 75 octets per line)
ics_body = _ics_fold(ics_body)
return Response(
content=ics_body,
media_type="text/calendar; charset=utf-8",
headers={"Cache-Control": "public, max-age=3600"},
)
def _ics_escape(text: str) -> str:
"""Escape text for ICS content lines (RFC 5545 §3.3.11)."""
if not text:
return ""
text = text.replace("\\", "\\\\")
text = text.replace(";", "\\;")
text = text.replace(",", "\\,")
# Preserve intentional \\n sequences (literal newlines in ICS)
# but escape any actual newlines
text = text.replace("\n", "\\n")
text = text.replace("\r", "")
return text
def _ics_fold(text: str) -> str:
"""Fold lines longer than 75 octets per RFC 5545 §3.1."""
result = []
for line in text.split("\r\n"):
encoded = line.encode("utf-8")
if len(encoded) <= 75:
result.append(line)
continue
# Fold: first line 75 octets, continuations 74 (space prefix)
chunks = []
while len(encoded) > 75:
# Cut at 75 octets for first chunk, 74 for subsequent
cut = 75 if not chunks else 74
# Don't break in middle of multi-byte UTF-8
split_at = cut
while split_at > 0 and (encoded[split_at] & 0xC0) == 0x80:
split_at -= 1
chunk = encoded[:split_at].decode("utf-8")
encoded = encoded[split_at:]
if chunks:
chunks.append(" " + chunk)
else:
chunks.append(chunk)
# Remaining
remaining = encoded.decode("utf-8")
if remaining:
if chunks:
chunks.append(" " + remaining)
else:
chunks.append(remaining)
result.append("\r\n".join(chunks))
return "\r\n".join(result)
@router.get("/ueberfaellig") @router.get("/ueberfaellig")
def get_ueberfaellige(conn=Depends(_db)): def get_ueberfaellige(conn=Depends(_db)):
"""Überfällige Fristen abrufen und automatisch Status setzen.""" """Überfällige Fristen abrufen und automatisch Status setzen."""