from datetime import date, datetime, timedelta from dateutil.relativedelta import relativedelta # BSD class Event: def __init__(self, headline, date, description=''): self.headline = headline.strip() self.date = date.strip() self.description = description.strip() if not headline or not date: raise ValueError("Please enter a headline and date!") def __eq__(self, other): return self.headline == other.headline and self.date == other.date def get_event_msg(self): msg = f'## {self.headline} ##\nDatum: {self.date}\n' if self.description: msg += f'Beschreibung:\n{self.description}\n' return msg def get_csv_line(self): return f"{self.headline};{self.date};{self.description}" # Kommata sind vllt in diesem Modul nicht die beste Idee. Würde auf was anderes ausweichen class ModuleEventReminder: def __init__(self, groups, send, log): self.tag = "%20s - " % "mEventReminder" self.log = log self.send = send self.events = {} self.groups = groups for group in groups: self.load_events_from_csv(group) self.log.debug(f"{self.tag}Modul EventReminder geladen") def load_events_from_csv(self, groupInfo): lines = [] try: with open(groupInfo["EVENTDIR"], 'r', encoding='utf16') as f: lines = f.readlines() except IOError as exc: self.log.error( f"{self.tag}Reading events from csv failed: {exc}") self.events[groupInfo["NAME"]] =[] for line in lines: try: self.events[groupInfo["NAME"]].append(self.interpret_event_line(line)) except (SyntaxError, ValueError) as exc: self.log.warning(f"{self.tag}Error while interpreting event line:{line}; Error: {exc}") def check_for_events(self): today = date.today() dm = today.strftime("%d.%m.%Y") for key, value in self.events.items(): for event in value: if dm == event.date: for group in self.groups: if group["NAME"] == key: self.send(event.get_event_msg(), group["ID"]) break else: # Nächster Tag tomorrow = date.today()+timedelta(days=1) dm = tomorrow.strftime("%d.%m.%Y") if dm == event.date: for group in self.groups: if group["NAME"] == key: self.send(event.get_event_msg(), group["ID"]) break def get_next_event(self, groupInfo): today = datetime.today() format = "%d.%m.%Y" delta = 800 next_event = None for event in self.events[groupInfo["NAME"]]: event_dt = datetime.strptime(event.date, format) delta_temp = (event_dt - today).days if 0 < delta_temp < delta: next_event = event delta = delta_temp message = "Es ist ruhig... zu ruhig!" if next_event: message = f"Nächstes Event:\n{next_event.get_event_msg()}\nNoch {delta} Tage!" return message def get_eventlist(self, groupInfo): message = "Alle Events:\n" for event in self.events[groupInfo["NAME"]]: # Dieses For geht bestimmt in einer Teile, so wie in signalbot.py message += f"{event.headline}, {event.date}\n" return message def interpret_event_line(self, line): splitted = line.strip().split(";") if not (1 < len(splitted) <= 3): # within range[2,3] raise SyntaxError("Invalid event syntax! Needs to be: " "headline ; date(Day.Month.Year) ; optional description.") headline = splitted[0].strip() if len(headline) == 0: raise SyntaxError("No headline given! Needs to be: " "headline; date(Day.Month.Year) ; optional description.") splittedDate = splitted[1].strip().split(".") if len(splittedDate) != 3: raise SyntaxError("Invalid date! Needs to be: Day.Month.Year eg.: 02.09.21") day = splittedDate[0].strip() month = splittedDate[1].strip() year = splittedDate[2].strip() if not (0 < len(day) < 3 and 0 < len(month) < 3 and 1 < len(year) < 5): raise SyntaxError("Invalid date! Needs to be: Day.Month.Year eg.: 02.09.21") if len(day) == 1: day = f"0{day}" if len(month) == 1: month = f"0{month}" if len(year) < 4: year = f"20{year[-2:]}" return Event(headline, f"{day}.{month}.{year}", " ".join(str(x) for x in splitted[2:])) def save_event(self, message, groupInfo): event: Event = None try: message = message.replace("\n"," ") event = self.interpret_event_line(message) if event in self.events[groupInfo["NAME"]]: return "Event is already saved" self.events[groupInfo["NAME"]].append(event) except (SyntaxError, ValueError) as exc: self.log.warning(f"{self.tag}Error while interpreting event line:{message}; Error: {exc}") return f"{exc}" try: with open(groupInfo["EVENTDIR"], 'a', encoding='utf16') as f: f.write(f"{event.get_csv_line()}\n") return "Successfully added event" except IOError as exc: self.log.error(f"{self.tag}Adding new event'{event.get_csv_line()}' to file failed: {exc}") return "Saving event failed"