claude
/
jarvis-assist
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: support two calendars (personal default, feuerwehr named) 

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V57jSQPqwkGG8BuAXg59X5
This commit is contained in:
Jonny 2026-09-13 20:26:57 +02:00
parent 529b9c3320
commit 246fe60ce7
2 changed files with 279 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
Claude outputs/main.py
View File

@ -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,18 +352,28 @@ 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()
def _list_upcoming_events_sync(days_ahead: int, calendar: str = None) -> list:
calendar_names = [calendar] if calendar else list(CALENDAR_URLS.keys())
events = []
for name in calendar_names:
cal = _caldav_calendar(name)
start = datetime.now()
end = start + timedelta(days=days_ahead)
results = calendar.search(start=start, end=end, event=True, expand=True)
events = []
results = cal.search(start=start, end=end, event=True, expand=True)
for result in results:
comp = result.icalendar_component
dtstart = comp.get("dtstart").dt
@ -376,14 +387,15 @@ def _list_upcoming_events_sync(days_ahead: int) -> list:
"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))

175
Claude outputs/tests/test_calendar.py Normal file
View File

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