From 92e1578bbc37b2209b2fabc842cc0746c20ad4ca Mon Sep 17 00:00:00 2001 From: Sam Date: Fri, 9 Feb 2024 12:08:31 +0000 Subject: [PATCH] Replace toml with humanfriendly library for email alert and update send\_email function with event details * Update import statement in email_alert.py to include humanfriendly library * Modify send_email function signature in email_alert.py to accept an event dictionary instead of separate arguments * Extract event name, description, location, date, and time until event from the event dictionary and use them in the email body * Update email body format string accordingly * Use humanfriendly library to format time difference between next alert and next event into a human-readable format. --- email_alert.py | 34 +++++++++++++------- remindme_caldav.py | 80 ++++++++++++++++++++++++++++++---------------- xmpp_alert.py | 45 ++++++++++++++++++-------- 3 files changed, 107 insertions(+), 52 deletions(-) diff --git a/email_alert.py b/email_alert.py index 45b7565..a2398e1 100644 --- a/email_alert.py +++ b/email_alert.py @@ -2,12 +2,9 @@ import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import ssl -import toml +import humanfriendly -def send_email(event_summary, next_alert, next_event, config): - with open('config.toml', 'r') as f: - config = toml.load(f) - +def send_email(event, next_alert, next_event, config): # Set up the SMTP server details smtp_server = config["email"]["smtp_server"] port = config["email"]["port"] @@ -16,9 +13,13 @@ def send_email(event_summary, next_alert, next_event, config): password = config["email"]["password"] # Event details - event_name = event_summary + event_name = event["summary"] + event_description = event["description"] + event_location = event["location"] event_date = next_event event_delta = next_event - next_alert + total_seconds = event_delta.total_seconds() + human_readable_time = humanfriendly.format_timespan(total_seconds) # Create a multipart message and set headers message = MIMEMultipart() @@ -29,9 +30,19 @@ def send_email(event_summary, next_alert, next_event, config): # Add body to email body = """\ Hi, - This is an alert for the event named '{}' which will occur on {}. - This event will start in '{}' - """.format(event_name, event_date, event_delta) + This is an event alert from remindme_caldav. + + Event details: + --------------------------------- + + Event name: {} + Date/time: {} + Description: {} + Location: {} + Time until event: {} + + --------------------------------- + """.format(event_name, event_date, event_description, event_location, human_readable_time) message.attach(MIMEText(body, "plain")) text = message.as_string() @@ -45,7 +56,8 @@ def send_email(event_summary, next_alert, next_event, config): # Send email here server.sendmail(sender_email, receiver_email, text) + print("Message sent via email") + return except Exception as e: # Print any error messages to stdout - print(e) - return print("Message sent via email") + print("An error occured when sending alert via email, please check your config. Message: {}".format(e)) diff --git a/remindme_caldav.py b/remindme_caldav.py index dc44c23..a23a1db 100644 --- a/remindme_caldav.py +++ b/remindme_caldav.py @@ -6,8 +6,7 @@ import datetime as dt import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler -import email_alert -#import xmpp_alert +import email_alert, xmpp_alert # Parse args parser = argparse.ArgumentParser(description="A simple calendar alerting daemon written in Python") @@ -33,6 +32,10 @@ except Exception as e: sys.exit(1) # Exit with error code cal_dir = config["app"]["calendar_dir"] +# Check if the path is a directory +if not os.path.isdir(cal_dir): + print("The provided path to .ics files does not exist: '{}'".format(cal_dir)) + exit(1) # Get all .ics files from your directory files = glob.glob(os.path.join(cal_dir, '*.ics')) @@ -47,9 +50,12 @@ class FileChangeHandler(FileSystemEventHandler): def on_modified(self, event): print(f"File modified: {event.src_path}") if not event.is_directory: # If it's a file and not a directory - print(str(dt.datetime.now()), "Sync detected, updating events") - with open(event.src_path, 'r') as f: - cal_str = f.read() + try: + with open(event.src_path, 'r') as f: + cal_str = f.read() + except Exception as e: + print("Not a valid file: {}. Error: {}".format(event.src_path, e)) + return # Parse the calendar try: @@ -95,8 +101,12 @@ class FileChangeHandler(FileSystemEventHandler): if not event.is_directory: # If it's a file and not a directory print(str(dt.datetime.now()), "Sync detected, updating events") - with open(event.src_path, 'r') as f: - cal_str = f.read() + try: + with open(event.src_path, 'r') as f: + cal_str = f.read() + except Exception as e: + print("Not a valid file: {}. Error: {}".format(event.src_path, e)) + return try: event_dict = calendar_parser(cal_str) @@ -118,10 +128,13 @@ class FileChangeHandler(FileSystemEventHandler): def calculate_recur_dates(dtstart, vrecur): rule_str = "RRULE:{}".format(vrecur.to_ical().decode('utf-8')) start_date = dtstart + infinite_recur = False + freq = vrecur.get('FREQ')[0] + interval = vrecur.get('INTERVAL')[0] + current_date = dt.datetime.now().replace(tzinfo=pytz.UTC) + if vrecur.get("COUNT") is None: # If no COUNT, calculate an end date based on FREQ and INTERVAL to prevent generating too many dates - freq = vrecur.get('FREQ')[0] - interval = vrecur.get('INTERVAL')[0] delta = None @@ -133,19 +146,20 @@ def calculate_recur_dates(dtstart, vrecur): delta = relativedelta(years=interval) count = 0 - current_date = dt.datetime.now().replace(tzinfo=pytz.UTC) origin_date = start_date while origin_date < current_date: count += interval origin_date += delta*interval rule_str += ";COUNT={}".format(count+10) + infinite_recur = True ruleset = rrulestr(rule_str, dtstart=start_date) # Generate future dates according to the rules dates = list(ruleset) - return [d for d in dates if d > start_date] + n_recur = len([d for d in dates if d > current_date]) + return [d for d in dates if d > start_date], infinite_recur, freq, interval, n_recur def calendar_parser(cal_str): # Parse the calendar @@ -158,10 +172,17 @@ def calendar_parser(cal_str): dtstart = dtstart if isinstance(dtstart, dt.datetime) else dt.datetime.combine(dtstart, dt.time.min) # Ensure dates are always as datetime dtstart = dtstart.replace(tzinfo=pytz.UTC) summary = component.get("SUMMARY") + description = component.get("DESCRIPTION") + location = component.get("LOCATION") vrecur = component.get("RRULE") recur_dates = [None] + recur_info=None if vrecur is not None: - recur_dates = calculate_recur_dates(dtstart, vrecur) + recur_dates, infinite_recur, freq, interval, n_recur = calculate_recur_dates(dtstart, vrecur) + if infinite_recur: + recur_info = "Number of recurs: {}, Interval: {}, Freq: {}".format(str(n_recur)+"++", interval, freq) + else: + recur_info = "Number of recurs: {}, Interval: {}, Freq: {}".format(str(n_recur), interval, freq) valarm_list = [] # List to hold all VALARM for this event for subcomponent in component.walk(): # Find all VALARMs for this VEVENT @@ -169,7 +190,7 @@ def calendar_parser(cal_str): valarm = Event.from_ical(subcomponent.to_ical()) timedelta = valarm.get("TRIGGER").dt valarm_list.append(timedelta) # Add this VALARM to the list - event_dict = {"uid": str(uid), "dtstart": dtstart, "summary": summary, "recur_dates": recur_dates, "valarm": valarm_list, "alert_history": []} + event_dict = {"uid": str(uid), "dtstart": dtstart, "summary": summary, "description": description, "location": location, "recur_dates": recur_dates, "recur_info": recur_info, "valarm": valarm_list, "alert_history": []} handler = FileChangeHandler() # Create an instance of the FileChangeHandler class new_hash = handler.calculate_event_hash(event_dict) # Calculate the hash of the event dictionary event_dict["hash"] = new_hash # Store the hash in the event dictionary @@ -195,6 +216,23 @@ def get_next_alert(event, current_time): next_alert = min(next_alert_list) return next_alert - dt.timedelta(seconds=5), next_event +def process_alert(current_time, next_alert, event): + if current_time >= next_alert and next_alert < next_alert + dt.timedelta(seconds=15): + if len(event["alert_history"]) == 0: + print("First alert for '{}' detected".format(event["summary"])) + event["alert_history"] = [{"timestamp_alert_triggered": current_time, "alert_defintition_time": next_alert}] + elif next_alert in [i["alert_defintition_time"] for i in event["alert_history"]]: + return # continue is not needed here as it's the end of function + else: + print("Posting alert for {}!".format(event["summary"])) + event["alert_history"].append({"timestamp_alert_triggered": current_time, "alert_defintition_time": next_alert}) + #xmpp_alert.send_xmpp(event["summary"], next_alert, next_event, args.config) + email_alert.send_email(event, next_alert, next_event, config) + xmpp_alert.send_xmpp(event, next_alert, next_event, config) + with open("alert_history", 'a') as f: + f.write(str(event)) # write expects a str not dict + return + # Create initial event_list using calendar_parser event_list = [] # List to hold dictionaries for each event for file in files: @@ -217,24 +255,12 @@ try: next_alert, next_event = get_next_alert(event, current_time) if next_alert == None: continue - monitor_status = "Current time: {}\nMonitoring: {}\nEvent date: {}\nNext alert on: {}\nAlert history: {}\n".format(current_time, event["summary"], next_event, next_alert, event["alert_history"]) + monitor_status = "Current time: {}\nMonitoring: {}\nEvent date: {}\nRecur Dates: {}\nNext alert on: {}\nRecur info: {}\nAlert history: {}\n".format(current_time, event["summary"], next_event, [str(i) for i in event["recur_dates"]], next_alert, event["recur_info"], event["alert_history"]) with open("status", 'a') as f: # Write the output to the file f.write(monitor_status) f.write("\n") - if current_time >= next_alert and next_alert < next_alert + dt.timedelta(seconds=15): - if len(event["alert_history"]) == 0: - print("First alert for '{}' detected".format(event["summary"])) - event["alert_history"] = [{"timestamp_alert_triggered": current_time, "alert_defintition_time": next_alert}] - elif next_alert in [i["alert_defintition_time"] for i in event["alert_history"]]: - continue - else: - print("Posting alert for {}!".format(event["summary"])) - event["alert_history"].append({"timestamp_alert_triggered": current_time, "alert_defintition_time": next_alert}) - #xmpp_alert.send_xmpp(event["summary"], next_alert, next_event, args.config) - email_alert.send_email(event["summary"], next_alert, next_event, args.config) - with open("alert_history", 'a') as f: - f.write(event) + process_alert(current_time, next_alert, event) time.sleep(1) except KeyboardInterrupt: observer.stop() diff --git a/xmpp_alert.py b/xmpp_alert.py index 962084f..65f1b32 100644 --- a/xmpp_alert.py +++ b/xmpp_alert.py @@ -1,5 +1,5 @@ import slixmpp -import toml +import humanfriendly class SendMsgBot(slixmpp.ClientXMPP): def __init__(self, jid, password, recipient, message): @@ -14,9 +14,14 @@ class SendMsgBot(slixmpp.ClientXMPP): self.send_message(mto=self.recipient, mbody=self.msg, mtype='chat') self.disconnect() -def send_xmpp(event_summary, next_alert, next_event, config): - with open('config.toml', 'r') as f: - config = toml.load(f) +def send_xmpp(event, next_alert, next_event, config): + event_name = event["summary"] + event_description = event["description"] + event_location = event["location"] + event_date = next_event + event_delta = next_event - next_alert + total_seconds = event_delta.total_seconds() + human_readable_time = humanfriendly.format_timespan(total_seconds) jid = config["xmpp"]["jid"] password = config["xmpp"]["password"] # replace with your password @@ -24,13 +29,25 @@ def send_xmpp(event_summary, next_alert, next_event, config): message = """\ Hi, - This is an alert for the event named '{}' which will occur on {}. - This event will start in '{}' - """.format(event_summary, next_event, next_alert) - - bot = SendMsgBot(jid, password, recipient, message) - bot.register_plugin('xep_0030') # Service Discovery - bot.register_plugin('xep_0199') # XMPP Ping - bot.connect() - bot.process(forever=False) - return print("Message sent via XMPP") + This is an event alert from remindme_caldav. + + Event details: + --------------------------------- + + Event name: {} + Date/time: {} + Description: {} + Location: {} + Time until event: {} + + --------------------------------- + """.format(event_name, event_date, event_description, event_location, human_readable_time) + try: + bot = SendMsgBot(jid, password, recipient, message) + bot.register_plugin('xep_0030') # Service Discovery + bot.register_plugin('xep_0199') # XMPP Ping + bot.connect() + bot.process(forever=False) + return print("Message sent via XMPP") + except Exception as e: + print("An error occured when sending alert via XMPP, please check your config. Message: {}".format(e))