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

Merge branch 'master' into web

Browse Source
This commit is contained in:
Matthias Bilger
2019-05-01 17:49:30 +02:00
parent 6700ad6f14 5136add2d6
commit 4088033695
6 changed files with 268 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
infomentor/__main__.py
View File

@@ -28,8 +28,11 @@ def parse_args(arglist):
parser = argparse.ArgumentParser(description='Infomentor Grabber and Notifier')
parser.add_argument('--nolog', action='store_true', help='print log instead of logging to file')
parser.add_argument('--adduser', type=str, help='add user')
parser.add_argument('--addfake', type=str, help='add fake')
parser.add_argument('--addpushover', type=str, help='add pushover')
parser.add_argument('--addmail', type=str, help='add mail')
parser.add_argument('--addcalendar', type=str, help='add icloud calendar')
parser.add_argument('--test', action='store_true', help='test')
args = parser.parse_args(arglist)
return args
@@ -63,6 +66,32 @@ def add_pushover(username):
user.notification = model.Notification(ntype=model.Notification.Types.PUSHOVER, info=id)
session.commit()
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_mail(username):
session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none()
@@ -95,6 +124,7 @@ def notify_users():
i = informer.Informer(user, im, logger=logger)
i.update_news()
i.update_homework()
i.update_calendar()
statusinfo['ok'] = True
statusinfo['degraded'] = False
except Exception as e:
@@ -125,6 +155,8 @@ def notify_users():
def main():
args = parse_args(sys.argv[1:])
if args.test:
return
if args.nolog:
logtoconsole()
else:
@@ -136,10 +168,14 @@ def main():
if not lock.aquire():
logger.info('EXITING - PREVIOUS IS RUNNING')
raise Exception()
if args.adduser:
if args.addfake:
add_fake(args.addfake)
elif args.adduser:
add_user(args.adduser)
elif args.addpushover:
add_pushover(args.addpushover)
elif args.addcalendar:
add_calendar(args.addcalendar)
else:
notify_users()
except Exception as e:

14
infomentor/connector.py
View File

@@ -13,6 +13,7 @@ import uuid
from infomentor import model
class InfomentorFile(object):
'''Represent a file which is downloaded'''
def __init__(self, directory, filename):
if directory is None:
raise Exception('directory is required')
@@ -306,6 +307,15 @@ class Infomentor(object):
)
return self.get_json_return()
def get_event(self, eventid):
self.logger.info('fetching calendar entry')
data = {'id': eventid}
self._do_post(
self._mim_url('Calendar/Calendar/getEntry'),
data=data
)
return self.get_json_return()
def get_homework(self, offset=0):
self.logger.info('fetching homework')
startofweek = self._get_start_of_week(offset)
@@ -375,7 +385,7 @@ class Infomentor(object):
def _get_week_dates(self, offset=0, weeks=1):
weekoffset = datetime.timedelta(days=7*offset)
startofweek = self._get_start_of_weekdays()
startofweek = self._get_start_of_week()
endofweek = startofweek + datetime.timedelta(days=5+7*(weeks-1))
startofweek += weekoffset
@@ -383,7 +393,7 @@ class Infomentor(object):
now = datetime.datetime.now()
utctime = datetime.datetime.utcnow()
utcoffset = (now.tm_hour - utctime.tm_hour)*60
utcoffset = (now.hour - utctime.hour)*60
data = {
'UTCOffset': utcoffset,

127
infomentor/icloudcalendar.py Normal file
View File

@@ -0,0 +1,127 @@
from datetime import datetime
import sys
from bs4 import BeautifulSoup
import caldav
from caldav.elements import dav, cdav
from lxml import etree
import requests
from requests.auth import HTTPBasicAuth
class iCloudConnector(object):
icloud_url = "https://caldav.icloud.com"
username = None
password = None
propfind_principal = (
u'''<?xml version="1.0" encoding="utf-8"?><propfind xmlns='DAV:'>'''
u'''<prop><current-user-principal/></prop></propfind>'''
)
propfind_calendar_home_set = (
u'''<?xml version="1.0" encoding="utf-8"?><propfind xmlns='DAV:' '''
u'''xmlns:cd='urn:ietf:params:xml:ns:caldav'><prop>'''
u'''<cd:calendar-home-set/></prop></propfind>'''
)
def __init__(self, username, password, **kwargs):
self.username = username
self.password = password
if 'icloud_url' in kwargs:
self.icloud_url = kwargs['icloud_url']
self.discover()
self.get_calendars()
# discover: connect to icloud using the provided credentials and discover
#
# 1. The principal URL
# 2 The calendar home URL
#
# These URL's vary from user to user
# once doscivered, these can then be used to manage calendars
def discover(self):
# Build and dispatch a request to discover the prncipal us for the
# given credentials
headers = {
'Depth': '1',
}
auth = HTTPBasicAuth(self.username, self.password)
principal_response = requests.request(
'PROPFIND',
self.icloud_url,
auth=auth,
headers=headers,
data=self.propfind_principal.encode('utf-8')
)
if principal_response.status_code != 207:
print('Failed to retrieve Principal: ',
principal_response.status_code)
exit(-1)
# Parse the resulting XML response
soup = BeautifulSoup(principal_response.content, 'lxml')
self.principal_path = soup.find(
'current-user-principal'
).find('href').get_text()
discovery_url = self.icloud_url + self.principal_path
# Next use the discovery URL to get more detailed properties - such as
# the calendar-home-set
home_set_response = requests.request(
'PROPFIND',
discovery_url,
auth=auth,
headers=headers,
data=self.propfind_calendar_home_set.encode('utf-8')
)
if home_set_response.status_code != 207:
print('Failed to retrieve calendar-home-set',
home_set_response.status_code)
exit(-1)
# And then extract the calendar-home-set URL
soup = BeautifulSoup(home_set_response.content, 'lxml')
self.calendar_home_set_url = soup.find(
'href',
attrs={'xmlns':'DAV:'}
).get_text()
# get_calendars
# Having discovered the calendar-home-set url
# we can create a local object to control calendars (thin wrapper around
# CALDAV library)
def get_calendars(self):
self.caldav = caldav.DAVClient(self.calendar_home_set_url,
username=self.username,
password=self.password)
self.principal = self.caldav.principal()
self.calendars = self.principal.calendars()
def get_named_calendar(self, name):
if len(self.calendars) > 0:
for calendar in self.calendars:
properties = calendar.get_properties([dav.DisplayName(), ])
display_name = properties['{DAV:}displayname']
if display_name == name:
return calendar
return None
def create_calendar(self,name):
return self.principal.make_calendar(name=name)
def delete_all_events(self,calendar):
for event in calendar.events():
event.delete()
return True
def create_events_from_ical(self, ical):
# to do
pass
def create_simple_timed_event(self,start_datetime, end_datetime, summary,
description):
# to do
pass
def create_simple_dated_event(self,start_datetime, end_datetime, summary,
description):
# to do
pass

47
infomentor/informer.py
View File

@@ -1,4 +1,4 @@
from infomentor import model, db
from infomentor import model, db, icloudcalendar
import logging
import uuid
import os
@@ -7,15 +7,20 @@ import dateparser
import datetime
import math
import pushover
from icalendar import Event, vDate, Calendar
pushover.init('***REMOVED***')
class Informer(object):
'''The Logic part of the infomentor notifier.
This class offers the methods required to notify a user of new News and Homework items posted on infomentor.'''
def __init__(self, user, im, logger):
self.logger = logger or logging.getLogger(__name__)
self.user = user
self.im = im
def send_status_update(self, text):
'''In case something unexpected happends and the user has activated the feature to get notified about it, this will send out the information'''
try:
if self.user.notification.ntype == model.Notification.Types.PUSHOVER:
pushover.Client(self.user.notification.info).send_message(
@@ -52,6 +57,9 @@ class Informer(object):
self._notify_news_pushover(news)
elif self.user.notification.ntype == model.Notification.Types.EMAIL:
self._notify_news_mail(news)
elif self.user.notification.ntype == model.Notification.Types.FAKE:
with open('{}.txt'.format(self.user.name), 'a+') as f:
f.write('Notification:\n---------8<-------\n{}\n---------8<-------\n\n'.format(news.content))
else:
raise Exception('invalid notification')
pass
@@ -146,6 +154,9 @@ class Informer(object):
self._notify_hw_pushover(hw)
elif self.user.notification.ntype == model.Notification.Types.EMAIL:
self._notify_hw_mail(hw)
elif self.user.notification.ntype == model.Notification.Types.FAKE:
with open('{}.txt'.format(self.user.name), 'a+') as f:
f.write('Notification:\n---------8<-------\n{}\n---------8<-------\n\n'.format(hw.text))
else:
raise Exception('invalid notification')
pass
@@ -206,3 +217,37 @@ class Informer(object):
s.login('infomentor@09a.de', '***REMOVED***')
s.send_message(mail)
s.quit()
def update_calendar(self):
session = db.get_db()
print(self.user.icalendar)
if self.user.icalendar is None:
return
icx = icloudcalendar.iCloudConnector(self.user.icalendar.icloud_user, self.user.icalendar.password)
cname = self.user.icalendar.calendarname
cal = icx.get_named_calendar(cname)
if not cal:
cal = icx.create_calendar(cname)
calentries = self.im.get_calendar(weeks=4)
known_entries = {}
for calevent in cal.events():
if calevent.data is None:
continue
uid = re.findall('UID:(.*)', calevent.data)[0]
known_entries[uid] = calevent
for entry in calentries:
self.logger.debug(entry)
event_details = self.im.get_event(entry['id'])
self.logger.debug(event_details)
calend = Calendar()
event = Event()
event.add('uid', 'infomentor_{}'.format(entry['id']))
event.add('summary', entry['title'])
event.add('description', event_details['notes'])
event.add('dtstart', vDate(dateparser.parse(entry['start'])))
event.add('dtend', vDate(dateparser.parse(entry['end'])))
self.logger.debug(event.to_ical())
calend.add_component(event)
cal.add_event(calend.to_ical())

45
infomentor/model.py
View File

@@ -17,6 +17,7 @@ def unpad(s):
return s[0:-s[-1]].decode('utf8')
class User(ModelBase):
'''The infomentor user.'''
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
@@ -24,6 +25,7 @@ class User(ModelBase):
enc_password = Column(String)
notification = relationship("Notification", back_populates="user", uselist=False)
apistatus = relationship("ApiStatus", back_populates="user", uselist=False)
icalendar = relationship("ICloudCalendar", back_populates="user", uselist=False)
wantstatus = Column(Boolean)
homeworks = relationship("Homework", back_populates="user")
news = relationship("News",back_populates="user")
@@ -55,11 +57,14 @@ class User(ModelBase):
class Notification(ModelBase):
'''This contains the information about the type of notification and additional the key to reach out to the user'''
__tablename__ = 'notifications'
class Types(enum.Enum):
'''Supported notification types'''
PUSHOVER = 1
EMAIL = 2
FAKE = 3
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
@@ -73,6 +78,7 @@ class Notification(ModelBase):
class Attachment(ModelBase):
'''General attachment type for homework and news'''
__tablename__ = 'attachments'
id = Column(Integer, primary_key=True)
@@ -89,6 +95,7 @@ class Attachment(ModelBase):
class News(ModelBase):
'''A News entry'''
__tablename__ = 'news'
id = Column(Integer, primary_key=True)
@@ -110,6 +117,7 @@ class News(ModelBase):
self.id, self.title)
class Homework(ModelBase):
'''A homework entry'''
__tablename__ = 'homework'
id = Column(Integer, primary_key=True)
@@ -124,6 +132,7 @@ class Homework(ModelBase):
user = relationship("User", back_populates="homeworks")
class ApiStatus(ModelBase):
'''Representing the result of the last trys to access the api, represented as one status'''
__tablename__ = 'api_status'
id = Column(Integer, primary_key=True)
@@ -142,3 +151,39 @@ class ApiStatus(ModelBase):
return "<ApiStatus(ok='%s', NOKs='%d', info='%s')>" % (
self.ok, self.degraded_count, self.info)
class ICloudCalendar(ModelBase):
'''An icloud account with a calendar name'''
__tablename__ = 'icloud_calendar'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
icloud_user = Column(String)
icloud_pwd = Column(String)
calendarname = Column(String)
user = relationship("User", back_populates="icalendar", uselist=False)
def __init__(self, *args, **kwargs):
self._setup_cipher()
super().__init__(*args, **kwargs)
def _setup_cipher(self):
if not hasattr(self, 'cipher'):
aeskey = hashlib.sha256(_PASSWORD_SECRET_KEY.encode()).digest()
self.cipher = AES.new(aeskey,AES.MODE_ECB)
@property
def password(self):
self._setup_cipher()
decoded = self.cipher.decrypt(base64.b64decode(self.icloud_pwd))
return unpad(decoded)
@password.setter
def password(self, value):
self._setup_cipher()
encoded = base64.b64encode(self.cipher.encrypt(pad(value)))
self.icloud_pwd = encoded
def __repr__(self):
return "<ICloudCalendar(user='%s' cal='%s')>" % (
self.icloud_user, self.calendarname)

2
setup.py
View File

@@ -8,5 +8,5 @@ setup(
author_email = 'matthias@bilger.info',
description = 'grab infomentor news and push or mail them',
packages = find_packages(),
install_requires = ['pycrypto', 'request', 'sqlalchemy', 'dateparser', 'python-pushover', 'flask', 'flask-bootstrap' ],
install_requires = ['pycrypto', 'request', 'sqlalchemy', 'dateparser', 'python-pushover', 'flask', 'flask-bootstrap', 'caldav', 'bs4', 'icalendar' ],
)