diff --git a/Claude outputs/main.py b/Claude outputs/main.py index bd3ebf8..56aa3e3 100644 --- a/Claude outputs/main.py +++ b/Claude outputs/main.py @@ -47,6 +47,7 @@ CLAUDE_API_KEY = os.getenv("CLAUDE_API_KEY") CLAUDE_MODEL = os.getenv("CLAUDE_MODEL", "claude-sonnet-5") API_KEY_ADMIN = os.getenv("API_KEY_ADMIN") NEXTCLOUD_CALDAV_URL = os.getenv("NEXTCLOUD_CALDAV_URL") +NEXTCLOUD_CALDAV_URL_PERSONAL = os.getenv("NEXTCLOUD_CALDAV_URL_PERSONAL") NEXTCLOUD_USER = os.getenv("NEXTCLOUD_USER") NEXTCLOUD_APP_PASSWORD = os.getenv("NEXTCLOUD_APP_PASSWORD") EMAIL_IMAP_HOST = os.getenv("EMAIL_IMAP_HOST") @@ -351,39 +352,50 @@ async def build_memory_context(user_message: str, conversation_id: int) -> str: return "\n\n".join(sections) -def _caldav_calendar(): - dav_root = NEXTCLOUD_CALDAV_URL.split("/calendars/")[0] + "/" +CALENDAR_URLS = { + "personal": NEXTCLOUD_CALDAV_URL_PERSONAL, + "feuerwehr": NEXTCLOUD_CALDAV_URL, +} +DEFAULT_CALENDAR = "personal" + + +def _caldav_calendar(calendar: str = DEFAULT_CALENDAR): + url = CALENDAR_URLS[calendar] + dav_root = url.split("/calendars/")[0] + "/" client = caldav.DAVClient(url=dav_root, username=NEXTCLOUD_USER, password=NEXTCLOUD_APP_PASSWORD) - return client.calendar(url=NEXTCLOUD_CALDAV_URL) + return client.calendar(url=url) -def _list_upcoming_events_sync(days_ahead: int) -> list: - calendar = _caldav_calendar() - start = datetime.now() - end = start + timedelta(days=days_ahead) - results = calendar.search(start=start, end=end, event=True, expand=True) +def _list_upcoming_events_sync(days_ahead: int, calendar: str = None) -> list: + calendar_names = [calendar] if calendar else list(CALENDAR_URLS.keys()) events = [] - for result in results: - comp = result.icalendar_component - dtstart = comp.get("dtstart").dt - dtend_prop = comp.get("dtend") - dtend = dtend_prop.dt if dtend_prop else dtstart - events.append( - { - "uid": str(comp.get("uid", "")), - "summary": str(comp.get("summary", "")), - "start": dtstart.isoformat(), - "end": dtend.isoformat(), - "description": str(comp.get("description", "")), - "all_day": not isinstance(dtstart, datetime), - } - ) + for name in calendar_names: + cal = _caldav_calendar(name) + start = datetime.now() + end = start + timedelta(days=days_ahead) + results = cal.search(start=start, end=end, event=True, expand=True) + for result in results: + comp = result.icalendar_component + dtstart = comp.get("dtstart").dt + dtend_prop = comp.get("dtend") + dtend = dtend_prop.dt if dtend_prop else dtstart + events.append( + { + "uid": str(comp.get("uid", "")), + "summary": str(comp.get("summary", "")), + "start": dtstart.isoformat(), + "end": dtend.isoformat(), + "description": str(comp.get("description", "")), + "all_day": not isinstance(dtstart, datetime), + "calendar": name, + } + ) events.sort(key=lambda e: e["start"]) return events -async def list_upcoming_events(days_ahead: int) -> list: - return await asyncio.to_thread(_list_upcoming_events_sync, days_ahead) +async def list_upcoming_events(days_ahead: int, calendar: str = None) -> list: + return await asyncio.to_thread(_list_upcoming_events_sync, days_ahead, calendar) CALENDAR_TIMEZONE = ZoneInfo("Europe/Berlin") @@ -397,19 +409,19 @@ def _as_calendar_local(dt: datetime) -> datetime: return dt -def _create_event_sync(summary: str, start: str, end: str, description: str = "") -> dict: - calendar = _caldav_calendar() - calendar.save_event( +def _create_event_sync(summary: str, start: str, end: str, description: str = "", calendar: str = DEFAULT_CALENDAR) -> dict: + cal = _caldav_calendar(calendar) + cal.save_event( dtstart=_as_calendar_local(datetime.fromisoformat(start)), dtend=_as_calendar_local(datetime.fromisoformat(end)), summary=summary, description=description, ) - return {"summary": summary, "start": start, "end": end, "description": description} + return {"summary": summary, "start": start, "end": end, "description": description, "calendar": calendar} -async def create_event(summary: str, start: str, end: str, description: str = "") -> dict: - return await asyncio.to_thread(_create_event_sync, summary, start, end, description) +async def create_event(summary: str, start: str, end: str, description: str = "", calendar: str = DEFAULT_CALENDAR) -> dict: + return await asyncio.to_thread(_create_event_sync, summary, start, end, description, calendar) def _update_event_sync( @@ -418,9 +430,10 @@ def _update_event_sync( end: str = None, summary: str = None, description: str = None, + calendar: str = DEFAULT_CALENDAR, ) -> dict: - calendar = _caldav_calendar() - event = calendar.event_by_uid(uid) + cal = _caldav_calendar(calendar) + event = cal.event_by_uid(uid) comp = event.icalendar_component if start is not None: comp["dtstart"].dt = _as_calendar_local(datetime.fromisoformat(start)) @@ -445,19 +458,20 @@ async def update_event( end: str = None, summary: str = None, description: str = None, + calendar: str = DEFAULT_CALENDAR, ) -> dict: - return await asyncio.to_thread(_update_event_sync, uid, start, end, summary, description) + return await asyncio.to_thread(_update_event_sync, uid, start, end, summary, description, calendar) -def _delete_event_sync(uid: str) -> dict: - calendar = _caldav_calendar() - event = calendar.event_by_uid(uid) +def _delete_event_sync(uid: str, calendar: str = DEFAULT_CALENDAR) -> dict: + cal = _caldav_calendar(calendar) + event = cal.event_by_uid(uid) event.delete() return {"deleted": True, "uid": uid} -async def delete_event(uid: str) -> dict: - return await asyncio.to_thread(_delete_event_sync, uid) +async def delete_event(uid: str, calendar: str = DEFAULT_CALENDAR) -> dict: + return await asyncio.to_thread(_delete_event_sync, uid, calendar) def _deck_base_url() -> str: @@ -689,21 +703,26 @@ async def forget_fact(query: str) -> dict: CALENDAR_TOOLS = [ { "name": "list_calendar_events", - "description": "List upcoming events from the FFW-Onza-Alle calendar within the next N days.", + "description": "List upcoming events within the next N days, from one calendar or both merged.", "input_schema": { "type": "object", "properties": { "days_ahead": { "type": "integer", "description": "How many days ahead to look, e.g. 7 for the next week", - } + }, + "calendar": { + "type": "string", + "description": "Filter by calendar: 'personal' or 'feuerwehr'. Omit to list events from both calendars merged.", + "enum": ["personal", "feuerwehr"], + }, }, "required": ["days_ahead"], }, }, { "name": "create_calendar_event", - "description": "Create a new event in the FFW-Onza-Alle calendar.", + "description": "Create a new event in a calendar.", "input_schema": { "type": "object", "properties": { @@ -711,6 +730,11 @@ CALENDAR_TOOLS = [ "start": {"type": "string", "description": "Start date/time in ISO 8601, e.g. 2026-09-20T10:00:00"}, "end": {"type": "string", "description": "End date/time in ISO 8601, e.g. 2026-09-20T11:00:00"}, "description": {"type": "string", "description": "Optional longer description"}, + "calendar": { + "type": "string", + "description": "Which calendar to create the event in: 'personal' (default, for everything unless the Feuerwehr is named) or 'feuerwehr' (Feuerwehr Onza events). Omit for 'personal'.", + "enum": ["personal", "feuerwehr"], + }, }, "required": ["summary", "start", "end"], }, @@ -718,9 +742,9 @@ CALENDAR_TOOLS = [ { "name": "update_calendar_event", "description": ( - "Reschedule or rename an existing event in the FFW-Onza-Alle " - "calendar. Call list_calendar_events first if you don't already " - "know the event's uid. Only pass the fields that should change." + "Reschedule or rename an existing event. Call list_calendar_events " + "first if you don't already know the event's uid. Only pass the " + "fields that should change." ), "input_schema": { "type": "object", @@ -730,6 +754,11 @@ CALENDAR_TOOLS = [ "end": {"type": "string", "description": "New end date/time in ISO 8601"}, "summary": {"type": "string", "description": "New event title"}, "description": {"type": "string", "description": "New description"}, + "calendar": { + "type": "string", + "description": "Which calendar the event is in - use the 'calendar' value list_calendar_events returned for it. Omit for 'personal'.", + "enum": ["personal", "feuerwehr"], + }, }, "required": ["uid"], }, @@ -737,14 +766,18 @@ CALENDAR_TOOLS = [ { "name": "delete_calendar_event", "description": ( - "Permanently delete an event from the FFW-Onza-Alle calendar. " - "Only call this after the user has explicitly confirmed the " - "deletion in the conversation." + "Permanently delete an event. Only call this after the user has " + "explicitly confirmed the deletion in the conversation." ), "input_schema": { "type": "object", "properties": { "uid": {"type": "string", "description": "The event's uid, from list_calendar_events"}, + "calendar": { + "type": "string", + "description": "Which calendar the event is in - use the 'calendar' value list_calendar_events returned for it. Omit for 'personal'.", + "enum": ["personal", "feuerwehr"], + }, }, "required": ["uid"], }, @@ -869,16 +902,25 @@ ORDER_TOOLS = [ ] CALENDAR_ASSISTANT_INSTRUCTIONS = ( - "Du hast Zugriff auf den Kalender 'FFW-Onza-Alle' ueber die Tools " - "list_calendar_events, create_calendar_event, update_calendar_event und " - "delete_calendar_event. create_calendar_event und update_calendar_event " - "kannst du direkt aufrufen, sobald du die noetigen Angaben hast - keine " - "Rueckfrage noetig. Bevor du delete_calendar_event aufrufst, frage den " - "Nutzer aber immer explizit im Klartext nach Bestaetigung und rufe das " - "Tool erst auf, nachdem der Nutzer im naechsten Chat-Beitrag zugestimmt " - "hat. Fuer update_calendar_event und delete_calendar_event brauchst du " - "die uid des Termins - ruf dafuer zuerst list_calendar_events auf, falls " - "du sie noch nicht aus dem Gespraech kennst.\n\n" + "Du hast Zugriff auf zwei Kalender ueber die Tools list_calendar_events, " + "create_calendar_event, update_calendar_event und delete_calendar_event: " + "'personal' (Standard-Kalender fuer alle Termine) und 'feuerwehr' (fuer " + "Termine der Feuerwehr Onza, z.B. Einsaetze, Uebungen, Wachdienste). " + "Nennt der Nutzer keinen Kalender oder keinen erkennbaren " + "Feuerwehr-Bezug, verwende 'personal'. Erkennst du am Termin einen " + "klaren Feuerwehr-Bezug (Feuerwehr/FFW/Onza/Einsatz/Uebung/Wache), " + "verwende 'feuerwehr'. Bei Unklarheit frage lieber kurz nach, statt zu " + "raten. create_calendar_event und update_calendar_event kannst du " + "direkt aufrufen, sobald du die noetigen Angaben hast - keine " + "Rueckfrage zur Bestaetigung noetig (nur ggf. zum Kalender, siehe " + "oben). Bevor du delete_calendar_event aufrufst, frage den Nutzer aber " + "immer explizit im Klartext nach Bestaetigung und rufe das Tool erst " + "auf, nachdem der Nutzer im naechsten Chat-Beitrag zugestimmt hat. " + "Fuer update_calendar_event und delete_calendar_event brauchst du die " + "uid des Termins sowie den passenden Kalender - ruf dafuer zuerst " + "list_calendar_events auf, falls du beides noch nicht aus dem " + "Gespraech kennst, und uebernimm das dort zurueckgegebene 'calendar'-" + "Feld unveraendert.\n\n" "Du hast ausserdem Zugriff auf das Postfach kontakt@mbo-tech-it.de ueber " "die Tools list_recent_emails und send_email. list_recent_emails kannst " "du direkt aufrufen. Bevor du send_email aufrufst, zeige dem Nutzer " @@ -917,7 +959,7 @@ ORDER_ASSISTANT_INSTRUCTIONS = ( async def execute_tool(name: str, tool_input: dict) -> str: if name == "list_calendar_events": - events = await list_upcoming_events(tool_input.get("days_ahead", 14)) + events = await list_upcoming_events(tool_input.get("days_ahead", 14), tool_input.get("calendar")) return json.dumps(events) if name == "create_calendar_event": result = await create_event( @@ -925,6 +967,7 @@ async def execute_tool(name: str, tool_input: dict) -> str: tool_input["start"], tool_input["end"], tool_input.get("description", ""), + tool_input.get("calendar") or DEFAULT_CALENDAR, ) return json.dumps(result) if name == "update_calendar_event": @@ -934,10 +977,11 @@ async def execute_tool(name: str, tool_input: dict) -> str: tool_input.get("end"), tool_input.get("summary"), tool_input.get("description"), + tool_input.get("calendar") or DEFAULT_CALENDAR, ) return json.dumps(result) if name == "delete_calendar_event": - result = await delete_event(tool_input["uid"]) + result = await delete_event(tool_input["uid"], tool_input.get("calendar") or DEFAULT_CALENDAR) return json.dumps(result) if name == "list_recent_emails": emails = await list_recent_emails(tool_input.get("limit", 10)) diff --git a/Claude outputs/tests/test_calendar.py b/Claude outputs/tests/test_calendar.py new file mode 100644 index 0000000..e117b3b --- /dev/null +++ b/Claude outputs/tests/test_calendar.py @@ -0,0 +1,175 @@ +import os +import sys +from datetime import datetime, date +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +import main + + +class _Val: + def __init__(self, v): + self.dt = v + + +def _fake_component(summary, dtstart_value, dtend_value, description="", uid="uid-1"): + comp = {} + comp["summary"] = summary + comp["description"] = description + comp["uid"] = uid + comp["dtstart"] = _Val(dtstart_value) + comp["dtend"] = _Val(dtend_value) + + def get(key, default=None): + return comp.get(key, default) + + fake = MagicMock() + fake.get = get + return fake + + +class _MutableComponent(dict): + def get(self, key, default=None): + return super().get(key, default) + + +def test_list_upcoming_events_maps_timed_and_all_day_events(): + timed = MagicMock() + timed.icalendar_component = _fake_component( + "THL FM LS", datetime(2026, 9, 14, 19, 0, 0), datetime(2026, 9, 14, 20, 0, 0) + ) + all_day = MagicMock() + all_day.icalendar_component = _fake_component( + "Volksfestwache", date(2026, 9, 19), date(2026, 9, 20) + ) + + fake_calendar = MagicMock() + fake_calendar.search.return_value = [all_day, timed] # unsorted on purpose + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar): + events = main._list_upcoming_events_sync(14, calendar="feuerwehr") + + assert [e["summary"] for e in events] == ["THL FM LS", "Volksfestwache"] + assert events[0]["all_day"] is False + assert events[1]["all_day"] is True + assert events[0]["uid"] == "uid-1" + assert events[0]["calendar"] == "feuerwehr" + + +def test_list_upcoming_events_merges_both_calendars_when_none_given(): + personal_event = MagicMock() + personal_event.icalendar_component = _fake_component( + "Zahnarzt", datetime(2026, 9, 14, 9, 0, 0), datetime(2026, 9, 14, 9, 30, 0), uid="uid-p" + ) + ffw_event = MagicMock() + ffw_event.icalendar_component = _fake_component( + "Einsatz", datetime(2026, 9, 14, 10, 0, 0), datetime(2026, 9, 14, 11, 0, 0), uid="uid-f" + ) + + def fake_caldav_calendar(calendar=main.DEFAULT_CALENDAR): + cal = MagicMock() + cal.search.return_value = [personal_event] if calendar == "personal" else [ffw_event] + return cal + + with patch.object(main, "_caldav_calendar", side_effect=fake_caldav_calendar): + events = main._list_upcoming_events_sync(14) + + assert [e["uid"] for e in events] == ["uid-p", "uid-f"] + assert [e["calendar"] for e in events] == ["personal", "feuerwehr"] + + +def test_create_event_calls_save_event_with_parsed_dates(): + fake_calendar = MagicMock() + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar): + result = main._create_event_sync( + "JARVIS Testtermin", "2026-09-20T10:00:00", "2026-09-20T11:00:00", "desc" + ) + + fake_calendar.save_event.assert_called_once() + _, kwargs = fake_calendar.save_event.call_args + assert kwargs["summary"] == "JARVIS Testtermin" + assert kwargs["dtstart"] == datetime(2026, 9, 20, 10, 0, 0, tzinfo=main.CALENDAR_TIMEZONE) + assert kwargs["dtstart"].tzinfo is not None + assert result["summary"] == "JARVIS Testtermin" + assert result["calendar"] == "personal" + + +def test_create_event_passes_calendar_argument_through(): + fake_calendar = MagicMock() + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar) as mock_get_calendar: + result = main._create_event_sync( + "Einsatz", "2026-09-20T10:00:00", "2026-09-20T11:00:00", calendar="feuerwehr" + ) + + mock_get_calendar.assert_called_once_with("feuerwehr") + assert result["calendar"] == "feuerwehr" + + +def test_update_event_moves_and_renames_by_uid(): + comp = _MutableComponent() + comp["dtstart"] = _Val(datetime(2026, 9, 27, 10, 0, 0)) + comp["dtend"] = _Val(datetime(2026, 9, 27, 11, 0, 0)) + comp["summary"] = "Old" + comp["description"] = "old desc" + + fake_event = MagicMock() + fake_event.icalendar_component = comp + + fake_calendar = MagicMock() + fake_calendar.event_by_uid.return_value = fake_event + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar): + result = main._update_event_sync("uid-123", start="2026-09-28T14:00:00", summary="New") + + fake_calendar.event_by_uid.assert_called_once_with("uid-123") + fake_event.save.assert_called_once() + assert comp["dtstart"].dt == datetime(2026, 9, 28, 14, 0, 0, tzinfo=main.CALENDAR_TIMEZONE) + assert comp["dtend"].dt == datetime(2026, 9, 27, 11, 0, 0) # untouched - not passed + assert comp["summary"] == "New" + assert comp["description"] == "old desc" # untouched - not passed + assert result["uid"] == "uid-123" + + +def test_update_event_passes_calendar_argument_through(): + comp = _MutableComponent() + comp["dtstart"] = _Val(datetime(2026, 9, 27, 10, 0, 0)) + comp["dtend"] = _Val(datetime(2026, 9, 27, 11, 0, 0)) + comp["summary"] = "Old" + comp["description"] = "old desc" + + fake_event = MagicMock() + fake_event.icalendar_component = comp + fake_calendar = MagicMock() + fake_calendar.event_by_uid.return_value = fake_event + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar) as mock_get_calendar: + main._update_event_sync("uid-123", summary="New", calendar="feuerwehr") + + mock_get_calendar.assert_called_once_with("feuerwehr") + + +def test_delete_event_by_uid(): + fake_event = MagicMock() + fake_calendar = MagicMock() + fake_calendar.event_by_uid.return_value = fake_event + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar): + result = main._delete_event_sync("uid-456") + + fake_calendar.event_by_uid.assert_called_once_with("uid-456") + fake_event.delete.assert_called_once() + assert result == {"deleted": True, "uid": "uid-456"} + + +def test_delete_event_passes_calendar_argument_through(): + fake_event = MagicMock() + fake_calendar = MagicMock() + fake_calendar.event_by_uid.return_value = fake_event + + with patch.object(main, "_caldav_calendar", return_value=fake_calendar) as mock_get_calendar: + main._delete_event_sync("uid-456", calendar="feuerwehr") + + mock_get_calendar.assert_called_once_with("feuerwehr")