matthias/infomentor
Archived
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

updated main and format
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Matthias Bilger
2019-09-27 21:38:42 +02:00
parent 1ea90db3ad
commit 51e1aab134
7 changed files with 128 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
infomentor/__init__.py
View File

@@ -1 +1 @@
#from infomentor.__main__ import *
# from infomentor.__main__ import *

215
infomentor/__main__.py
View File

@@ -8,13 +8,15 @@ import requests
from infomentor import db, model, connector, informer, config
logformat = "{asctime} - {name:25s}[{filename:20s}:{lineno:3d}] - {levelname:8s} - {message}"
logformat = (
"{asctime} - {name:25s}[{filename:20s}:{lineno:3d}] - {levelname:8s} - {message}"
)
def logtofile():
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler("log.txt", maxBytes=1024*1024, backupCount=10)
handler = RotatingFileHandler("log.txt", maxBytes=1024 * 1024, backupCount=10)
logging.basicConfig(
level=logging.INFO, format=logformat, handlers=[handler], style="{"
)
@@ -29,109 +31,78 @@ def parse_args(arglist):
parser.add_argument(
"--nolog", action="store_true", help="print log instead of logging to file"
)
parser.add_argument("--adduser", action='store_true', help="add user")
parser.add_argument("--addfake", action='store_true', help="add fake")
parser.add_argument("--addpushover", action='store_true', help="add pushover")
parser.add_argument("--addmail", action='store_true', help="add mail")
parser.add_argument("--addcalendar", action='store_true', help="add icloud calendar")
parser.add_argument("--addinvitation", action='store_true', help="add calendar invitation")
parser.add_argument("--test", action="store_true", help="test")
parser.add_argument("--username", type=str, nargs='?', help="username")
parser.add_argument("--username", type=str, nargs="?", help="infomentor username")
parser.add_argument("--password", type=str, nargs="?", help="infomentor password")
parser.add_argument("--fake", action="store_true", help="add fake")
parser.add_argument("--pushover", type=str, nargs="?", help="pushover user id")
parser.add_argument("--mail", type=str, nargs="?", help="e-mail for notification")
parser.add_argument(
"--iclouduser", type=str, nargs="?", help="icloud calendar user"
)
parser.add_argument(
"--icloudpwd", type=str, nargs="?", help="icloud calendar password"
)
parser.add_argument(
"--icloudcalendar", type=str, nargs="?", help="icloud calendar calendar"
)
parser.add_argument(
"--invitationmail", type=str, nargs="?", help="e-mail for notification"
)
args = parser.parse_args(arglist)
return args
def add_user(username):
def perform_user_update(args):
username = args.username
session = db.get_db()
existing_user = (
session.query(model.User).filter(model.User.name == username).one_or_none()
)
if existing_user is not None:
print("user exists, change pw")
print("user exists, changing pw")
else:
print(f"Adding user: {username}")
import getpass
if args.password is None:
import getpass
password = getpass.getpass(prompt="Password: ")
else:
password = args.password
password = getpass.getpass(prompt="Password: ")
if existing_user is not None:
existing_user.password = password
else:
user = model.User(name=username, password=password)
session.add(user)
session.commit()
if args.pushover is not None:
user.notification = model.Notification(
ntype=model.Notification.Types.PUSHOVER, info=args.pushover
)
elif args.mail:
user.notification = model.Notification(
ntype=model.Notification.Types.EMAIL, info=args.mail
)
elif args.fake:
user.notification = model.Notification(
ntype=model.Notification.Types.FAKE, info=""
)
def add_pushover(username):
session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None:
print("user does not exist")
return
else:
print(f"Adding PUSHOVER for user: {username}")
id = input("PUSHOVER ID: ")
user.notification = model.Notification(
ntype=model.Notification.Types.PUSHOVER, info=id
)
session.commit()
if (
args.iclouduser is not None
and args.icloudpwd is not None
and args.icloudcalendar is not None
):
user.icalendar = model.ICloudCalendar(
icloud_user=args.iclouduser,
password=args.icloudpwd,
calendarname=args.icloudcalendar,
)
if args.invitationmail:
user.invitation = model.Invitation(email=mail)
def add_fake(username):
session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None:
print("user does not exist")
return
else:
print(f"Adding FAKE for user: {username}")
user.notification = model.Notification(ntype=model.Notification.Types.FAKE, info="")
session.commit()
def add_calendar(username):
session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None:
print("user does not exist")
return
else:
print(f"Adding icloud calendar for user: {username}")
id = input("Apple ID: ")
import getpass
password = getpass.getpass(prompt="iCloud Password: ")
calendar = input("Calendar: ")
user.icalendar = model.ICloudCalendar(
icloud_user=id, password=password, calendarname=calendar
)
session.commit()
def add_invitation(username):
session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None:
print("user does not exist")
return
else:
print(f"Adding Mail for calendar invitation for user: {username}")
mail = input("Mail: ")
user.invitation = model.Invitation(email=mail)
session.commit()
def add_mail(username):
session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None:
print("user does not exist")
return
else:
print(f"Adding MAIL for user: {username}")
address = input("MAIL ADDRESS: ")
user.notification = model.Notification(
ntype=model.Notification.Types.EMAIL, info=address
)
session.commit()
@@ -140,7 +111,7 @@ def notify_users():
session = db.get_db()
cfg = config.load()
if cfg["healthchecks"]["url"] != "":
requests.get(cfg["healthchecks"]["url"] + '/start')
requests.get(cfg["healthchecks"]["url"] + "/start")
for user in session.query(model.User):
logger.info("==== USER: %s =====", user.name)
@@ -195,8 +166,6 @@ def notify_users():
def main():
args = parse_args(sys.argv[1:])
if args.test:
return
if args.nolog:
logtoconsole()
else:
@@ -206,20 +175,10 @@ def main():
try:
lock = flock.flock()
if not lock.aquire():
logger.info("EXITING - PREVIOUS IS RUNNING")
logger.info("EXITING - PREVIOUS IS STILL RUNNING")
raise Exception()
if args.addfake:
add_fake(args.username)
elif args.adduser:
add_user(args.username)
elif args.addpushover:
add_pushover(args.username)
elif args.addmail:
add_mail(args.username)
elif args.addcalendar:
add_calendar(args.username)
elif args.addinvitation:
add_invitation(args.username)
if args.username:
perform_user_update(args)
else:
notify_users()
except Exception as e:
@@ -228,64 +187,6 @@ def main():
finally:
logger.info("EXITING--------------------- %s", os.getpid())
def run_notify():
run_without_args(notify_users)
def run_adduser():
run_with_args(add_user)
def run_addfake():
run_with_args(add_fake)
def run_addpushover():
run_with_args(add_pushover)
def run_addmail():
run_with_args(add_mail)
def run_addcalendar():
run_with_args(add_calendar)
def run_addinvitation():
run_with_args(add_invitation)
def run_with_args(fct):
args = parse_args(sys.argv[1:])
logtofile()
logger = logging.getLogger("Infomentor Notifier")
logger.info("STARTING-------------------- %s", os.getpid())
lock = flock.flock()
try:
if not lock.aquire():
logger.info("EXITING - PREVIOUS IS RUNNING")
raise Exception()
if args.username is None:
print('Provide Username using --username')
raise Exception('No username provided')
fct(args.username)
except Exception as e:
logger.info("Exceptional exit")
logger.exception("Info")
finally:
logger.info("EXITING--------------------- %s", os.getpid())
def run_without_args(fct):
args = parse_args(sys.argv[1:])
logtofile()
logger = logging.getLogger("Infomentor Notifier")
logger.info("STARTING-------------------- %s", os.getpid())
try:
lock = flock.flock()
if not lock.aquire():
logger.info("EXITING - PREVIOUS IS RUNNING")
raise Exception()
fct()
except Exception as e:
logger.info("Exceptional exit")
logger.exception("Info")
finally:
logger.info("EXITING--------------------- %s", os.getpid())
if __name__ == "__main__":
main()

15
infomentor/config.py
View File

@@ -4,9 +4,7 @@ import os
_config = None
_defaults = {
"pushover": {
"apikey": "",
},
"pushover": {"apikey": ""},
"general": {
"secretkey": "",
"baseurl": "",
@@ -14,16 +12,11 @@ _defaults = {
"im1url": "https://im1.infomentor.de/Germany/Germany/Production",
"mimrul": "https://mein.infomentor.de",
},
"smtp": {
"server": "",
"username": "",
"password": "",
},
"healthchecks": {
"url": "",
},
"smtp": {"server": "", "username": "", "password": ""},
"healthchecks": {"url": ""},
}
def _set_defaults(config):
config = _defaults

8
infomentor/connector.py
View File

@@ -18,11 +18,13 @@ from infomentor import model, config
class InfomentorFile(object):
"""Represent a file which is downloaded"""
def __init__(self, directory, filename, seed=''):
def __init__(self, directory, filename, seed=""):
if directory is None:
raise Exception("directory is required")
self.filename = filename
self.randomid = hashlib.sha1('{}{}'.format(filename, seed).encode('utf-8')).hexdigest()
self.randomid = hashlib.sha1(
"{}{}".format(filename, seed).encode("utf-8")
).hexdigest()
self.directory = directory
@property
@@ -266,7 +268,7 @@ class Infomentor(object):
self.logger.info("fetching news")
self._do_post(self._mim_url("Communication/News/GetNewsList"))
news_json = self.get_json_return()
return news_json['items']
return news_json["items"]
def get_news_article(self, news_entry):
"""Receive all the article information"""

37
infomentor/icloudcalendar.py
View File

@@ -20,7 +20,6 @@ class iCloudConnector(object):
password = None
propfind_principal = '<A:propfind xmlns:A="DAV:"><A:prop><A:current-user-principal/></A:prop></A:propfind>'
propfind_calendar_home_set = "<propfind xmlns='DAV:' xmlns:cd='urn:ietf:params:xml:ns:caldav'><prop> <cd:calendar-home-set/></prop></propfind>"
def __init__(self, username, password, **kwargs):
self.username = username
@@ -30,7 +29,6 @@ class iCloudConnector(object):
self.discover()
self.get_calendars()
# discover: connect to icloud using the provided credentials and discover
#
# 1. The principal URL
@@ -45,10 +43,7 @@ class iCloudConnector(object):
headers = {"Depth": "1"}
auth = HTTPBasicAuth(self.username, self.password)
principal_response = self.repeated_request(
"PROPFIND",
self.icloud_url,
auth=auth,
data=self.propfind_principal
"PROPFIND", self.icloud_url, auth=auth, data=self.propfind_principal
)
# Parse the resulting XML response
soup = BeautifulSoup(principal_response.content, "lxml")
@@ -60,15 +55,20 @@ class iCloudConnector(object):
# Next use the discovery URL to get more detailed properties - such as
# the calendar-home-set
home_set_response = self.repeated_request(
"PROPFIND",
discovery_url,
auth=auth,
data=self.propfind_calendar_home_set,
"PROPFIND", discovery_url, auth=auth, data=self.propfind_calendar_home_set
)
_logger.debug("Result code: {}".format(home_set_response.status_code))
if home_set_response.status_code != 207:
_logger.error("Failed to retrieve calendar-home-set {}".format(home_set_response.status_code))
raise Exception("failed to retrieve calender home set {}".format(home_set_response.content))
_logger.error(
"Failed to retrieve calendar-home-set {}".format(
home_set_response.status_code
)
)
raise Exception(
"failed to retrieve calender home set {}".format(
home_set_response.content
)
)
# And then extract the calendar-home-set URL
soup = BeautifulSoup(home_set_response.content, "lxml")
self.calendar_home_set_url = soup.find(
@@ -77,18 +77,20 @@ class iCloudConnector(object):
def repeated_request(self, *args, **kwargs):
for _ in range(0, 5):
response = requests.request(
*args, **kwargs
)
response = requests.request(*args, **kwargs)
_logger.debug("Request result code: {}".format(response.status_code))
if response.status_code != 207:
_logger.error("Failed to retrieve response: {}".format(response.status_code))
_logger.error(
"Failed to retrieve response: {}".format(response.status_code)
)
_logger.error("Retry")
time.sleep(0.25)
if response.status_code == 207:
break
else:
raise Exception("failed to retrieve {} {}".format(response.content, response.headers))
raise Exception(
"failed to retrieve {} {}".format(response.content, response.headers)
)
return response
# get_calendars
@@ -135,4 +137,3 @@ class iCloudConnector(object):
):
# to do
pass

71
infomentor/informer.py
View File

@@ -56,13 +56,13 @@ class Informer(object):
for news_entry in newslist:
news = (
session.query(model.News)
.filter(model.News.news_id == news_entry['id'])
.filter(model.News.date == news_entry['publishedDate'])
.filter(model.News.news_id == news_entry["id"])
.filter(model.News.date == news_entry["publishedDate"])
.with_parent(self.user, "news")
.one_or_none()
)
if news is not None:
self.logger.debug('Skipping news %s', news_entry['id'])
self.logger.debug("Skipping news %s", news_entry["id"])
continue
news = self.im.get_news_article(news_entry)
self._notify_news(news)
@@ -71,7 +71,7 @@ class Informer(object):
def _notify_news(self, news):
if self.user.notification is None:
self.logger.debug('Warn: no notification for user')
self.logger.debug("Warn: no notification for user")
return
if self.user.notification.ntype == model.Notification.Types.PUSHOVER:
self._notify_news_pushover(news)
@@ -93,7 +93,9 @@ class Informer(object):
for attachment in news.attachments:
fid, fname = attachment.localpath.split("/")
text += """<br>Attachment {0}: {2}/{1} <br>""".format(
fname, urllib.parse.quote(attachment.localpath), cfg["general"]["baseurl"]
fname,
urllib.parse.quote(attachment.localpath),
cfg["general"]["baseurl"],
)
parsed_date = dateparser.parse(news.date)
now = datetime.datetime.now()
@@ -177,7 +179,9 @@ class Informer(object):
for attachment in hw.attachments:
fid, fname = attachment.localpath.split("/")
text += """<br>Attachment {0}: {2}/{1}<br>""".format(
fname, urllib.parse.quote(attachment.localpath), cfg["general"]["baseurl"]
fname,
urllib.parse.quote(attachment.localpath),
cfg["general"]["baseurl"],
)
if len(text) > 900:
url = self._make_site(text)
@@ -240,37 +244,38 @@ class Informer(object):
def _send_invitation(self, calobj, to, fr="infomentor@09a.de"):
event = calobj.subcomponents[0]
eml_body = event['description']
eml_body_bin = event['description']
msg = MIMEMultipart('mixed')
msg['Reply-To']=fr
msg['Subject'] = event['summary']
msg['From'] = fr
msg['To'] = to
eml_body = event["description"]
eml_body_bin = event["description"]
msg = MIMEMultipart("mixed")
msg["Reply-To"] = fr
msg["Subject"] = event["summary"]
msg["From"] = fr
msg["To"] = to
part_email = MIMEText(eml_body,"html")
part_email_text = MIMEText(eml_body,"plain")
part_cal = MIMEText(calobj.to_ical().decode('utf-8'),'calendar;method=REQUEST')
part_email = MIMEText(eml_body, "html")
part_email_text = MIMEText(eml_body, "plain")
part_cal = MIMEText(calobj.to_ical().decode("utf-8"), "calendar;method=REQUEST")
msgAlternative = MIMEMultipart('alternative')
msgAlternative = MIMEMultipart("alternative")
msg.attach(msgAlternative)
ical_atch = MIMEBase('application/ics',' ;name="%s"'%("invite.ics"))
ical_atch.set_payload(calobj.to_ical().decode('utf-8'))
ical_atch = MIMEBase("application/ics", ' ;name="%s"' % ("invite.ics"))
ical_atch.set_payload(calobj.to_ical().decode("utf-8"))
encoders.encode_base64(ical_atch)
ical_atch.add_header('Content-Disposition', 'attachment; filename="%s"'%("invite.ics"))
ical_atch.add_header(
"Content-Disposition", 'attachment; filename="%s"' % ("invite.ics")
)
eml_atch = MIMEBase('text/plain','')
eml_atch.set_payload('')
eml_atch = MIMEBase("text/plain", "")
eml_atch.set_payload("")
encoders.encode_base64(eml_atch)
eml_atch.add_header('Content-Transfer-Encoding', "")
eml_atch.add_header("Content-Transfer-Encoding", "")
msg.attach(part_email)
msg.attach(part_email_text)
msg.attach(part_cal)
self._send_mail(msg)
def _setup_icloudconnector(self):
if self.cal is None:
try:
@@ -284,17 +289,19 @@ class Informer(object):
self.logger.warn("using icloud")
except Exception as e:
self.logger.exception("using icloud dummy connector")
class Dummy(object):
def add_event(self, *args, **kwargs):
pass
self.cal = Dummy()
def _write_icalendar(self, calend):
try:
self._setup_icloudconnector()
self.cal.add_event(calend.to_ical())
except Exception as e:
self.logger.exception('Calendar failed')
self.logger.exception("Calendar failed")
def update_calendar(self):
session = db.get_db()
@@ -304,7 +311,9 @@ class Informer(object):
calentries = self.im.get_calendar()
for entry in calentries:
self.logger.debug(entry)
uid = str(uuid.uuid5(uuid.NAMESPACE_URL, "infomentor_{}".format(entry["id"])))
uid = str(
uuid.uuid5(uuid.NAMESPACE_URL, "infomentor_{}".format(entry["id"]))
)
event_details = self.im.get_event(entry["id"])
calend = Calendar()
event = Event()
@@ -319,13 +328,13 @@ class Informer(object):
event.add("dtend", dateparser.parse(entry["end"]).date())
description = event_details["notes"]
eventinfo = event_details['info']
eventinfo = event_details["info"]
new_cal_entry = calend.to_ical().replace(b"\r", b"")
new_cal_hash = hashlib.sha1(new_cal_entry).hexdigest()
for res in eventinfo['resources']:
for res in eventinfo["resources"]:
f = self.im.download_file(res["url"], directory="files")
description += """\nAttachment {0}: {2}/{1}""".format(
res['title'], urllib.parse.quote(f), cfg["general"]["baseurl"]
res["title"], urllib.parse.quote(f), cfg["general"]["baseurl"]
)
event.add("description", description)
@@ -364,4 +373,4 @@ class Informer(object):
self.user.calendarentries.append(calendarentry)
session.commit()
except Exception as e:
self.logger.exception('Calendar failed')
self.logger.exception("Calendar failed")

5
infomentor/model.py
View File

@@ -252,8 +252,5 @@ class Invitation(ModelBase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __repr__(self):
return "<Invitation(email='%s')>" % (
self.email,
)
return "<Invitation(email='%s')>" % (self.email,)