from icalendar import Calendar, Event import toml, argparse, os, sys, hashlib, json, pytz, glob, os, time from dateutil.relativedelta import relativedelta from dateutil.rrule import rrulestr import datetime as dt import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler import email_alert, xmpp_alert from pprint import pprint import humanfriendly # Parse args parser = argparse.ArgumentParser(description="A simple calendar alerting daemon written in Python") parser.add_argument('--config', type=str, help='Path to config file. Must be .toml') args = parser.parse_args() if not args.config: print("Error: No config file provided.") sys.exit(1) # Exit with error code elif not os.path.isfile(args.config): print("Error: The specified config file does not exist.") sys.exit(1) # Exit with error code else: print("Config file path: ", args.config) # Get config try: with open(args.config, 'r') as f: config = toml.load(f) except Exception as e: print("Error: Failed to parse TOML file.") print(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')) class DateTimeEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, (dt.datetime, dt.timedelta)): return str(o) # Convert the datetime or timedelta object to a string return super().default(o) class FileChangeHandler(FileSystemEventHandler): """ `FileChangeHandler` is a custom event handler for the `watchdog.observers.Observer` class that handles file system events such as file modifications, deletions and creations. It inherits from `watchdog.events.FileSystemEventHandler` providing methods to handle these events: `on_modified`, `on_deleted`, and `on_created`. Each method is overridden for specific functionality when a file system event occurs. The class also includes a method `calculate_event_hash` that generates an MD5 hash for each event dictionary based on its contents. This is used to track changes in the events and determine if they have been modified or deleted. For example, when a file is modified: - It reads the content of the file, parses it into an event dictionary using `calendar_parser`. - Calculates the hash for the event and checks if there's already an existing event with the same UID in `event_list`. - If there is one, it compares the new hash with the old hash. If they differ, it prints that the event has been modified or deleted. Otherwise, it prints that the event hasn't been modified. For file deletion and creation, similar operations are performed but without comparison of hashes. """ def on_modified(self, event): print(f"File modified: {event.src_path}") if not event.is_directory: 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) except Exception as i: print("Failed to parse calendar event at: {}.\n Error:\n{}".format(event.src_path,i)) return # Handle the modified event self.handle_modified(old_event=None, event_dict=event_dict) def on_deleted(self, event): print(f"File deleted: {event.src_path}") if not event.is_directory: uid = os.path.splitext(os.path.basename(event.src_path))[0] # Get the UID from the file path without extension self.handle_modified(old_event=None, event_dict={"uid": uid}, remove=True) def on_created(self, event): print(f"File created: {event.src_path}") if not event.is_directory: 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) except Exception as i: print("Failed to parse calendar event at: {}.\n Error:\n{}".format(event.src_path,i)) return self.handle_modified(old_event=None, event_dict=event_dict) def handle_modified(self, old_event, event_dict, remove=False): if not remove: for i, old_event in enumerate(event_list): if old_event["uid"] == event_dict["uid"]: old_hash = old_event["hash"] new_hash = self.calculate_event_hash(event_dict) if new_hash != old_hash: print("Event with UID {} has been modified or deleted".format(old_event["uid"])) event_list[i] = event_dict else: print("Event with UID {} hasn't been modified".format(old_event["uid"])) break else: event_list.append(event_dict) else: # If remove is True, remove the event from the list for i, old_event in enumerate(event_list): if old_event["uid"] == event_dict["uid"]: print("Event with UID {} has been deleted".format(old_event["uid"])) del event_list[i] break def calculate_event_hash(self, event): return hashlib.md5(json.dumps(event, sort_keys=True, cls=DateTimeEncoder).encode()).hexdigest() def generate_recurring_event_dates(dtstart, rrule): """ Generate recurring event dates based on a start date and an RRULE. This function takes in a start date (`dtstart`) and an RRULE (`rrule`), which is used to generate future dates according to the rules specified by the RRULE. If no count rule is present, it generates a date array from `dtstart` to the current datetime, then adds an arbitrary number of dates into the future (here it's 10). This is done because generating a date array using the rrulestr function without a count rule will return a very large number of elements. The function returns a dictionary containing information about the recurring event: - `recur_dates`: A list of future dates generated by the RRULE. - `infinite_recur`: Boolean indicating whether the recurrence is infinite. - `recur_freq`: The frequency of the recurrence (e.g., 'DAILY'). - `recur_interval`: The interval between each occurrence of the event. - `n_recur_dates_left`: The number of future dates left. Parameters: dtstart (datetime): The start date of the recurring event. rrule (rrule): The RRULE object specifying the recurrence rules. Returns: dict: A dictionary containing information about the recurring event dates, frequency, interval and count. """ recur_info = { "recur_dates": [dtstart], "infinite_recur": False, "recur_freq": None, "recur_interval": None, "n_recur_dates_left": None } if rrule is None: return recur_info rule_str = "RRULE:{}".format(rrule.to_ical().decode('utf-8')) start_date = dtstart infinite_recur = False freq = rrule.get('FREQ')[0] count = rrule.get("COUNT") interval = rrule.get('INTERVAL')[0] current_date = dt.datetime.now().replace(tzinfo=pytz.UTC) if count is None:# or until is not None: delta = None if freq == "DAILY": delta = relativedelta(days=interval) elif freq == "MONTHLY": delta = relativedelta(months=interval) elif freq == "YEARLY": delta = relativedelta(years=interval) count = 0 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) recur_dates = [None] n_recur = None dates = list(ruleset) # Generate future dates according to the rules recur_dates = [i for i in dates if i >= current_date] n_recur = "inf" if infinite_recur is True else len(recur_dates) recur_info["recur_dates"] = recur_dates recur_info["infinite_recur"] = infinite_recur recur_info["recur_freq"] = freq recur_info["recur_interval"] = interval recur_info["n_recur_dates_left"] = n_recur return recur_info def calendar_parser(cal_str): # Parse the calendar cal = Calendar.from_ical(cal_str) # Iterate over each event in the calendar for event in cal.walk('vevent'): event_info = { "uid": None, "dtstart": "", "exdate": [], "summary": None, "description": None, "location": None, "rrule": None } # Catch errors for missing components for info in event_info: try: event_info[info] = event[info] except Exception: pass uid = str(event_info["uid"]) dtstart = event_info["dtstart"].dt exdates = event_info["exdate"] if exdates is not []: if isinstance(exdates, list): exdates = [i.dts.dt.replace(tzinfo=pytz.UTC) for i in exdates] else: exdates = [exdates.dts[0].dt.replace(tzinfo=pytz.UTC)] summary = event["summary"] description = event_info["description"] location = event_info["location"] rrule = event_info["rrule"] # Ensure dates are always as datetime if isinstance(dtstart, dt.datetime): dtstart = dtstart.replace(tzinfo=pytz.UTC) else: dtstart = dt.datetime.combine(dtstart, dt.time.min).replace(tzinfo=pytz.UTC) # Get recurring events if they exist recur_info = generate_recurring_event_dates(dtstart, rrule) event_dates = recur_info["recur_dates"] # Remove exdates event_dates = [i for i in event_dates if i not in exdates] valarms = [] for subcomponent in event.walk("valarm"): valarm = Event.from_ical(subcomponent.to_ical()) timedelta = valarm["trigger"].dt valarms.append(timedelta) event_dict = { "uid": uid, "dtstart": dtstart, "summary": summary, "description": description, "location": location, "event_dates": event_dates, "recur_info": "Recur freq: {}, Recur interval: {}, N dates left: {}".format( recur_info["recur_freq"], recur_info["recur_interval"], recur_info["n_recur_dates_left"] ), "valarms": valarms, "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 return event_dict def get_next_alert(event, current_time): event_dates = event["event_dates"] valarm_deltas = event["valarms"] if event_dates == [] or current_time > event_dates[-1]: return None, None next_event = [i for i in event_dates if i >= current_time][0] next_alert_list = [next_event + i for i in valarm_deltas] 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: with open(file, 'r') as f: cal_str = f.read() event_dict = calendar_parser(cal_str) event_list.append(event_dict) observer = Observer() handler = FileChangeHandler() observer.schedule(handler, cal_dir, recursive=True) observer.start() try: while True: with open("status", 'w') as f: #Refresh the status file f.write("") current_time = dt.datetime.now().replace(tzinfo=pytz.UTC) for event in event_list: next_alert, next_event = get_next_alert(event, current_time) if next_alert == None: continue event_delta = next_alert-current_time total_seconds = event_delta.total_seconds() human_readable_time = humanfriendly.format_timespan(total_seconds) monitor_status = "Current time: {}\nMonitoring: {}\nEvent date: {}\nRecur Dates: {}\nNext alert on: {} in {}\nRecur info: {}\nAlert history: {}\n".format(current_time, event["summary"], next_event, [str(i) for i in event["event_dates"]], next_alert, human_readable_time, 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") process_alert(current_time, next_alert, event) time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()