Compare commits

..

No commits in common. "1.4.0" and "main" have entirely different histories.
1.4.0 ... main

11 changed files with 287 additions and 589 deletions

View File

@ -16,20 +16,5 @@
</library>
</libraries>
</data-source>
<data-source source="LOCAL" name="db [2]" uuid="653863ac-b4c8-4525-be5f-881f75f27fff">
<driver-ref>sqlite.xerial</driver-ref>
<synchronize>true</synchronize>
<jdbc-driver>org.sqlite.JDBC</jdbc-driver>
<jdbc-url>jdbc:sqlite:$PROJECT_DIR$/db.sqlite3</jdbc-url>
<working-dir>$ProjectFileDir$</working-dir>
<libraries>
<library>
<url>file://$APPLICATION_CONFIG_DIR$/jdbc-drivers/Xerial SQLiteJDBC/3.45.1/org/xerial/sqlite-jdbc/3.45.1.0/sqlite-jdbc-3.45.1.0.jar</url>
</library>
<library>
<url>file://$APPLICATION_CONFIG_DIR$/jdbc-drivers/Xerial SQLiteJDBC/3.45.1/org/slf4j/slf4j-api/1.7.36/slf4j-api-1.7.36.jar</url>
</library>
</libraries>
</data-source>
</component>
</project>

View File

@ -6,5 +6,5 @@
<component name="MarkdownSettingsMigration">
<option name="stateVersion" value="1" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (RoomBooking)" project-jdk-type="Python SDK" />
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8 (tazPlease)" project-jdk-type="Python SDK" />
</project>

View File

@ -1,6 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

View File

@ -23,5 +23,5 @@ USER app-user
EXPOSE 8000
#CMD ["nanodjango", "serve", "booking.py"]
CMD ["gunicorn", "--timeout", "180", "--bind", "0.0.0.0:8000", "booking:app"]
CMD ["nanodjango", "serve", "booking.py"]
#CMD ["nanodjango", "run", "booking.py"]

View File

@ -35,7 +35,6 @@ with the following JSON payload:
```
Curl example:
```bash
curl -s \
-H "Authorization: Bearer secrettoken" \
@ -47,18 +46,14 @@ Curl example:
The response will contain the event ID:
```json
{
"id": "4zx3QAFzoxV3vaZSKGaH2S"
}
{"id": "4zx3QAFzoxV3vaZSKGaH2S"}
```
### Delete an event
To delete an event, you need to send a DELETE request to
`/api/{calendar}/events/{event_id}`.
To delete an event, you need to send a DELETE request to `/api/{calendar}/events/{event_id}`.
Curl example:
```bash
curl -s -X DELETE \
-H "Authorization: Bearer secrettoken" \
@ -81,7 +76,6 @@ The following parameters can be passed as query parameters:
return what would be deleted.
Curl example:
```bash
curl -s -X DELETE \
-H "Authorization: Bearer secrettoken" \
@ -165,30 +159,22 @@ Email templates can be configured in the Django Admin interface. Jinja2 syntax
is used for the templates. The following variables are available in the
templates:
- `booking` - The booking that was declined
- `calendar_name` - The name of the calendar
**Lists**
- `overlap_bookings` - A list of overlap bookings, each containing:
- `other_bookings` - A list of other bookings at the same day(s) as the declined
- `overview` - A list of all bookings in the calendar for the day(s) of the
declined booking
- `alternatives` - List of alternative calendars (rooms) for the time slot of
the
**Attributes**
Each event has the following attributes:
- `uid` - The unique ID of the event
- `name` - The name of the event
- `start` - The start time (datetime) of the event
- `end` - The end time (datetime) of the event
- `duration` - The duration of the event (timedelta)
- `created` - The datetime the event was created
- `status` - The status of the event
- `organizer` - The organizer of the event
- `is_deleted` - Whether the event was deleted
- `is_cancelled` - Whether the event was cancelled
- `is_recurring` - Whether the event is recurring
- `is_prioritized` - Whether the event is prioritized
- `is_deleted` - Whether the event was deleted
- `is_priorized` - Whether the event is prioritized
- `is_deprioritized` - Whether the event is deprioritized in favor of another
event
- `organizer` - The organizer of the event
- `overlap_events` - A list of overlap events, each containing:
- `name` - The name of the overlap event
- `start` - The start time (datetime) of the overlap event
- `end` - The end time (datetime) of the overlap event
- `created` - The datetime the overlap event was created
- `organizer` - The organizer of the overlap event
- `is_priorized` - Whether the overlap event is prioritized
- `calendar_name` - The name of the calendar

View File

@ -16,7 +16,6 @@ services:
- SMTP_SERVER=your.smtp.server # adjust this
- SMTP_PORT=587 # adjust this if necessary
- SMTP_SENDER_NAME=Room Booking System # adjust this if you want
- MAX_SEARCH_HORIZON=14 # adjust maximal range of days (longer search periods will get split into multiple requests)
volumes:
app-data:

View File

@ -74,12 +74,29 @@ class EmailTemplate(models.Model):
id = models.AutoField(primary_key=True)
name = models.CharField(max_length=100, unique=True)
subject = models.CharField(max_length=200)
plaintext = models.TextField(blank=True, help_text="""\
body = models.TextField(help_text="""\
Email template to use for sending notifications about deleted events.<br>
If not set, no email will be sent.<br>
Jinja2 template syntax can be used in the subject and body.<br>
See README.md for a list of available variables.""")
html = models.TextField(blank=True, help_text="""\
Jinja2 template syntax can be used in the subject and body.<br>
See README.md for a list of available variables.""")
Available variables:<br>
- name: The name of the event<br>
- start: The start time (datetime) of the event<br>
- end: The end time (datetime) of the event<br>
- created: The datetime the event was created<br>
- is_deleted: Whether the event was deleted<br>
- is_cancelled: Whether the event was cancelled<br>
- is_recurring: Whether the event is recurring<br>
- organizer: The organizer of the event<br>
- is_prioritized: Whether the event is prioritized<br>
- is_deprioritized: Whether the event was deprioritized in favor of another event<br>
- overlap_events: List of overlap events, each with:<br>
- name: The name of the overlap event<br>
- start: The start time (datetime) of the overlap event<br>
- end: The end time (datetime) of the overlap event<br>
- created: The datetime the overlap event was created<br>
- organizer: The organizer of the overlap event<br>
- is_prioritized: Whether the overlap event is prioritized<br>
- calendar_name: The name of the calendar""")
def __str__(self):
return self.name
@ -87,8 +104,7 @@ class EmailTemplate(models.Model):
@app.admin(
list_display=("name", "all_users", "auto_clear_bookings",
"auto_clear_overlap_horizon_days", "email_template",
"all_alternatives"),
"auto_clear_overlap_horizon_days", "email_template"),
list_filter=("auto_clear_bookings",))
class Calendar(models.Model):
"""
@ -108,10 +124,6 @@ class Calendar(models.Model):
help_text="""\
Email template to use for sending notifications about deleted events.<br>
If not set, no email will be sent.""")
alternatives = models.ManyToManyField("self", blank=True,
help_text="Calendars (rooms) to be "
"considered as alternatives"
" for this one.")
def save(self, *args, **kwargs):
self.url = self.url.rstrip("/")
@ -136,14 +148,6 @@ class Calendar(models.Model):
"""
return ", ".join([u.username for u in self.users.all()])
def all_alternatives(self) -> str:
"""
Get the alternatives of the calendar.
It's kind of inefficient, but in small databases it should be fine.
:return: A list of alternatives.
"""
return ", ".join([a.name for a in self.alternatives.all()])
@app.admin(ordering=("start", "end", "calendar", "name"),
list_filter=("cancelled",),
@ -172,9 +176,6 @@ class Event(models.Model):
self.created = datetime.now(tz=timezone.get_current_timezone())
self.ics = self.create_ics()
# Round down to the nearest minute to avoid collisions between events
self.end = self.end.replace(second=0, microsecond=0)
# Send the event to the CalDAV server if it has not been cancelled yet
if not self.cancelled:
try:

View File

@ -1,5 +1,3 @@
import email.utils
import json
import os
import re
import smtplib
@ -9,218 +7,28 @@ from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import format_datetime
from pprint import pprint
from typing import NamedTuple
import caldav
import pytz
from caldav import CalendarObjectResource, Principal, Calendar, DAVObject
from caldav import CalendarObjectResource
from jinja2 import Template
from django.core.exceptions import ObjectDoesNotExist
tz = pytz.timezone(os.getenv("TIME_ZONE", "Europe/Berlin"))
MAX_SEARCH_HORIZON = int(os.getenv("MAX_SEARCH_HORIZON", 14))
class DavEvent:
"""
Wrapper for calendar events fetched from a CalDAV server.
"""
uid: str
name: str
start: datetime
end: datetime
duration: timedelta
created: datetime
status: str
organizer: str
calendar: DAVObject
obj: CalendarObjectResource
missing_required_fields: list
def __init__(self, event: CalendarObjectResource):
"""
Initialize a DavEvent object.
:param event: CalendarObjectResource object representing the event.
"""
for component in event.icalendar_instance.walk():
if component.name == "VEVENT":
# Check if the event has all required fields
required = {
"uid": component.get("uid", False),
"dtstart": component.get("dtstart", False),
"dtend": component.get("dtend", False),
"dtstamp": component.get("dtstamp", False),
}
if not all(required.values()):
self.missing_required_fields = [k for k, v in required.items() if v is False]
else:
self.missing_required_fields = []
self.uid = event.icalendar_component.get("uid")
self.name = event.icalendar_component.get("summary", "No Name")
self.start = self._handle_date(event.icalendar_component.get("dtstart").dt)
self.end = self._handle_date(event.icalendar_component.get("dtend").dt)
self.duration = self.end - self.start
self.created = self._handle_date(event.icalendar_component.get("dtstamp").dt)
self.status = event.icalendar_component.get("status", "CONFIRMED")
self.organizer = event.icalendar_component.get("organizer", "").replace("mailto:", "").strip()
self.calendar = event.parent
self.obj = event
self.is_deprioritized = False
self.is_deleted = False
def __hash__(self):
return hash(self.extended_uid)
def __eq__(self, other):
if not isinstance(other, DavEvent):
return NotImplemented
return self.extended_uid == other.extended_uid
@property
def extended_uid(self):
"""
Create a more unique identifier for the event.
This is necessary to distinguish between events of the same series, which
have the same UID but different start and end times.
:return: Unique identifier string
"""
# Fallback to calendar id if UID is not set
uid = self.uid if self.uid else self.calendar.id
return f"{uid}_{int(self.start.timestamp())}_{int(self.end.timestamp())}"
@property
def is_recurring(self) -> bool:
"""
Check if the event is a recurring event.
:return: True if the event is recurring, False otherwise.
"""
return 'RECURRENCE-ID' in self.obj.icalendar_component
@property
def is_cancelled(self) -> bool:
"""
Check if the event is a cancelled event.
:return: True if the event is cancelled, False otherwise.
"""
return self.obj.icalendar_component.get("status") == "CANCELLED"
@property
def is_prioritized(self) -> bool:
"""
Check if the event is a prioritized event by finding and validating the
priority token.
:return:
"""
from booking import PriorityEventToken
description = self.obj.icalendar_component.get("DESCRIPTION", "")
uid = self.uid
# Check if there is a token in the event description
token_pattern = r"\#[A-Z0-9]{12}\#"
match = re.search(token_pattern, str(description))
if match:
token = match.group()
print(f"Priority token found in event description: {token}")
try:
token_obj = PriorityEventToken.objects.get(token=token)
except ObjectDoesNotExist:
print(f"Priority token could not be found in database: {token}")
return False
# If token is already used, verify signature
if token_obj.is_used:
# Check if the signature is valid
if token_obj.validate_signature(uid):
print(f"Priority token '{token}' is valid.")
return True
else:
print(f"Priority token '{token}' is invalid.")
# TODO: Notify about invalid token usage
return False
# If the token hasn't been used yet, redeem it
else:
print(f"Redeeming priority token '{token}' for event '{uid}'.")
token_obj.redeem(uid)
# TODO: Notify about token redemption
return True
# If no token is found, return False
return False
def delete(self):
"""
Delete the event from the calendar.
:return:
"""
self.is_deleted = True
self.obj.delete()
def cancel(self):
"""
Cancel the event.
:return:
"""
self.obj.icalendar_component["status"] = "CANCELLED"
self.obj.save(no_create=True, increase_seqno=False)
def serialize(self) -> dict[str, str | datetime | timedelta]:
"""
Serialize the event to a dictionary.
:return:
"""
return {
"uid": self.uid,
"name": self.name,
"start": self.start,
"end": self.end,
"duration": self.duration,
"created": self.created,
"status": self.status,
"organizer": self.organizer,
"calendar_id": self.calendar.id,
"calendar_name": self.calendar.name,
"is_cancelled": self.is_cancelled,
"is_recurring": self.is_recurring,
"is_prioritized": self.is_prioritized,
"missing_required_fields": self.missing_required_fields,
}
def dumps(self, indent=4, datetime_format="%Y-%m-%d %X") -> str:
"""
Dump a json string with the event data.
:return: string with json data
"""
return json.dumps({
"uid": self.uid,
"name": self.name,
"start": self.start.strftime(datetime_format),
"end": self.end.strftime(datetime_format),
"duration": self.duration.total_seconds(),
"created": self.created.strftime(datetime_format),
"status": self.status,
"organizer": self.organizer,
"calendar_id": self.calendar.id,
"calendar_name": self.calendar.name,
"is_cancelled": self.is_cancelled,
"is_recurring": self.is_recurring,
"is_prioritized": self.is_prioritized,
"missing_required_fields": self.missing_required_fields,
}, indent=indent)
@staticmethod
def _handle_date(date_value: datetime | date) -> datetime:
"""
Date objects will be converted to datetime objects in the specified timezone.
"""
if isinstance(date_value, datetime):
date_value = date_value.astimezone(tz)
else:
date_value = tz.localize(datetime.combine(date_value, datetime.min.time()))
return date_value
DavEvent = NamedTuple("DavEvent", [
("uid", str),
("name", str),
("start", datetime),
("end", datetime),
("created", datetime),
("status", str),
("organizer", str),
("obj", caldav.Event),
("is_cancelled", bool),
("is_recurring", bool),
("is_prioritized", bool),
])
def clear(target_calendars: list, is_test: bool=False) -> dict:
"""
@ -263,91 +71,107 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
# Get events from calendar
print(f"--- Clearing cancelled bookings and overlaps in calendar: {calendar.id}")
horizon = tcal_by_name[calendar.id].auto_clear_overlap_horizon_days
events_fetched = calendar.search(
start=datetime.now(),
end=date.today() + timedelta(days=horizon),
event=True,
expand=True,
split_expanded=True,
)
# Split horizon search into multiple requests if horizon is bigger
# than MAX_SEARCH_HORIZON
horizons = []
while horizon > 0:
if horizon >= MAX_SEARCH_HORIZON:
horizons.append(MAX_SEARCH_HORIZON)
else:
horizons.append(horizon)
horizon -= MAX_SEARCH_HORIZON
events_fetched = []
start_delta = 0
end_delta = 0
today = datetime.now(tz=tz).date()
for h in horizons:
end_delta += h
try:
events_fetched.extend(calendar.search(
start=today + timedelta(days=start_delta),
end=today + timedelta(days=end_delta),
event=True,
expand=True,
split_expanded=True,
))
except Exception as e:
print(f"--- Failed to fetch events for calendar: {calendar.id}: {e}")
continue
start_delta += h
# Create DavEvent objects from fetched events
events = []
for event in events_fetched:
try:
events.append(DavEvent(event))
except ValueError as e:
print(f"Error creating DavEvent object: {e}")
continue
for component in event.icalendar_instance.walk():
# Filter out events that are missing required fields
events_missing_required_fields = [
event for event in events if event.missing_required_fields
]
for event in events_missing_required_fields:
result = {
"uid": event.uid or "No UID",
"name": event.name or "No Name",
"reason": f"Missing required fields: {', '.join(event.missing_required_fields)}",
"calendar": calendar.id,
}
results["ignored_overlapping_events"].append(result)
print("Skipping event:")
pprint(result)
print("------")
continue
events = [event for event in events if not event.missing_required_fields]
# Delete cancelled non-recurring events if not in test mode
if (component.name == "VEVENT"
and is_cancelled_event(event)
and not is_recurring_event(event)):
# Delete cancelled non-recurring events if not in test mode
for event in events:
if event.is_cancelled and not event.is_recurring:
if not is_test:
event.obj.delete()
result = {
"uid": event.uid,
"name": event.name or "No Name",
"calendar": event.calendar.id,
}
results["deleted_cancelled_events"].append(result)
print("Deleted cancelled event:")
pprint(result)
print("------")
if not is_test:
event.delete()
result = {
"uid": component.get("uid"),
"name": component.get("summary", "No Name"),
"calendar": calendar.id,
}
results["deleted_cancelled_events"].append(result)
print("Deleted cancelled event:")
pprint(result)
print("------")
# Create DavEvent objects for not cancelled events
elif component.name == "VEVENT" and not is_cancelled_event(event):
# Skip events which miss required fields
required = {
"uid": component.get("uid", False),
"dtstart": component.get("dtstart", False),
"dtend": component.get("dtend", False),
"dtstamp": component.get("dtstamp", False),
}
if not all(required.values()):
result = {
"uid": component.get("uid"),
"name": component.get("summary", "No Name"),
"calendar": calendar.id,
"reason": f"Missing required fields: {', '.join([k for k, v in required.items() if v is False])}",
}
results["ignored_overlapping_events"].append(result)
print("Skipping event:")
pprint(result)
print("------")
continue
# Create DavEvent object
events.append(
DavEvent(
uid=component.get("uid"),
name=component.get("summary", "No Name"),
start=handle_date(component.get("dtstart").dt),
end=handle_date(component.get("dtend").dt),
created=component.get("dtstamp").dt.astimezone(tz),
status=component.get("status", "CONFIRMED"),
organizer=component.get("organizer", "").replace("mailto:", "").strip(),
obj=event,
is_cancelled=is_cancelled_event(event),
is_recurring=is_recurring_event(event),
is_prioritized=is_prioritized_event(event),
)
)
# Find overlapping events
overlapping_events = find_overlapping_events(events)
overlapping_events_json = json.dumps([json.loads(o.get("event").dumps()) for _, o in overlapping_events.items()], indent=2)
print(f"Found overlapping events:\n{overlapping_events_json}")
# Delete overlapping events and send emails to organizers
for overlap in overlapping_events.values():
event = overlap["event"]
overlaps_with = overlap["overlaps_with"]
event.is_deprioritized = any(ov.is_prioritized for ov in overlaps_with)
is_deleted = False
is_cancelled = False
is_deprioritized = any(ov.is_prioritized for ov in overlaps_with)
result = {
"event": event.serialize(),
"overlaps_with": [ov.serialize() for ov in overlaps_with],
"event": {
"uid": event.uid,
"name": event.name,
"start": event.start,
"end": event.end,
"created": event.created,
"organizer": event.organizer,
"is_recurring": event.is_recurring,
"is_prioritized": event.is_prioritized,
"is_deprioritized": is_deprioritized,
},
"overlaps_with": [{
"uid": ov.uid,
"name": ov.name,
"start": ov.start,
"end": ov.end,
"created": ov.created,
"organizer": ov.organizer,
"is_recurring": ov.is_recurring,
"is_prioritized": ov.is_prioritized,
} for ov in overlaps_with],
"calendar": calendar.id,
}
try:
@ -355,8 +179,8 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
if not event.is_recurring:
# but only if we are not in test mode
if not is_test:
event.delete()
event.obj.delete()
is_deleted = True
result["cancellation_type"] = "deleted"
results["deleted_overlapping_events"].append(result)
print("Deleted overlapping event:")
@ -365,66 +189,35 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
# we need to cancel it now
elif not event.is_cancelled:
if not is_test:
event.cancel()
event.obj.decline_invite()
event.obj.icalendar_component["status"] = "CANCELLED"
event.obj.save()
is_cancelled = True
result["cancellation_type"] = "cancelled"
results["cancelled_overlapping_recurring_events"].append(result)
print("Cancelled overlapping recurring event:")
# Get all events of the affected calendar for the days of the
# declined event
other_bookings = find_other_bookings(event, calendar)
# Find free alternative rooms for the time slot of the declined event
alternatives = find_alternatives(event, principal, target_calendars)
# Create a set for all events as an overview
event.is_declined = True
event.is_overlapping = False
event.is_other_booking = False
overview = {event}
# Add the overlapping events to the overview
for ov in overlaps_with:
ov.is_declined = False
ov.is_overlapping = True
ov.is_other_booking = False
overview.add(ov)
# Add the other bookings to the overview
for ob in other_bookings:
ob.is_declined = False
ob.is_overlapping = False
ob.is_other_booking = True
overview.add(ob)
# Make a list from the set to ensure order
overview = list(overview)
# Sort the overview by start time
overview.sort(key=lambda ev: ev.start)
# Send email to organizer of the event if not in test mode
email_template = tcal_by_name[calendar.id].email_template
if email_template:
if email_template and not is_test:
try:
send_mail_to_organizer(
booking=event,
overlap_bookings=overlaps_with,
event_overview=overview,
other_bookings=other_bookings,
alternatives=alternatives,
calendar=calendar,
email_template=email_template,
is_test=is_test,
)
send_mail_to_organizer(event, overlaps_with, calendar,
is_deleted, is_cancelled,
is_deprioritized, email_template)
result["email_sent"] = True
except Exception as e:
print("Failed to send email to organizer for event "
f"{event.name}: {e}")
result["email_sent"] = False
elif email_template and is_test:
print("Email template found, but not sending email in test mode.")
send_mail_to_organizer(event, overlaps_with, calendar,
is_deleted, is_cancelled,
is_deprioritized, email_template,
is_test=True)
result["email_sent"] = True
else:
print(f"No email template found for calendar {calendar.id}. "
"Skipping email sending.")
result["email_sent"] = False
# Print results
@ -461,8 +254,7 @@ def find_overlapping_events(events: list[DavEvent]) -> dict:
for event in events:
# Skip if the event is already in the overlapping events dictionary
# or if it is already canceled
if overlapping_events.get(event.extended_uid) or event.is_cancelled:
if overlapping_events.get(create_event_identifier(event)):
continue
for compare_event in events:
@ -471,12 +263,6 @@ def find_overlapping_events(events: list[DavEvent]) -> dict:
if event.uid == compare_event.uid:
continue
# Skip if the compare_event is already in the overlapping events
# dictionary or if it is already canceled
if (overlapping_events.get(compare_event.extended_uid)
or compare_event.is_cancelled):
continue
# Check if the events overlap
if event_starts_later(event, compare_event) or \
event_starts_earlier(event, compare_event) or \
@ -496,44 +282,114 @@ def find_overlapping_events(events: list[DavEvent]) -> dict:
continue
# Add to overlapping events dictionary
if e := overlapping_events.get(event.extended_uid):
event_identifier = create_event_identifier(compare_event)
if e := overlapping_events.get(event_identifier):
# If the event is already in the dictionary, extend the overlaps
e["overlaps_with"].append(event)
else:
# Create a new entry for the overlapping event
overlapping_events[event.extended_uid] = {
overlapping_events[event_identifier] = {
"event": compare_event,
"overlaps_with": [event]
}
return overlapping_events
def create_event_identifier(event: DavEvent) -> str:
"""
Create a unique identifier for the event.
This is necessary to distinguish between events of the same series, which
have the same UID but different start and end times.
:param event: DavEvent object
:return: Unique identifier string
"""
return f"{event.uid}_{int(event.start.timestamp())}_{int(event.end.timestamp())}"
def is_recurring_event(event: CalendarObjectResource) -> bool:
"""
Check if the event is a recurring event.
:param event: CalendarObjectResource object representing the event.
:return: True if the event is recurring, False otherwise.
"""
return 'RECURRENCE-ID' in event.icalendar_component
def is_cancelled_event(event: CalendarObjectResource) -> bool:
"""
Check if the event is a cancelled event.
:param event: CalendarObjectResource object representing the event.
:return: True if the event is cancelled, False otherwise.
"""
return event.icalendar_component.get("status") == "CANCELLED"
def is_prioritized_event(event: CalendarObjectResource) -> bool:
"""
Check if the event is a prioritized event by finding and validating the
priority token.
:param event:
:return:
"""
from booking import PriorityEventToken
description = event.icalendar_component.get("DESCRIPTION", "")
uid = event.icalendar_component.get("uid")
# Check if there is a token in the event description
token_pattern = r"\#[A-Z1-9]{12}\#"
match = re.search(token_pattern, description)
if match:
token = match.group()
print(f"Priority token found in event description: {token}")
token_obj = PriorityEventToken.objects.get(token=token)
# Check if the token object exists in the database
if not token_obj:
print(f"Priority token '{token}' not found in database.")
return False
# If token is already used, verify signature
if token_obj.is_used:
# Check if the signature is valid
if token_obj.validate_signature(uid):
print(f"Priority token '{token}' is valid.")
return True
else:
print(f"Priority token '{token}' is invalid.")
# TODO: Notify about invalid token usage
return False
# If the token hasn't been used yet, redeem it
else:
print(f"Redeeming priority token '{token}' for event '{uid}'.")
token_obj.redeem(uid)
# TODO: Notify about token redemption
return True
# If no token is found, return False
return False
def send_mail_to_organizer(
booking: DavEvent,
overlap_bookings: list[DavEvent],
event_overview: list[DavEvent],
event: DavEvent,
overlap_events: list[DavEvent],
calendar: caldav.Calendar,
is_deleted: bool,
is_cancelled: bool,
is_deprioritized: bool,
email_template,
other_bookings: list[DavEvent]=None,
alternatives: list[Calendar]=None,
is_test: bool=False
):
"""
Send email to organizer of the event.
:param booking: Booking that was declined
:param overlap_bookings: List of bookings which overlap with the declined event
:param event_overview: Sorted list of all events in the calendar, including the declined event and overlaps
:param event: Event that was declined
:param overlap_events: List of events which overlap with the declined event
:param calendar: Calendar to send the email from
:param is_deleted: True if the event was deleted, False otherwise
:param is_cancelled: True if the event was cancelled, False otherwise
:param is_deprioritized: True if the event was deprioritized in favor of
another event, False otherwise
:param email_template: Email template to use for the email
:param other_bookings: List of other bookings at the same day(s) as the declined booking
:param alternatives: List of alternative calendars (rooms) for the time slot of the declined booking
:param is_test: If True, do not send the email, just print it
"""
if is_test:
print("Not sending email in test mode.")
# Check if environment variables for SMTP are set
if not all([
os.getenv('SMTP_EMAIL'),
@ -542,15 +398,28 @@ def send_mail_to_organizer(
]):
raise Exception("SMTP environment variables are not set.")
recipient = booking.organizer
recipient = event.organizer
# Prepare the email content
context = {
"booking": booking,
"overlap_bookings": overlap_bookings,
"other_bookings": other_bookings if other_bookings else [],
"overview": event_overview,
"alternatives": alternatives if alternatives else [],
"name": event.name,
"start": event.start,
"end": event.end,
"created": event.created,
"is_deleted": is_deleted,
"is_cancelled": is_cancelled,
"is_recurring": is_recurring_event(event.obj),
"organizer": event.organizer,
"is_prioritized": event.is_prioritized,
"is_deprioritized": is_deprioritized,
"overlap_events": [{
"name": overlap_event.name,
"start": overlap_event.start,
"end": overlap_event.end,
"created": overlap_event.created,
"organizer": overlap_event.organizer,
"is_prioritized": overlap_event.is_prioritized,
} for overlap_event in overlap_events],
"calendar_name": calendar.get_display_name(),
}
@ -558,8 +427,8 @@ def send_mail_to_organizer(
# Create the email message
message = MIMEMultipart("alternative")
message["From"] = email.utils.formataddr((os.getenv('SMTP_SENDER_NAME', 'Room Booking'), os.getenv('SMTP_EMAIL')))
message["To"] = email.utils.formataddr((None, recipient))
message["From"] = f"{os.getenv('SMTP_SENDER_NAME', 'Room Booking')} <{os.getenv('SMTP_EMAIL')}>"
message["To"] = recipient
if bcc := os.getenv("SMTP_BCC"):
message["Bcc"] = bcc
recipients += bcc.split(",")
@ -567,16 +436,18 @@ def send_mail_to_organizer(
# Try to render subject and body from the email template
message["Subject"] = Template(email_template.subject).render(**context)
plaintext = Template(email_template.plaintext).render(**context)
html = Template(email_template.html).render(**context)
message.attach(MIMEText(plaintext, "plain", "utf-8"))
message.attach(MIMEText(html, "html", "utf-8"))
body = Template(email_template.body).render(**context)
message.attach(MIMEText(body, "plain"))
# Create a secure SSL context
ssl_context = ssl.create_default_context()
# Send the email with starttls or SSL based on environment variable
print(f"Sending email to {', '.join(recipients)} with subject: {message['Subject']}")
overlaps = ", ".join([f"'{overlap.name}'" for overlap in overlap_events])
print(f"Sending email to {', '.join(recipients)} with subject: {message['Subject']}:")
print('"""')
print(body)
print('"""')
if os.getenv('SMTP_STARTTLS', False):
print("Using STARTTLS for SMTP connection.")
if not is_test:
@ -588,97 +459,21 @@ def send_mail_to_organizer(
else:
print("Using SSL for SMTP connection.")
if not is_test:
with smtplib.SMTP_SSL(os.getenv('SMTP_SERVER'),
with smtplib.SMTP_SSL("mail.extrasolar.space",
os.getenv('SMTP_PORT', 465),
context=ssl_context) as server:
server.login(os.getenv("SMTP_USER_NAME", os.getenv("SMTP_EMAIL")), os.getenv("SMTP_PASSWORD"))
server.sendmail(os.getenv("SMTP_EMAIL"), recipients, message.as_string())
def find_other_bookings(event: DavEvent, calendar: caldav.Calendar) -> list[DavEvent]:
def handle_date(date_value: datetime | date) -> datetime:
"""
Find other bookings in the calendar that are at the same day(s) as the declined event.
:param event: The event to check for overlaps.
:param calendar: The calendar to search in.
:return: List of DavEvent objects.
Date objects will be converted to datetime objects in the specified timezone.
"""
# Get all events in the calendar for the days of the declined event
start_date = event.start.date()
end_date = event.end.date()
events_fetched = calendar.search(
start=start_date,
end=end_date + timedelta(days=1),
event=True,
expand=True,
split_expanded=True,
)
# Create DavEvent objects from fetched events
all_events = [DavEvent(event) for event in events_fetched]
# Filter out cancelled events and events that are
# - the exact same as the given event
# - that do not have all required fields
return [e for e in all_events if not e.is_cancelled
and e.extended_uid != event.extended_uid
and not e.missing_required_fields]
def find_alternatives(event: DavEvent, principal: Principal,
target_calendars: list) -> list[Calendar]:
"""
Find alternative rooms for the time slot of the declined event.
:param event: The event to find alternatives for.
:param principal: The principal to search for alternative rooms.
:param target_calendars: Calendars administrated by the current user
:return: List of alternative calendars that are available for the time slot.
"""
# Get all calendars of the principal
calendars = principal.calendars()
alternative_calendars = {
calendar.name: [c.name for c in calendar.alternatives.all()]
for calendar in target_calendars
}
print("Alternative calendars for event calendar: "
f"{', '.join([a for a in alternative_calendars[event.obj.parent.id]])}")
alternatives = []
for calendar in calendars:
# Skip the calendar of the declined event
if calendar.id == event.obj.parent.id:
continue
# Skip if the calendar is not administrated by the current user
if calendar.id not in alternative_calendars.keys():
continue
# Skip if the calendar is not an alternative to the calendar of the
# declined event
if calendar.id not in alternative_calendars[event.obj.parent.id]:
continue
# Search for events in the alternative calendar that overlap with the
# declined event
events_fetched = calendar.search(
start=event.start,
end=event.end,
event=True,
expand=True,
split_expanded=True,
)
# Create DavEvent objects from fetched events
all_events = [DavEvent(event) for event in events_fetched]
# Filter out cancelled events and events that are missing required fields
blocked = [e for e in all_events if not e.is_cancelled
and not e.missing_required_fields]
# If there are no blocked events, the calendar is available
if not blocked:
alternatives.append(calendar)
print("Available alternative calendars for event: "
f"{', '.join([a.id for a in alternatives])}")
return alternatives
if isinstance(date_value, datetime):
date_value = date_value.astimezone(tz)
elif isinstance(date_value, date):
date_value = tz.localize(datetime.combine(date_value, datetime.min.time()))
else:
raise ValueError(f"date_value must be a datetime or date object, {type(date_value)} given.")
return date_value

View File

@ -1,29 +0,0 @@
# Generated by Django 5.1.4 on 2025-07-21 14:10
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("booking", "0011_alter_emailtemplate_body_and_more"),
]
operations = [
migrations.AddField(
model_name="calendar",
name="alternatives",
field=models.ManyToManyField(
blank=True,
help_text="Calendars (rooms) to be considered as alternatives for this one.",
to="booking.calendar",
),
),
migrations.AlterField(
model_name="emailtemplate",
name="body",
field=models.TextField(
help_text=" Email template to use for sending notifications about deleted events.<br>\n If not set, no email will be sent.<br>\n Jinja2 template syntax can be used in the subject and body.<br>\n See README.md for a list of available variables."
),
),
]

View File

@ -1,33 +0,0 @@
# Generated by Django 5.1.4 on 2025-07-22 13:49
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("booking", "0012_calendar_alternatives_alter_emailtemplate_body"),
]
operations = [
migrations.RemoveField(
model_name="emailtemplate",
name="body",
),
migrations.AddField(
model_name="emailtemplate",
name="html",
field=models.TextField(
blank=True,
help_text=" Jinja2 template syntax can be used in the subject and body.<br>\n See README.md for a list of available variables.",
),
),
migrations.AddField(
model_name="emailtemplate",
name="plaintext",
field=models.TextField(
blank=True,
help_text=" Jinja2 template syntax can be used in the subject and body.<br>\n See README.md for a list of available variables.",
),
),
]

View File

@ -1 +1 @@
1.4.0
1.2.0