marc
/
room-booking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

💥 execute room booking clearing in separate script

This commit is contained in:
Marc Koch 2025-09-15 16:12:56 +02:00
parent 9c4d699036
commit 81eb7de254
12 changed files with 596 additions and 307 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.idea/RoomBooking.iml
View File

@ -4,7 +4,7 @@
<content url="file://$MODULE_DIR$"> <content url="file://$MODULE_DIR$">
<excludeFolder url="file://$MODULE_DIR$/.venv" /> <excludeFolder url="file://$MODULE_DIR$/.venv" />
</content> </content>
<orderEntry type="jdk" jdkName="Python 3.12 (RoomBooking)" jdkType="Python SDK" /> <orderEntry type="jdk" jdkName="uv (RoomBooking)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" /> <orderEntry type="sourceFolder" forTests="false" />
</component> </component>
</module> </module>

33
.idea/dataSources.xml
View File

@ -1,11 +1,11 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<project version="4"> <project version="4">
<component name="DataSourceManagerImpl" format="xml" multifile-model="true"> <component name="DataSourceManagerImpl" format="xml" multifile-model="true">
<data-source source="LOCAL" name="db" uuid="37a129b8-8038-4c13-bba8-73cf8fa77538"> <data-source source="LOCAL" name="db" uuid="582add31-3085-4739-9171-fdd62b5bf3f9">
<driver-ref>sqlite.xerial</driver-ref> <driver-ref>sqlite.xerial</driver-ref>
<synchronize>true</synchronize> <synchronize>true</synchronize>
<jdbc-driver>org.sqlite.JDBC</jdbc-driver> <jdbc-driver>org.sqlite.JDBC</jdbc-driver>
<jdbc-url>jdbc:sqlite:$PROJECT_DIR$/src/db.sqlite3</jdbc-url> <jdbc-url>jdbc:sqlite:$PROJECT_DIR$/src/data/db.sqlite3</jdbc-url>
<working-dir>$ProjectFileDir$</working-dir> <working-dir>$ProjectFileDir$</working-dir>
<libraries> <libraries>
<library> <library>
@ -16,7 +16,7 @@
</library> </library>
</libraries> </libraries>
</data-source> </data-source>
<data-source source="LOCAL" name="db [2]" uuid="653863ac-b4c8-4525-be5f-881f75f27fff"> <data-source source="LOCAL" name="db [2]" uuid="d0501638-9467-459e-9ea1-5972b821732f">
<driver-ref>sqlite.xerial</driver-ref> <driver-ref>sqlite.xerial</driver-ref>
<synchronize>true</synchronize> <synchronize>true</synchronize>
<jdbc-driver>org.sqlite.JDBC</jdbc-driver> <jdbc-driver>org.sqlite.JDBC</jdbc-driver>
@ -29,6 +29,33 @@
<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> <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> </library>
<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>
<data-source source="LOCAL" name="db [3]" uuid="1e9a9b86-5cdd-41ef-8a57-22badc9a40b8">
<driver-ref>sqlite.xerial</driver-ref>
<synchronize>true</synchronize>
<jdbc-driver>org.sqlite.JDBC</jdbc-driver>
<jdbc-url>jdbc:sqlite:$PROJECT_DIR$/src/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>
<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> </libraries>
</data-source> </data-source>
</component> </component>

2
.idea/misc.xml
View File

@ -6,5 +6,5 @@
<component name="MarkdownSettingsMigration"> <component name="MarkdownSettingsMigration">
<option name="stateVersion" value="1" /> <option name="stateVersion" value="1" />
</component> </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="uv (RoomBooking)" project-jdk-type="Python SDK" />
</project> </project>

42
Dockerfile
View File

@ -1,27 +1,41 @@
FROM python:3.12-slim-bookworm FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS builder
LABEL authors="Marc Koch"
ENV PYTHONDONTWRITEBYTECODE 1 ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED 1 ENV PYTHONUNBUFFERED=1
WORKDIR /app WORKDIR /app
RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=uv.lock,target=uv.lock \
--mount=type=bind,source=pyproject.toml,target=pyproject.toml \
uv sync --frozen --no-install-project --no-dev
COPY src/ /app
COPY README.md /app/index.md
COPY version.txt /app/
FROM python:3.12-slim-bookworm
LABEL authors="Marc Koch"
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
RUN mkdir /data RUN mkdir /data
RUN mkdir /static-collected
COPY src/ ./ COPY --from=builder --chown=app-user:app-user /app /app
COPY requirements.txt .
COPY README.md index.md
COPY version.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN groupadd -r app-user && useradd -r -g app-user app-user && \ RUN groupadd -r app-user && useradd -r -g app-user app-user && \
chown -R app-user:app-user /app && \ chown -R app-user:app-user /app && \
chown -R app-user:app-user /data chown -R app-user:app-user /data && \
chown -R app-user:app-user /static-collected && \
chmod +x /app/clear_bookings.py
WORKDIR /app
ENV PATH="/app/.venv/bin:$PATH"
USER app-user USER app-user
EXPOSE 8000 EXPOSE 8000
#CMD ["nanodjango", "serve", "booking.py"]
CMD ["gunicorn", "--timeout", "180", "--bind", "0.0.0.0:8000", "booking:app"]

97
README.md
View File

@ -5,8 +5,7 @@ This application allows you to
1. create an event in a room booking calendar via a REST API. 1. create an event in a room booking calendar via a REST API.
2. delete an event in a room booking calendar via a REST API. 2. delete an event in a room booking calendar via a REST API.
3. clear overlapping and canceled events in a room booking calendar via a REST 3. clear colliding and canceled events in a room booking calendar via script.
API.
## Setup ## Setup
@ -68,76 +67,26 @@ curl -s -X DELETE \
The response will be empty but the status code will be `204`. The response will be empty but the status code will be `204`.
### Clear overlapping and canceled events ### Clear colliding and canceled events
To clear overlapping and canceled events, you need to send a DELETE request to To clear colliding and canceled events, you can run the `clear_bookings.py`
`/api/clear-bookings`. script.
The following parameters can be passed as query parameters:
- `calendar`: The calendar to clear overlapping events from. If not specified,
all calendars of the user will be cleared.
- `test`: If set to `1` or `ok`, the API will not delete any events but only
return what would be deleted.
Curl example:
```bash ```bash
curl -s -X DELETE \ python clear_bookings.py [--calendar CALENDAR] [--dry-run DRY_RUN]
-H "Authorization: Bearer secrettoken" \
localhost:8000/api/clear-bookings?calendar=meeting-room-1&test=on" | jq "."
``` ```
The response will contain a list of events that would be deleted: The following parameters can be passed as arguments:
```json - `--calendar`: The calendars to clear collisions events from. If not
{ specified, all calendars (marked for auto clearing) will be cleared.
"deleted_cancelled_events": [], - `--dry-run`: If set, the script will only simulate the clearing and not
"deleted_overlapping_events": [ actually delete any events.
{
"event": { Execute script as short-lived docker container:
"uid": "abeda082-2a4b-48a7-a92a-ba24575622ae",
"name": "Fundraising Weekly", ```bash
"start": "2025-07-14T00:00:00+02:00", docker docker compose -f path/to/docker-compose.yaml run --rm room-booking python clear_bookings.py
"end": "2025-07-19T00:00:00+02:00",
"created": "2025-07-09T15:11:19+02:00",
"organizer": "head-of-fundraising@my-awesome.org"
},
"overlaps_with": {
"uid": "2051008d-6ce2-4489-b7d9-38d164c5e66e",
"name": "Finance Monthly",
"start": "2025-07-15T10:00:00+02:00",
"end": "2025-07-15T12:00:00+02:00",
"created": "2025-01-14T08:33:20+01:00",
"organizer": "head-of-finance@my-awesome.org"
},
"calendar": "meeting-room-1"
}
],
"cancelled_overlapping_recurring_events": [
{
"event": {
"uid": "be6d2d42-513b-45ca-bb43-663e2a10a1d7",
"name": "Job Interviews",
"start": "2025-07-14T14:00:00+02:00",
"end": "2025-07-14T15:00:00+02:00",
"created": "2025-06-30T12:04:14+02:00",
"organizer": "head-of-hr@my-awesome.org"
},
"overlaps_with": {
"uid": "f2b7b703-dfba-4ae7-a0e9-bf0e7637c7e4",
"name": "Workers Council Meeting",
"start": "2025-07-14T14:00:00+02:00",
"end": "2025-07-14T15:00:00+02:00",
"created": "2025-02-10T12:33:39+01:00",
"organizer": "workers-council@my-awesome.org"
},
"calendar": "meeting-room-2"
}
],
"ignored_overlapping_events": [],
"test_mode": true
}
``` ```
#### Email Notifications #### Email Notifications
@ -161,6 +110,8 @@ The following environment variables can be set to configure the application:
SSL and SSL and
`587` for STARTTLS. `587` for STARTTLS.
##### Email Template Variables
Email templates can be configured in the Django Admin interface. Jinja2 syntax Email templates can be configured in the Django Admin interface. Jinja2 syntax
is used for the templates. The following variables are available in the is used for the templates. The following variables are available in the
templates: templates:
@ -168,16 +119,19 @@ templates:
- `booking` - The booking that was declined - `booking` - The booking that was declined
- `calendar_name` - The name of the calendar - `calendar_name` - The name of the calendar
**Lists** ###### Lists
- `overlap_bookings` - A list of overlap bookings, each containing:
- `colliding_bookings` - A list of colliding bookings
- `other_bookings` - A list of other bookings at the same day(s) as the declined - `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 - `overview` - A list of all bookings in the calendar for the day(s) of the
declined booking declined booking
- `alternatives` - List of alternative calendars (rooms) for the time slot of - `alternatives` - List of alternative calendars (rooms) for the time slot of
the the
**Attributes** ###### Attributes
Each event has the following attributes: Each event has the following attributes:
- `uid` - The unique ID of the event - `uid` - The unique ID of the event
- `name` - The name of the event - `name` - The name of the event
- `start` - The start time (datetime) of the event - `start` - The start time (datetime) of the event
@ -191,4 +145,7 @@ Each event has the following attributes:
- `is_prioritized` - Whether the event is prioritized - `is_prioritized` - Whether the event is prioritized
- `is_deleted` - Whether the event was deleted - `is_deleted` - Whether the event was deleted
- `is_deprioritized` - Whether the event is deprioritized in favor of another - `is_deprioritized` - Whether the event is deprioritized in favor of another
event - `is_collision` - Whether the event is a collision with the booking
- `is_other_booking` - Whether the event is another booking at the same day(s)
as the booking but not colliding.
- `is_declined` - Whether the event was declined due to a collision

12
docker-compose.yaml Normal file → Executable file
View File

@ -8,15 +8,9 @@ services:
- "8000:8000" - "8000:8000"
volumes: volumes:
- app-data:/data - app-data:/data
environment: env_file:
- DJANGO_SECRET_KEY=A_VERY_SECRETKEY_KEY # adjust this - .env
- DJANGO_ALLOWED_HOSTS=www.example.org # adjust this (comma-separated list of allowed hosts) command: ".venv/bin/python -m nanodjango serve booking.py 0.0.0.0:8000"
- SMTP_EMAIL=room-booking@example.org # adjust this
- SMTP_PASSWORD=YOUR_SMTP_PASSWORD # adjust this
- 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: volumes:
app-data: app-data:

42
src/booking.py
View File

@ -17,8 +17,6 @@ from ics import Calendar as ICS_Calendar, Event as ICS_Event
from nanodjango import Django from nanodjango import Django
from shortuuid.django_fields import ShortUUIDField from shortuuid.django_fields import ShortUUIDField
from clear_bookings import clear
DEBUG = os.getenv("DJANGO_DEBUG", False) DEBUG = os.getenv("DJANGO_DEBUG", False)
SECRET_KEY = os.getenv("DJANGO_SECRET_KEY") \ SECRET_KEY = os.getenv("DJANGO_SECRET_KEY") \
if os.getenv("DJANGO_SECRET_KEY") \ if os.getenv("DJANGO_SECRET_KEY") \
@ -26,6 +24,11 @@ SECRET_KEY = os.getenv("DJANGO_SECRET_KEY") \
BASE_DIR = Path(__file__).resolve().parent BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = Path(os.getenv("DJANGO_DATA_DIR", BASE_DIR.parent / "data")) DATA_DIR = Path(os.getenv("DJANGO_DATA_DIR", BASE_DIR.parent / "data"))
if DEBUG:
print(f"DEBUG mode is {'on' if DEBUG else 'off'}")
print(f"BASE_DIR: {BASE_DIR}")
print(f"DATA_DIR: {DATA_DIR}")
# Check if all required values are set # Check if all required values are set
if not SECRET_KEY and not DEBUG: if not SECRET_KEY and not DEBUG:
print("DJANGO_SECRET_KEY is not set") print("DJANGO_SECRET_KEY is not set")
@ -87,7 +90,7 @@ class EmailTemplate(models.Model):
@app.admin( @app.admin(
list_display=("name", "all_users", "auto_clear_bookings", list_display=("name", "all_users", "auto_clear_bookings",
"auto_clear_overlap_horizon_days", "email_template", "auto_clear_collision_horizon_days", "email_template",
"all_alternatives"), "all_alternatives"),
list_filter=("auto_clear_bookings",)) list_filter=("auto_clear_bookings",))
class Calendar(models.Model): class Calendar(models.Model):
@ -101,7 +104,7 @@ class Calendar(models.Model):
users = models.ManyToManyField(app.settings.AUTH_USER_MODEL, users = models.ManyToManyField(app.settings.AUTH_USER_MODEL,
related_name='calendars') related_name='calendars')
auto_clear_bookings = models.BooleanField(default=False) auto_clear_bookings = models.BooleanField(default=False)
auto_clear_overlap_horizon_days = models.IntegerField(default=30) auto_clear_collision_horizon_days = models.IntegerField(default=30)
email_template = models.ForeignKey(EmailTemplate, on_delete=models.SET_NULL, email_template = models.ForeignKey(EmailTemplate, on_delete=models.SET_NULL,
null=True, blank=True, null=True, blank=True,
related_name='calendars', related_name='calendars',
@ -458,37 +461,6 @@ def delete_event(request, calendar: str, event_id: str):
return 204, None return 204, None
@api.delete("/clear-bookings",
response={200: dict, 204: None, 401: None, 404: None})
@csrf_exempt
def clear_bookings(request, calendar: str = None, test: bool = False):
user = get_user(request)
# Get optional calendar name from the request
if calendar:
cal = get_object_or_404(Calendar, name=calendar,
auto_clear_bookings=True)
if user not in cal.users.all():
raise HttpError(401,
f"User not authorised to clear bookings in calendar '{cal.name}'")
calendars = [cal]
# If no calendar is specified, get all calendars for the user that have
# auto_clear_bookings enabled
else:
calendars = user.calendars.filter(auto_clear_bookings=True)
if not calendars:
return 204, None # No bookings to clear
result = clear(list(calendars), is_test=test)
if all(len(r) == 0 for r in result.values() if type(r) is list):
return 204, None # No bookings to clear
else:
return 200, result
app.route("api/", include=api.urls) app.route("api/", include=api.urls)

435
src/clear_bookings.py
View File

@ -1,5 +1,8 @@
#! /app/.venv/bin/python
import email.utils import email.utils
import json import json
import logging
import os import os
import re import re
import smtplib import smtplib
@ -8,17 +11,27 @@ from datetime import datetime, date, timedelta
from email.mime.multipart import MIMEMultipart from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText from email.mime.text import MIMEText
from email.utils import format_datetime from email.utils import format_datetime
from pprint import pprint
import caldav import caldav
import click
import pytz import pytz
from caldav import CalendarObjectResource, Principal, Calendar, DAVObject from caldav import CalendarObjectResource, Principal, Calendar, DAVObject
from jinja2 import Template
from django.core.exceptions import ObjectDoesNotExist from django.core.exceptions import ObjectDoesNotExist
from jinja2 import Template
from booking import PriorityEventToken, Calendar as RoomBookingCalendar
from mylogger.logger import setup_logging
tz = pytz.timezone(os.getenv("TIME_ZONE", "Europe/Berlin")) tz = pytz.timezone(os.getenv("TIME_ZONE", "Europe/Berlin"))
MAX_SEARCH_HORIZON = int(os.getenv("MAX_SEARCH_HORIZON", 14)) MAX_SEARCH_HORIZON = int(os.getenv("MAX_SEARCH_HORIZON", 14))
FILE_LOG_LEVEL = os.getenv("FILE_LOG_LEVEL", "INFO")
STDOUT_LOG_LEVEL = os.getenv("STDOUT_LOG_LEVEL", "DEBUG")
LOG_DIR = os.getenv("LOG_DIR", "/data/room-booking.log.jsonl")
LOGROTATE_BACKUP_COUNT = int(os.getenv("LOGROTATE_BACKUP_COUNT", 6))
LOGROTATE_MAX_BYTES = int(os.getenv("LOGROTATE_MAX_BYTES", 10 * 1024 * 1024))
class DavEvent: class DavEvent:
""" """
Wrapper for calendar events fetched from a CalDAV server. Wrapper for calendar events fetched from a CalDAV server.
@ -51,18 +64,26 @@ class DavEvent:
"dtstamp": component.get("dtstamp", False), "dtstamp": component.get("dtstamp", False),
} }
if not all(required.values()): if not all(required.values()):
self.missing_required_fields = [k for k, v in required.items() if v is False] self.missing_required_fields = [k for k, v in
required.items() if
v is False]
else: else:
self.missing_required_fields = [] self.missing_required_fields = []
self.uid = event.icalendar_component.get("uid") self.uid = event.icalendar_component.get("uid")
self.name = event.icalendar_component.get("summary", "No Name") self.name = event.icalendar_component.get("summary", "No Name")
self.start = self._handle_date(event.icalendar_component.get("dtstart").dt) self.start = self._handle_date(
self.end = self._handle_date(event.icalendar_component.get("dtend").dt) event.icalendar_component.get("dtstart").dt)
self.end = self._handle_date(
event.icalendar_component.get("dtend").dt)
self.duration = self.end - self.start self.duration = self.end - self.start
self.created = self._handle_date(event.icalendar_component.get("dtstamp").dt) self.created = self._handle_date(
self.status = event.icalendar_component.get("status", "CONFIRMED") event.icalendar_component.get("dtstamp").dt)
self.organizer = event.icalendar_component.get("organizer", "").replace("mailto:", "").strip() self.status = event.icalendar_component.get("status",
"CONFIRMED")
self.organizer = event.icalendar_component.get("organizer",
"").replace(
"mailto:", "").strip()
self.calendar = event.parent self.calendar = event.parent
self.obj = event self.obj = event
self.is_deprioritized = False self.is_deprioritized = False
@ -111,7 +132,6 @@ class DavEvent:
priority token. priority token.
:return: :return:
""" """
from booking import PriorityEventToken
description = self.obj.icalendar_component.get("DESCRIPTION", "") description = self.obj.icalendar_component.get("DESCRIPTION", "")
uid = self.uid uid = self.uid
@ -165,34 +185,13 @@ class DavEvent:
self.obj.icalendar_component["status"] = "CANCELLED" self.obj.icalendar_component["status"] = "CANCELLED"
self.obj.save(no_create=True, increase_seqno=False) self.obj.save(no_create=True, increase_seqno=False)
def serialize(self) -> dict[str, str | datetime | timedelta]: def dumps(self, datetime_format="%Y-%m-%d %X") -> dict:
""" """
Serialize the event to a dictionary. Dump a dictionary with the event data.
:return: :param datetime_format: Format for datetime fields
"""
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: string with json data
""" """
return json.dumps({ return {
"uid": self.uid, "uid": self.uid,
"name": self.name, "name": self.name,
"start": self.start.strftime(datetime_format), "start": self.start.strftime(datetime_format),
@ -207,7 +206,7 @@ class DavEvent:
"is_recurring": self.is_recurring, "is_recurring": self.is_recurring,
"is_prioritized": self.is_prioritized, "is_prioritized": self.is_prioritized,
"missing_required_fields": self.missing_required_fields, "missing_required_fields": self.missing_required_fields,
}, indent=indent) }
@staticmethod @staticmethod
def _handle_date(date_value: datetime | date) -> datetime: def _handle_date(date_value: datetime | date) -> datetime:
@ -217,52 +216,72 @@ class DavEvent:
if isinstance(date_value, datetime): if isinstance(date_value, datetime):
date_value = date_value.astimezone(tz) date_value = date_value.astimezone(tz)
else: else:
date_value = tz.localize(datetime.combine(date_value, datetime.min.time())) date_value = tz.localize(
datetime.combine(date_value, datetime.min.time()))
return date_value return date_value
def clear(target_calendars: list, is_test: bool=False) -> dict: @click.command()
@click.option("--calendars", "-c", multiple=True, required=False,
help="Calendars to check for collisions (can be specified multiple times).")
@click.option("--dry-run", is_flag=True,
help="Run in test mode (do not delete events, just print what would be deleted).",
default=False)
def clear_bookings(calendars: list, dry_run: bool = False):
""" """
Clear overlapping and cancelled events in calendars. Clear colliding and cancelled events in calendars.
:param target_calendars: List of calendars to check for overlaps. :param calendars: List of calendars to check for collisions.
:param is_test: Do not delete events, just print what would be deleted. :param dry_run: Do not delete events, just print what would be deleted.
:return: Dictionary with results of the operation.
""" """
print(f"--- Clear bookings.{' Test mode enabled.' if is_test else ''}")
results = { logger = logging.getLogger(__name__)
"deleted_cancelled_events": [], setup_logging(file_log_level=FILE_LOG_LEVEL,
"deleted_overlapping_events": [], stdout_log_level=STDOUT_LOG_LEVEL,
"cancelled_overlapping_recurring_events": [], logdir=LOG_DIR)
"ignored_overlapping_events": [],
"test_mode": is_test, logger.debug(f"Clear bookings.{' Test mode enabled.' if dry_run else ''}")
} # Fetch calendar objects from database
if len(calendars) > 0:
calendars_found = list(
RoomBookingCalendar.objects.filter(name__in=calendars))
for cal in calendars:
if not cal in [c.name for c in calendars_found]:
logger.debug(f"Calendar '{cal}' not found in database. Skipping.")
else:
calendars_found = list(RoomBookingCalendar.objects.all())
if not calendars_found:
logger.debug("No valid calendars found. Exiting.")
exit(0)
calendars = calendars_found
dav_client = caldav.DAVClient( dav_client = caldav.DAVClient(
url=target_calendars[0].url, url=calendars[0].url,
username=target_calendars[0].username, username=calendars[0].username,
password=target_calendars[0].calendar_password, password=calendars[0].calendar_password,
) )
principal = dav_client.principal() principal = dav_client.principal()
# Filter calendars to only those that are in the target_calendars list # Filter calendars to only those that are in the calendars list
# Note: The id of the objects returned by principal.calendars() corresponds # Note: The id of the objects returned by principal.calendars() corresponds
# to the name of the Django Calendar model instances. # to the name f the Django Calendar model instances.
tcal_by_name = {c.name: c for c in target_calendars} tcal_by_name = {c.name: c for c in calendars}
calendars = [cal for cal in principal.calendars() calendars_filtered = [cal for cal in principal.calendars()
if cal.id in tcal_by_name.keys()] if cal.id in tcal_by_name.keys()]
if not calendars: if not calendars_filtered:
print("--- No calendars to clear overlaps in. Exiting.") logger.debug("No calendars to clear collisions in. Exiting.")
return results exit(0)
for calendar in calendars: for calendar in calendars_filtered:
# Get events from calendar # Get events from calendar
print(f"--- Clearing cancelled bookings and overlaps in calendar: {calendar.id}") logger.debug(
horizon = tcal_by_name[calendar.id].auto_clear_overlap_horizon_days f"Clearing cancelled bookings and collisions in calendar: {calendar.id}")
horizon = tcal_by_name[calendar.id].auto_clear_collision_horizon_days
# Split horizon search into multiple requests if horizon is bigger # Split horizon search into multiple requests if horizon is bigger
# than MAX_SEARCH_HORIZON # than MAX_SEARCH_HORIZON
@ -289,7 +308,13 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
split_expanded=True, split_expanded=True,
)) ))
except Exception as e: except Exception as e:
print(f"--- Failed to fetch events for calendar: {calendar.id}: {e}") logger.exception(
f"Failed to fetch events for calendar: {calendar.id}: {e}",
extra={
"calendar": calendar.id,
"max_search_horizon": MAX_SEARCH_HORIZON,
"username": calendars[0].username,
})
continue continue
start_delta += h start_delta += h
@ -297,9 +322,12 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
events = [] events = []
for event in events_fetched: for event in events_fetched:
try: try:
events.append(DavEvent(event)) events.append(DavEvent(event))
except ValueError as e: except ValueError as e:
print(f"Error creating DavEvent object: {e}") logger.exception(f"Could not create DavEvent object.",
exc_info=e, extra={
"event": str(event)
})
continue continue
# Filter out events that are missing required fields # Filter out events that are missing required fields
@ -307,94 +335,92 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
event for event in events if event.missing_required_fields event for event in events if event.missing_required_fields
] ]
for event in events_missing_required_fields: for event in events_missing_required_fields:
result = { event_result = {
"uid": event.uid or "No UID", "uid": event.uid or "No UID",
"name": event.name or "No Name", "name": event.name or "No Name",
"reason": f"Missing required fields: {', '.join(event.missing_required_fields)}", "reason": f"Missing required fields: {', '.join(event.missing_required_fields)}",
"calendar": calendar.id, "calendar": calendar.id,
} }
results["ignored_overlapping_events"].append(result) logger.warning("Skipping event because missing required fields",
print("Skipping event:") extra=event_result)
pprint(result)
print("------")
continue continue
events = [event for event in events if not event.missing_required_fields] events = [event for event in events if
not event.missing_required_fields]
# Delete cancelled non-recurring events if not in test mode # Delete cancelled non-recurring events if not in dry_run mode
for event in events: for event in events:
if event.is_cancelled and not event.is_recurring: if event.is_cancelled and not event.is_recurring:
if not is_test: if not dry_run:
event.obj.delete() event.obj.delete()
result = { event_result = {
"uid": event.uid, "uid": event.uid,
"name": event.name or "No Name", "name": event.name or "No Name",
"calendar": event.calendar.id, "calendar": event.calendar.id,
} }
results["deleted_cancelled_events"].append(result) logger.info("Deleted cancelled event.", extra=event_result)
print("Deleted cancelled event:")
pprint(result)
print("------")
# Find overlapping events # Find collisions
overlapping_events = find_overlapping_events(events) collisions = find_collisions(events)
overlapping_events_json = json.dumps([json.loads(o.get("event").dumps()) for _, o in overlapping_events.items()], indent=2) logger.info(
print(f"Found overlapping events:\n{overlapping_events_json}") f"Found {len(collisions)} event collisions.", extra={
"calendar": calendar.id,
"collisions": [{"event": ov["event"].dumps(),
"collides_with": [cw.dumps() for cw in
ov['collides_with']]} for ov
in collisions.values()],
})
# Delete overlapping events and send emails to organizers # Delete colliding events and send emails to organizers
for overlap in overlapping_events.values(): for collision in collisions.values():
event = overlap["event"] event = collision["event"]
overlaps_with = overlap["overlaps_with"] collides_with = collision["collides_with"]
event.is_deprioritized = any(ov.is_prioritized for ov in overlaps_with) event.is_deprioritized = any(
result = { ov.is_prioritized for ov in collides_with)
"event": event.serialize(), collision_result = {
"overlaps_with": [ov.serialize() for ov in overlaps_with], "event": event.dumps(),
"collides_with": [ov.dumps() for ov in collides_with],
"calendar": calendar.id, "calendar": calendar.id,
} }
try: try:
# If this is not a recurring event, we can delete it directly # If this is not a recurring event, we can delete it directly
if not event.is_recurring: if not event.is_recurring:
# but only if we are not in test mode # but only if we are not in dry_run mode
if not is_test: if not dry_run:
event.delete() event.delete()
collision_result["cancellation_type"] = "deleted"
result["cancellation_type"] = "deleted"
results["deleted_overlapping_events"].append(result)
print("Deleted overlapping event:")
# If this is a recurring event, and not already cancelled, # If this is a recurring event, and not already cancelled,
# we need to cancel it now # we need to cancel it now
elif not event.is_cancelled: elif not event.is_cancelled:
if not is_test: if not dry_run:
event.cancel() event.cancel()
result["cancellation_type"] = "cancelled" collision_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 # Get all events of the affected calendar for the days of the
# declined event # declined event
other_bookings = find_other_bookings(event, calendar) other_bookings = find_other_bookings(event, calendar)
# Find free alternative rooms for the time slot of the declined event # Find free alternative rooms for the time slot of the declined event
alternatives = find_alternatives(event, principal, target_calendars) alternatives = find_alternatives(event, principal, calendars)
# Create a set for all events as an overview # Create a set for all events as an overview
event.is_declined = True event.is_declined = True
event.is_overlapping = False event.is_collision = False
event.is_other_booking = False event.is_other_booking = False
overview = {event} overview = {event}
# Add the overlapping events to the overview # Add the colliding events to the overview
for ov in overlaps_with: for cw in collides_with:
ov.is_declined = False cw.is_declined = False
ov.is_overlapping = True cw.is_collision = True
ov.is_other_booking = False cw.is_other_booking = False
overview.add(ov) overview.add(cw)
# Add the other bookings to the overview # Add the other bookings to the overview
for ob in other_bookings: for ob in other_bookings:
if not ob.is_cancelled: if not ob.is_cancelled:
ob.is_declined = False ob.is_declined = False
ob.is_overlapping = False ob.is_collision = False
ob.is_other_booking = True ob.is_other_booking = True
overview.add(ob) overview.add(ob)
@ -404,66 +430,89 @@ def clear(target_calendars: list, is_test: bool=False) -> dict:
# Sort the overview by start time # Sort the overview by start time
overview.sort(key=lambda ev: ev.start) overview.sort(key=lambda ev: ev.start)
# Send email to organizer of the event if not in test mode # Send email to organizer of the event if not in dry_run mode
email_template = tcal_by_name[calendar.id].email_template email_template = tcal_by_name[calendar.id].email_template
if email_template: if email_template:
try: try:
send_mail_to_organizer( send_mail_to_organizer(
booking=event, booking=event,
overlap_bookings=overlaps_with, colliding_bookings=collides_with,
event_overview=overview, event_overview=overview,
other_bookings=other_bookings, other_bookings=other_bookings,
alternatives=alternatives, alternatives=alternatives,
calendar=calendar, calendar=calendar,
email_template=email_template, email_template=email_template,
is_test=is_test, is_test=dry_run,
) )
result["email_sent"] = True collision_result["email_sent"] = True
except Exception as e: except Exception as e:
print("Failed to send email to organizer for event " logger.exception(
f"{event.name}: {e}") "Failed to send email to organizer"
result["email_sent"] = False f" for event '{event.name}'",
exc_info=e,
extra={
"calendar": calendar.id,
"event": event.dumps(),
"colliding_with": [cw.dumps() for cw in
collides_with],
"overview": [ov.dumps() for ov in overview],
"alternatives": [cal.id for cal in alternatives],
"email_template": email_template.id,
})
collision_result["email_sent"] = False
else: else:
print(f"No email template found for calendar {calendar.id}. " logger.debug(
"Skipping email sending.") f"No email template found for calendar '{calendar.id}'. "
result["email_sent"] = False "Skipping email sending.",
extra={"calendar": calendar.id}
)
# Print results collision_result["email_sent"] = False
pprint(result)
print("------") # Log results
logger.info("Processed colliding event.",
extra=collision_result)
except Exception as e: except Exception as e:
print(f"Failed to delete event {event.uid}: {e}") logger.exception(
f"Failed to process colliding event '{event.uid}'.",
exc_info=e, extra={
"calendar": calendar.id,
"event": event.dumps(),
"collides_with": [ov.dumps() for ov in collides_with],
"dry_run": dry_run,
})
print("------") print("------")
continue continue
print(f"--- Clearing completed.{' Test mode enabled.' if is_test else ''}") logger.debug(f"Clearing completed.{' Test mode enabled.' if dry_run else ''}")
return results
def find_overlapping_events(events: list[DavEvent]) -> dict:
def find_collisions(events: list[DavEvent]) -> dict:
""" """
Find overlapping events. Find colliding events.
:param events: List of events to check for overlaps. :param events: List of events to check for collisions.
:return: Dictionary of overlapping events with their UIDs as keys and :return: Dictionary of colliding events with their UIDs as keys and
a dictionary containing the event and the event it overlaps with a dictionary containing the event and the event it collides with
as values. as values.
""" """
overlapping_events = {} logger = logging.getLogger(__name__)
colliding_events = {}
# Order events by created time # Order events by created time
events = sorted(events, key=lambda item: item.created) events = sorted(events, key=lambda item: item.created)
# Define lambda functions to check for overlaps # Define lambda functions to check for collisions
event_starts_later = lambda e1, e2: e1.end >= e2.end > e1.start event_starts_later = lambda e1, e2: e1.end >= e2.end > e1.start
event_starts_earlier = lambda e1, e2: e1.end > e2.start >= e1.start event_starts_earlier = lambda e1, e2: e1.end > e2.start >= e1.start
event_inbetween = lambda e1, e2: e2.start < e1.start < e1.end < e2.end event_inbetween = lambda e1, e2: e2.start < e1.start < e1.end < e2.end
# Find overlapping events # Find colliding events
for event in events: for event in events:
# Skip if the event is already in the overlapping events dictionary # Skip if the event is already in the colliding events dictionary
# or if it is already canceled # or if it is already canceled
if overlapping_events.get(event.extended_uid) or event.is_cancelled: if colliding_events.get(event.extended_uid) or event.is_cancelled:
continue continue
for compare_event in events: for compare_event in events:
@ -476,53 +525,59 @@ def find_overlapping_events(events: list[DavEvent]) -> dict:
if compare_event.is_cancelled: if compare_event.is_cancelled:
continue continue
# Check if the events overlap # Check if the events collide
if event_starts_later(event, compare_event) or \ if event_starts_later(event, compare_event) or \
event_starts_earlier(event, compare_event) or \ event_starts_earlier(event, compare_event) or \
event_inbetween(event, compare_event): event_inbetween(event, compare_event):
# If both events are prioritized, skip the overlap # If both events are prioritized, skip the collision
if compare_event.is_prioritized and event.is_prioritized: if compare_event.is_prioritized and event.is_prioritized:
print(f"Skipping overlap between prioritized events: " logger.info("Skipping collisions detection between "
f"'{event.name}' and '{compare_event.name}'") "two prioritized events "
f"'{event.name}' and '{compare_event.name}'",
extra={"event_1": event.dumps(),
"event_2": compare_event.dumps()})
continue continue
# If the compare_event is prioritized and the event is not, # If the compare_event is prioritized and the event is not,
# skip the overlap # skip the collision
elif compare_event.is_prioritized and not event.is_prioritized: elif compare_event.is_prioritized and not event.is_prioritized:
print(f"Skipping overlap between event '{event.name}' " logger.debug("Skipping collisions detection between "
f"and prioritized event '{compare_event.name}'") f"event '{event.name}' and prioritized "
f"event '{compare_event.name}'",
extra={"event_1": event.dumps(),
"event_2": compare_event.dumps()})
continue continue
# Add to overlapping events dictionary # Add to colliding events dictionary
if e := overlapping_events.get(compare_event.extended_uid): if e := colliding_events.get(compare_event.extended_uid):
# If the event is already in the dictionary, extend the overlaps # If the event is already in the dictionary, extend the collisions
e["overlaps_with"].append(event) e["collides_with"].append(event)
else: else:
# Create a new entry for the overlapping event # Create a new entry for the colliding event
overlapping_events[compare_event.extended_uid] = { colliding_events[compare_event.extended_uid] = {
"event": compare_event, "event": compare_event,
"overlaps_with": [event] "collides_with": [event]
} }
return overlapping_events return colliding_events
def send_mail_to_organizer( def send_mail_to_organizer(
booking: DavEvent, booking: DavEvent,
overlap_bookings: list[DavEvent], colliding_bookings: list[DavEvent],
event_overview: list[DavEvent], event_overview: list[DavEvent],
calendar: caldav.Calendar, calendar: caldav.Calendar,
email_template, email_template,
other_bookings: list[DavEvent]=None, other_bookings: list[DavEvent] = None,
alternatives: list[Calendar]=None, alternatives: list[Calendar] = None,
is_test: bool=False is_test: bool = False
): ):
""" """
Send email to organizer of the event. Send email to organizer of the event.
:param booking: Booking that was declined :param booking: Booking that was declined
:param overlap_bookings: List of bookings which overlap with the declined event :param colliding_bookings: List of bookings which collide with the declined event
:param event_overview: Sorted list of all events in the calendar, including the declined event and overlaps :param event_overview: Sorted list of all events in the calendar, including the declined event and collisions
:param calendar: Calendar to send the email from :param calendar: Calendar to send the email from
another event, False otherwise another event, False otherwise
:param email_template: Email template to use for the email :param email_template: Email template to use for the email
@ -530,8 +585,10 @@ def send_mail_to_organizer(
:param alternatives: List of alternative calendars (rooms) for the time slot of 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 :param is_test: If True, do not send the email, just print it
""" """
logger = logging.getLogger(__name__)
if is_test: if is_test:
print("Not sending email in test mode.") logger.debug("Not sending email in test mode.")
# Check if environment variables for SMTP are set # Check if environment variables for SMTP are set
if not all([ if not all([
@ -543,10 +600,15 @@ def send_mail_to_organizer(
recipient = booking.organizer recipient = booking.organizer
# Do not send email if recipient is empty
if not recipient or recipient == "":
logger.debug("Not sending email to empty recipient.")
return
# Prepare the email content # Prepare the email content
context = { context = {
"booking": booking, "booking": booking,
"overlap_bookings": overlap_bookings, "colliding_bookings": colliding_bookings,
"other_bookings": other_bookings if other_bookings else [], "other_bookings": other_bookings if other_bookings else [],
"overview": event_overview, "overview": event_overview,
"alternatives": alternatives if alternatives else [], "alternatives": alternatives if alternatives else [],
@ -557,7 +619,9 @@ def send_mail_to_organizer(
# Create the email message # Create the email message
message = MIMEMultipart("alternative") message = MIMEMultipart("alternative")
message["From"] = email.utils.formataddr((os.getenv('SMTP_SENDER_NAME', 'Room Booking'), os.getenv('SMTP_EMAIL'))) message["From"] = email.utils.formataddr(
(os.getenv('SMTP_SENDER_NAME', 'Room Booking'),
os.getenv('SMTP_EMAIL')))
message["To"] = email.utils.formataddr((None, recipient)) message["To"] = email.utils.formataddr((None, recipient))
if bcc := os.getenv("SMTP_BCC"): if bcc := os.getenv("SMTP_BCC"):
message["Bcc"] = bcc message["Bcc"] = bcc
@ -575,28 +639,44 @@ def send_mail_to_organizer(
ssl_context = ssl.create_default_context() ssl_context = ssl.create_default_context()
# Send the email with starttls or SSL based on environment variable # Send the email with starttls or SSL based on environment variable
print(f"Sending email to {', '.join(recipients)} with subject: {message['Subject']}") logger.info(f"Sending email.",
extra={
"to": recipients,
"subject": message["Subject"],
"calendar": calendar.id,
"email_template": email_template.id,
"plaintext": plaintext,
"html": html,
"is_test": is_test,
})
if os.getenv('SMTP_STARTTLS', False): if os.getenv('SMTP_STARTTLS', False):
print("Using STARTTLS for SMTP connection.") logger.debug("Using STARTTLS for SMTP connection.")
if not is_test: if not is_test:
with smtplib.SMTP(os.getenv('SMTP_SERVER'), with smtplib.SMTP(os.getenv('SMTP_SERVER'),
os.getenv('SMTP_PORT', 587)) as server: os.getenv('SMTP_PORT', 587)) as server:
server.starttls(context=ssl_context) server.starttls(context=ssl_context)
server.login(os.getenv(" ", os.getenv("SMTP_EMAIL")), os.getenv("SMTP_PASSWORD")) server.login(os.getenv(" ", os.getenv("SMTP_EMAIL")),
server.sendmail(os.getenv("SMTP_EMAIL"), recipients, message.as_string()) os.getenv("SMTP_PASSWORD"))
server.sendmail(os.getenv("SMTP_EMAIL"), recipients,
message.as_string())
else: else:
print("Using SSL for SMTP connection.") logger.debug("Using SSL for SMTP connection.")
if not is_test: if not is_test:
with smtplib.SMTP_SSL(os.getenv('SMTP_SERVER'), with smtplib.SMTP_SSL(os.getenv('SMTP_SERVER'),
os.getenv('SMTP_PORT', 465), os.getenv('SMTP_PORT', 465),
context=ssl_context) as server: context=ssl_context) as server:
server.login(os.getenv("SMTP_USER_NAME", os.getenv("SMTP_EMAIL")), os.getenv("SMTP_PASSWORD")) server.login(
server.sendmail(os.getenv("SMTP_EMAIL"), recipients, message.as_string()) 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 find_other_bookings(event: DavEvent, calendar: caldav.Calendar) -> list[
DavEvent]:
""" """
Find other bookings in the calendar that are at the same day(s) as the declined event. 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 event: The event to check for collisions.
:param calendar: The calendar to search in. :param calendar: The calendar to search in.
:return: List of DavEvent objects. :return: List of DavEvent objects.
""" """
@ -621,6 +701,7 @@ def find_other_bookings(event: DavEvent, calendar: caldav.Calendar) -> list[DavE
and e.extended_uid != event.extended_uid and e.extended_uid != event.extended_uid
and not e.missing_required_fields] and not e.missing_required_fields]
def find_alternatives(event: DavEvent, principal: Principal, def find_alternatives(event: DavEvent, principal: Principal,
target_calendars: list) -> list[Calendar]: target_calendars: list) -> list[Calendar]:
""" """
@ -630,13 +711,15 @@ def find_alternatives(event: DavEvent, principal: Principal,
:param target_calendars: Calendars administrated by the current user :param target_calendars: Calendars administrated by the current user
:return: List of alternative calendars that are available for the time slot. :return: List of alternative calendars that are available for the time slot.
""" """
logger = logging.getLogger(__name__)
# Get all calendars of the principal # Get all calendars of the principal
calendars = principal.calendars() calendars = principal.calendars()
alternative_calendars = { alternative_calendars = {
calendar.name: [c.name for c in calendar.alternatives.all()] calendar.name: [c.name for c in calendar.alternatives.all()]
for calendar in target_calendars for calendar in target_calendars
} }
print("Alternative calendars for event calendar: " logger.debug("Alternative calendars for event calendar: "
f"{', '.join([a for a in alternative_calendars[event.obj.parent.id]])}") f"{', '.join([a for a in alternative_calendars[event.obj.parent.id]])}")
alternatives = [] alternatives = []
@ -654,8 +737,7 @@ def find_alternatives(event: DavEvent, principal: Principal,
if calendar.id not in alternative_calendars[event.obj.parent.id]: if calendar.id not in alternative_calendars[event.obj.parent.id]:
continue continue
# Search for events in the alternative calendar that collide with the
# Search for events in the alternative calendar that overlap with the
# declined event # declined event
events_fetched = calendar.search( events_fetched = calendar.search(
start=event.start, start=event.start,
@ -676,8 +758,13 @@ def find_alternatives(event: DavEvent, principal: Principal,
if not blocked: if not blocked:
alternatives.append(calendar) alternatives.append(calendar)
print("Available alternative calendars for event: " logger.debug("Found available alternatives for event.",
f"{', '.join([a.id for a in alternatives])}") extra={
"event": event.dumps(),
"alternatives": [a.id for a in alternatives],
})
return alternatives return alternatives
if __name__ == "__main__":
clear_bookings()

18
src/migrations/0014_rename_auto_clear_overlap_horizon_days_calendar_auto_clear_collision_horizon_days.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 5.2.6 on 2025-09-15 11:36
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("booking", "0013_remove_emailtemplate_body_emailtemplate_html_and_more"),
]
operations = [
migrations.RenameField(
model_name="calendar",
old_name="auto_clear_overlap_horizon_days",
new_name="auto_clear_collision_horizon_days",
),
]

0
src/mylogger/__init__.py Normal file
View File

165
src/mylogger/logger.py Executable file
View File

@ -0,0 +1,165 @@
import atexit
import datetime as dt
import json
import logging
import logging.config
from pathlib import Path
from typing import override
PACKAGE_PATH = Path(__file__).parent
LOG_RECORD_BUILTIN_ATTRS = {
"args",
"asctime",
"created",
"exc_info",
"exc_text",
"filename",
"funcName",
"levelname",
"levelno",
"lineno",
"module",
"msecs",
"message",
"msg",
"name",
"pathname",
"process",
"processName",
"relativeCreated",
"stack_info",
"thread",
"threadName",
"taskName",
}
def setup_logging(
logdir: Path | str | None = None,
**kwargs,
):
"""
Setup logging configuration
:param logdir: Directory to store the log file
:keyword file_log_level: Log level for the file handler
:keyword stdout_log_level: Log level for the stdout handler
:return:
"""
handlers = ["file", "stdout"]
config_file = PACKAGE_PATH / "resources" / "logging_config.json"
with open(config_file, "r") as file:
config = json.load(file)
# Override log level if provided
for handler in handlers:
if log_level := kwargs.get(f"{handler}_log_level"):
level_str = logging.getLevelName(log_level.upper())
if not isinstance(level_str, str):
level_str = logging.getLevelName(level_str)
config["handlers"][handler]["level"] = level_str
# Set log file path to user log directory
if logdir:
logdir = Path(logdir)
if logdir.is_dir():
config["handlers"]["file"]["filename"] = logdir / "room-booking.log.jsonl"
else:
config["handlers"]["file"]["filename"] = logdir
# Set max backup count if provided
if backup_count := kwargs.get("LOGROTATE_BACKUP_COUNT"):
config["handlers"]["file"]["backupCount"] = int(backup_count)
# Set max bytes if provided
if max_bytes := kwargs.get("LOGROTATE_MAX_BYTES"):
config["handlers"]["file"]["maxBytes"] = int(max_bytes)
# Create path and file if it does not exist
Path(config["handlers"]["file"]["filename"]).parent.mkdir(
parents=True, exist_ok=True)
Path(config["handlers"]["file"]["filename"]).touch()
logging.config.dictConfig(config)
queue_handler = logging.getHandlerByName("queue_handler")
if queue_handler is not None:
queue_handler.listener.start()
atexit.register(queue_handler.listener.stop)
class JSONFormatter(logging.Formatter):
"""
A custom JSON formatter for logging
"""
HIDE_KEYS = ["password", "token", "api_key", "site_key"]
def __init__(
self,
*,
fmt_keys: dict[str, str] | None = None,
):
super().__init__()
self.fmt_keys = fmt_keys if fmt_keys is not None else {}
@override
def format(self, record: logging.LogRecord) -> str:
message = self._prepare_log_dict(record)
# Exclude passwords from the log
self._hide_passwords(record)
return json.dumps(message, default=str)
def _hide_passwords(self, log_record: logging.LogRecord|dict):
"""
Recursively replace all values with keys containing 'password',
'token', etc. with '********'
:param log_record:
:return:
"""
if not isinstance(log_record, dict):
dict_obj = log_record.__dict__
else:
dict_obj = log_record
for key, value in dict_obj.items():
if isinstance(value, dict):
dict_obj = self._hide_passwords(value)
elif any(hide_key in key.lower() for hide_key in self.HIDE_KEYS):
dict_obj[key] = "********"
if isinstance(log_record, logging.LogRecord):
for key, value in dict_obj.items():
setattr(log_record, key, value)
return log_record
else:
return dict_obj
def _prepare_log_dict(self, record: logging.LogRecord) -> dict:
always_fields = {
"message": record.getMessage(),
"timestamp": dt.datetime.fromtimestamp(
record.created, tz=dt.timezone.utc
).isoformat() if type(record.created) == float else record.created,
}
if record.exc_info is not None:
always_fields["exc_info"] = self.formatException(record.exc_info)
if record.stack_info is not None:
always_fields["stack_info"] = self.formatStack(record.stack_info)
message = {
key: msg_val
if (msg_val := always_fields.pop(val, None)) is not None
else getattr(record, val)
for key, val in self.fmt_keys.items()
}
message.update(always_fields)
# Include all other attributes
for key, val, in record.__dict__.items():
if key not in LOG_RECORD_BUILTIN_ATTRS:
message[key] = val
return message

55
src/mylogger/resources/logging_config.json Executable file
View File

@ -0,0 +1,55 @@
{
"version": 1,
"disable_existing_loggers": false,
"formatters": {
"simple": {
"format": "%(levelname)s - %(message)s",
"datefmt": "%Y-%m-%dT%H:%M:%Sz"
},
"json": {
"()": "mylogger.logger.JSONFormatter",
"fmt_keys": {
"level": "levelname",
"message": "message",
"timestamp": "timestamp",
"logger": "name",
"module": "module",
"function": "funcName",
"line": "lineno",
"thread_name": "threadName"
}
}
},
"handlers": {
"stdout": {
"class": "logging.StreamHandler",
"formatter": "simple",
"stream": "ext://sys.stdout",
"level": "INFO"
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"formatter": "json",
"filename": "room-booking.log.jsonl",
"level": "INFO",
"maxBytes": 10000000,
"backupCount": 6
},
"queue_handler": {
"class": "logging.handlers.QueueHandler",
"handlers": [
"stdout",
"file"
],
"respect_handler_level": true
}
},
"loggers": {
"root": {
"handlers": [
"queue_handler"
],
"level": "DEBUG"
}
}
}