1 files changed, 151 insertions, 0 deletions
diff --git a/signal_event_notifier.py b/signal_event_notifier.py
new file mode 100755
index 0000000..80282f7
--- /dev/null
+++ b/signal_event_notifier.py
@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+# Copyright 2023 Florian Fischer
+# This program is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free Software
+# Foundation, either version 3 of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program. If not, see <https://www.gnu.org/licenses/>.
+"""Notify the responsible people about occuring events via signal messages"""
+
+import argparse
+import datetime
+import pathlib
+import os
+import sys
+import typing as t
+import subprocess
+
+import icalevents
+from icalevents.icalevents import events
+
+RESPONSIBILITIES_ENV_VAR_NAME = 'SEN_RESPONSIBILITIES'
+
+
+def load_events(uri: str, start: datetime.datetime,
+                end: datetime.datetime) -> list[icalevents.icalparser.Event]:
+    """Retrieve all events from the specified ics calendars"""
+    evs = []
+
+    if uri.startswith("http"):
+        evs += events(url=uri, start=start, end=end)
+    elif pathlib.Path(uri).is_file():
+        evs += events(file=uri, start=start, end=end)
+
+    return evs
+
+
+def load_responsibilities(start: datetime.date,
+                          resp_file_path='') -> list[str]:
+    """Load resposibility definitions
+    
+    If no file path containing responsibilities is passed se tyhe environment.
+    The environment variable 'SEN_RESPONSIBILITIES' is used to pass
+    a line separated list of responsibilities per calendar week."""
+
+    all_resp = ''
+    if resp_file_path and pathlib.Path(resp_file_path).is_file():
+        with open(resp_file_path, 'r', encoding='utf-8') as resp_file:
+            all_resp = resp_file.read()
+
+    all_resp = all_resp or os.environ.get(RESPONSIBILITIES_ENV_VAR_NAME, '')
+    if not all_resp:
+        print(f'{RESPONSIBILITIES_ENV_VAR_NAME} not set', file=sys.stderr)
+        sys.exit(1)
+
+    resps_order = all_resp.splitlines()
+    cal_week = start.isocalendar()[1]
+    if len(resps_order) == 52:
+        resp = resps_order[cal_week - 1]
+    else:
+        resp = resps_order[(cal_week % len(resps_order)) - 1]
+
+    return resp.split(',')
+
+
+def fmt_mentions(responsibilities: list[str]) -> t.Tuple[str, str]:
+    """Return a string of mentions and the required cli arguments for signal-cli"""
+
+    if not responsibilities:
+        return '', ''
+
+    mention_notification_prfx = ''
+    mention_arg = ''
+    for resp in responsibilities:
+        tokens = resp.split(':')
+        if len(tokens) > 1:
+            name = tokens[0]
+            num = tokens[1]
+        else:
+            name = num = tokens[0]
+
+        mention_arg += f' --mention {len(mention_notification_prfx)+1}:{len(name)}:{num}'
+        mention_notification_prfx += f', {name}'
+
+    return mention_notification_prfx[2:], mention_arg
+
+
+def notify(notification: str,
+           signal_user: str,
+           recipients: t.Optional[list[str]] = None,
+           group: t.Optional[str] = None):
+    '''notify a group or a list of recipients about events'''
+
+    if not recipients or group:
+        return
+
+    mentions, mentions_args = fmt_mentions(recipients)
+    if mentions:
+        notification = f'{mentions} {notification}'
+
+    cmd = f'signal-cli -u {signal_user} send --message-from-stdin{mentions_args}'
+    if group:
+        cmd += f' -g {group}'
+    print(cmd)
+    cmd = 'cat -'
+
+    subprocess.run(cmd.split(), input=notification.encode(), check=True)
+
+
+def main():
+    """Programm entry point loading data from webuntis and printing the remaining lessons"""
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--date', help='The date to notify for')
+    parser.add_argument('--calendar', help='The calendars URI')
+    parser.add_argument('--responsibilities-file',
+                        help='A file specifying the responsibilities per line')
+    parser.add_argument('--group',
+                        help='The group notifications should be send to')
+    parser.add_argument('user', help='The signal user sending messages')
+    args = parser.parse_args()
+
+    if args.date:
+        tomorrow = datetime.date.fromisoformat(args.date)
+    else:
+        tomorrow = datetime.date.today() + datetime.timedelta(days=1)
+    end = tomorrow + datetime.timedelta(days=1)
+
+    recipients = load_responsibilities(
+        tomorrow, resp_file_path=args.responsibilities_file)
+
+    # load ICS events
+    evs = load_events(args.calendar, tomorrow, end)
+    if not evs:
+        return
+
+    notification = ''
+    for event in evs:
+        if event.start.date() != tomorrow:
+            continue
+        notification += f' {event.summary}'
+
+    if notification:
+        notify(notification, args.user, recipients, args.group)
+
+
+if __name__ == '__main__':
+    main()