sam
/
remindme_caldav
2
2
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor code to combine email and XMPP alerting into single class 

- Create a new module `alert_processor.py` for handling both email and XMPP alerts.
- Remove the email_alert.py and xmpp_alert.py files.
This commit is contained in:
mrsu 2024-02-12 02:19:49 +00:00
parent ea34a0ef32
commit 2c8151870c
4 changed files with 135 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
alert_processor.py Normal file
View File

@ -0,0 +1,118 @@
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import ssl
import humanfriendly
import slixmpp
class AlertProcessor:
def __init__(self, config):
self.config = config
def _create_email(self, event, next_alert, next_event):
# Set up the SMTP server details
smtp_server = self.config["email"]["smtp_server"]
port = self.config["email"]["port"]
sender_email = self.config["email"]["username"]
receiver_email = self.config["email"]["recipient"]
password = self.config["email"]["password"]
# Event details
event_name, event_description, event_location, event_date, human_readable_time = self._get_event_details(event, next_alert, next_event)
# Create a multipart message and set headers
message = MIMEMultipart()
message["From"] = sender_email
message["To"] = receiver_email
message["Subject"] = "Event Alert: '{}' @ {}".format(event_name, event_date)
# Add body to email
body = self._create_email_body(event_name, event_date, event_description, event_location, human_readable_time)
message.attach(MIMEText(body, "plain"))
text = message.as_string()
return smtp_server, port, sender_email, receiver_email, password, text
def _get_event_details(self, event, next_alert, next_event):
# Event details
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)
return event_name, event_description, event_location, event_date, human_readable_time
def _create_email_body(self, event_name, event_date, event_description, event_location, human_readable_time):
body = """\
Hi,
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)
return body
def send_email(self, event, next_alert, next_event):
try:
smtp_server, port, sender_email, receiver_email, password, text = self._create_email(event, next_alert, next_event)
# Create a secure SSL context and connect to the server
context = ssl.create_default_context()
with smtplib.SMTP(smtp_server, port) as server: # Use 'with' statement for automatic cleanup
server.ehlo() # Can be omitted
server.starttls(context=context) # Secure the connection
server.login(sender_email, password)
# Send email here
server.sendmail(sender_email, receiver_email, text)
return print("Message sent via email")
except Exception as e:
# Print any error messages to stdout
print("An error occured when sending alert via email, please check your config. Message: {}".format(e))
def _create_xmpp_bot(self, event, next_alert, next_event):
class SendMsgBot(slixmpp.ClientXMPP):
def __init__(self, jid, password, recipient, message):
super().__init__(jid, password)
self.recipient = recipient
self.msg = message
self.add_event_handler("session_start", self.start)
def start(self, event):
self.send_presence()
self.get_roster()
self.send_message(mto=self.recipient, mbody=self.msg, mtype='chat')
self.disconnect()
jid = self.config["xmpp"]["jid"]
password = self.config["xmpp"]["password"] # replace with your password
recipient = self.config["xmpp"]["recipient"]
event_name, event_description, event_location, event_date, human_readable_time = self._get_event_details(event, next_alert, next_event)
message = self._create_email_body(event_name, event_date, event_description, event_location, human_readable_time)
return SendMsgBot(jid, password, recipient, message)
def send_xmpp(self, event, next_alert, next_event):
try:
bot = self._create_xmpp_bot(event, next_alert, next_event)
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))

62
email_alert.py
View File

@ -1,62 +0,0 @@
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import ssl
import humanfriendly
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"]
sender_email = config["email"]["username"]
receiver_email = config["email"]["recipient"]
password = config["email"]["password"]
# Event details
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()
message["From"] = sender_email
message["To"] = receiver_email
message["Subject"] = "Event Alert: '{}' @ {}".format(event_name, event_date)
# Add body to email
body = """\
Hi,
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()
try:
# Create a secure SSL context and connect to the server
context = ssl.create_default_context()
with smtplib.SMTP(smtp_server, port) as server: # Use 'with' statement for automatic cleanup
server.ehlo() # Can be omitted
server.starttls(context=context) # Secure the connection
server.login(sender_email, password)
# Send email here
server.sendmail(sender_email, receiver_email, text)
return print("Message sent via email")
except Exception as e:
# Print any error messages to stdout
print("An error occured when sending alert via email, please check your config. Message: {}".format(e))

30
remindme_caldav.py
View File

@ -7,13 +7,10 @@ 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
from pathlib import Path
import argparse
import textwrap
import logging
import argparse, textwrap, logging
from alert_processor import AlertProcessor
def setup_logger(loglevel):
"""Setup basic logging."""
@ -148,7 +145,7 @@ class RecurringEventGenerator:
"infinite_recur": False,
"recur_freq": None,
"recur_interval": None,
"n_recur_dates_left": None
"n_recur_dates_left": 1
}
def generate(self):
@ -161,9 +158,10 @@ class RecurringEventGenerator:
freq = self.rrule.get('FREQ')[0]
count = self.rrule.get("COUNT")
interval = self.rrule.get('INTERVAL')[0]
until = self.rrule.get('UNTIL')
current_date = dt.datetime.now().replace(tzinfo=pytz.UTC)
if count is None:
if count is None or until is not None:
delta = None
if freq == "DAILY":
@ -211,7 +209,6 @@ class CalendarParser:
generator = RecurringEventGenerator(dtstart, event_dict["rrule"])
recur_info = generator.generate()
event_dates = self.remove_exdates(event_dict["exdate"], recur_info["recur_dates"])
valarms = self.process_valarm(event)
event_dict = {
@ -290,15 +287,21 @@ def get_next_alert(event, current_time):
if event_dates == [] or event_dates is None 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)
next_alert_list = [next_event + i for i in valarm_deltas if next_event + i >= current_time]
if len(next_alert_list) == 0:
next_alert = next_event
else:
next_alert = min(next_alert_list)
return next_alert - dt.timedelta(seconds=5), next_event
def process_alert(current_time, next_alert, next_event, event, config):
"""
This function processes a given alert and passes it to a messaging client.
"""
if current_time >= next_alert and next_alert < next_alert + dt.timedelta(seconds=15):
if current_time >= next_alert and current_time < next_alert + dt.timedelta(seconds=15):
print(current_time, next_alert, next_alert + dt.timedelta(seconds=15))
print(True if current_time >= next_alert else False)
print(True if next_alert < next_alert + dt.timedelta(seconds=15) else False)
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}]
@ -308,8 +311,9 @@ def process_alert(current_time, next_alert, next_event, event, config):
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)
processor = AlertProcessor(config)
processor.send_email(event, next_alert, next_event)
#processor.send_xmpp(event, next_alert, next_event)
with open("alert_history", 'a') as f:
f.write(str(event)) # write expects a str not dict
return

53
xmpp_alert.py
View File

@ -1,53 +0,0 @@
import slixmpp
import humanfriendly
class SendMsgBot(slixmpp.ClientXMPP):
def __init__(self, jid, password, recipient, message):
super().__init__(jid, password)
self.recipient = recipient
self.msg = message
self.add_event_handler("session_start", self.start)
def start(self, event):
self.send_presence()
self.get_roster()
self.send_message(mto=self.recipient, mbody=self.msg, mtype='chat')
self.disconnect()
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
recipient = config["xmpp"]["recipient"]
message = """\
Hi,
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))