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

update format using black

Browse Source
This commit is contained in:
Matthias Bilger
2019-05-08 05:06:37 +02:00
parent ad9eb25df0
commit 693711abc1
12 changed files with 488 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
README.md
View File

@@ -50,3 +50,19 @@ python -m infomentor --addcalendar <username>
The login process is a bit scary and mostly hacked. It happens often on the first run, that the login is not ready, the second run then should work without errors. The login process is a bit scary and mostly hacked. It happens often on the first run, that the login is not ready, the second run then should work without errors.
The script shall be run every 10 minutes, that will keep the session alive and minimize errors. The script shall be run every 10 minutes, that will keep the session alive and minimize errors.
TODO:
::
{'id': 1342049, 'title': 'Jade LZK HSU 1. und 2. Klasse', 'time': '10:30 - 11:30', 'notes': '', 'enumType': 'Custom1', 'type': 'cal-custom1', 'info': {'id': 0, 'type': None,
'resources': [{'id': 589680, 'fileType': 'docx', 'title': 'Lernziele HSU das Jahr.docx', 'url': '/Resources/Resource/Download/589680?api=IM2', 'fileTypeName': 'Word processor', 'apiType': 'IM2', 'connectionId': 1342049, 'connectionType':
'Calendar'}]}, 'establishmentName': None, 'date': '18.01.2019', 'isEditable': False, 'isDeletable': False, 'startDate': '2019-01-18', 'startTime': '10:30', 'endDate': '2019-01-18', 'endTime': '11:30', 'allDayEvent': False, 'resourcesNeed
ingConnection': None, 'thirdPartyApiCalendarEventId': None, 'thirdPartyApiCalendarSeriesId': None, 'thirdPartyApiCalendarId': None, 'attendeeIds': None}

159
infomentor/__main__.py
View File

@@ -7,45 +7,50 @@ import os
from infomentor import db, model, connector, informer from infomentor import db, model, connector, informer
logformat='{asctime} - {name:25s} - {levelname:8s} - {message}' logformat = "{asctime} - {name:25s} - {levelname:8s} - {message}"
def logtofile(): def logtofile():
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('log.txt', maxBytes=51200, backupCount=5)
handler = RotatingFileHandler("log.txt", maxBytes=51200, backupCount=5)
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO, format=logformat, handlers=[handler], style="{"
format=logformat,
handlers=[handler],
style='{'
)
def logtoconsole():
logging.basicConfig(
level=logging.DEBUG,
format=logformat,
style='{'
) )
def logtoconsole():
logging.basicConfig(level=logging.DEBUG, format=logformat, style="{")
def parse_args(arglist): def parse_args(arglist):
parser = argparse.ArgumentParser(description='Infomentor Grabber and Notifier') 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(
parser.add_argument('--adduser', type=str, help='add user') "--nolog", action="store_true", help="print log instead of logging to file"
parser.add_argument('--addfake', type=str, help='add fake') )
parser.add_argument('--addpushover', type=str, help='add pushover') parser.add_argument("--adduser", type=str, help="add user")
parser.add_argument('--addmail', type=str, help='add mail') parser.add_argument("--addfake", type=str, help="add fake")
parser.add_argument('--addcalendar', type=str, help='add icloud calendar') parser.add_argument("--addpushover", type=str, help="add pushover")
parser.add_argument('--test', action='store_true', help='test') 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) args = parser.parse_args(arglist)
return args return args
def add_user(username): def add_user(username):
session = db.get_db() session = db.get_db()
existing_user = session.query(model.User).filter(model.User.name == username).one_or_none() existing_user = (
session.query(model.User).filter(model.User.name == username).one_or_none()
)
if existing_user is not None: if existing_user is not None:
print('user exists, change pw') print("user exists, change pw")
else: else:
print(f'Adding user: {username}') print(f"Adding user: {username}")
import getpass import getpass
password = getpass.getpass(prompt='Password: ')
password = getpass.getpass(prompt="Password: ")
if existing_user is not None: if existing_user is not None:
existing_user.password = password existing_user.password = password
else: else:
@@ -58,98 +63,111 @@ def add_pushover(username):
session = db.get_db() session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none() user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None: if user is None:
print('user does not exist') print("user does not exist")
return return
else: else:
print(f'Adding PUSHOVER for user: {username}') print(f"Adding PUSHOVER for user: {username}")
id = input('PUSHOVER ID: ') id = input("PUSHOVER ID: ")
user.notification = model.Notification(ntype=model.Notification.Types.PUSHOVER, info=id) user.notification = model.Notification(
ntype=model.Notification.Types.PUSHOVER, info=id
)
session.commit() session.commit()
def add_fake(username): def add_fake(username):
session = db.get_db() session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none() user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None: if user is None:
print('user does not exist') print("user does not exist")
return return
else: else:
print(f'Adding FAKE for user: {username}') print(f"Adding FAKE for user: {username}")
user.notification = model.Notification(ntype=model.Notification.Types.FAKE, info='') user.notification = model.Notification(ntype=model.Notification.Types.FAKE, info="")
session.commit() session.commit()
def add_calendar(username): def add_calendar(username):
session = db.get_db() session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none() user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None: if user is None:
print('user does not exist') print("user does not exist")
return return
else: else:
print(f'Adding icloud calendar for user: {username}') print(f"Adding icloud calendar for user: {username}")
id = input('Apple ID: ') id = input("Apple ID: ")
import getpass import getpass
password = getpass.getpass(prompt='iCloud Password: ')
calendar = input('Calendar: ') password = getpass.getpass(prompt="iCloud Password: ")
user.icalendar = model.ICloudCalendar(icloud_user=id, password=password, calendarname=calendar) calendar = input("Calendar: ")
user.icalendar = model.ICloudCalendar(
icloud_user=id, password=password, calendarname=calendar
)
session.commit() session.commit()
def add_mail(username): def add_mail(username):
session = db.get_db() session = db.get_db()
user = session.query(model.User).filter(model.User.name == username).one_or_none() user = session.query(model.User).filter(model.User.name == username).one_or_none()
if user is None: if user is None:
print('user does not exist') print("user does not exist")
return return
else: else:
print(f'Adding MAIL for user: {username}') print(f"Adding MAIL for user: {username}")
address = input('MAIL ADDRESS: ') address = input("MAIL ADDRESS: ")
user.notification = model.Notification(ntype=model.Notification.Types.EMAIL, info=address) user.notification = model.Notification(
ntype=model.Notification.Types.EMAIL, info=address
)
session.commit() session.commit()
def notify_users(): def notify_users():
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
session = db.get_db() session = db.get_db()
for user in session.query(model.User): for user in session.query(model.User):
logger.info('==== USER: %s =====', user.name) logger.info("==== USER: %s =====", user.name)
if user.password == '': if user.password == "":
logger.warning('User %s not enabled', user.name) logger.warning("User %s not enabled", user.name)
continue continue
now = datetime.datetime.now() now = datetime.datetime.now()
im = connector.Infomentor(user.name) im = connector.Infomentor(user.name)
im.login(user.password) im.login(user.password)
logger.info('User loggedin') logger.info("User loggedin")
statusinfo = {'datetime': now, 'ok': False, 'info': '', 'degraded_count':0} statusinfo = {"datetime": now, "ok": False, "info": "", "degraded_count": 0}
if user.apistatus is None: if user.apistatus is None:
user.apistatus = model.ApiStatus(**statusinfo) user.apistatus = model.ApiStatus(**statusinfo)
logger.info('Former API status: %s', user.apistatus) logger.info("Former API status: %s", user.apistatus)
try: try:
i = informer.Informer(user, im, logger=logger) i = informer.Informer(user, im, logger=logger)
i.update_news() i.update_news()
i.update_homework() i.update_homework()
i.update_calendar() i.update_calendar()
statusinfo['ok'] = True statusinfo["ok"] = True
statusinfo['degraded'] = False statusinfo["degraded"] = False
except Exception as e: except Exception as e:
inforstr = 'Exception occured:\n{}:{}\n'.format(type(e).__name__, e) inforstr = "Exception occured:\n{}:{}\n".format(type(e).__name__, e)
statusinfo['ok'] = False statusinfo["ok"] = False
statusinfo['info'] = inforstr statusinfo["info"] = inforstr
logger.exception("Something went wrong") logger.exception("Something went wrong")
finally: finally:
if user.apistatus.ok == True and statusinfo['ok'] == False: if user.apistatus.ok == True and statusinfo["ok"] == False:
logger.error('Switching to degraded state %s', user.name) logger.error("Switching to degraded state %s", user.name)
statusinfo['degraded_count'] = 1 statusinfo["degraded_count"] = 1
if user.apistatus.ok == False and statusinfo['ok'] == False: if user.apistatus.ok == False and statusinfo["ok"] == False:
if user.apistatus.degraded_count == 1 and user.wantstatus: if user.apistatus.degraded_count == 1 and user.wantstatus:
im.send_status_update(statusinfo['info']) im.send_status_update(statusinfo["info"])
try: try:
statusinfo['degraded_count'] = user.apistatus['degraded_count'] + 1 statusinfo["degraded_count"] = user.apistatus["degraded_count"] + 1
except Exception as e: except Exception as e:
statusinfo['degraded_count'] = 1 statusinfo["degraded_count"] = 1
if user.apistatus.ok == False and statusinfo['ok'] == True: if user.apistatus.ok == False and statusinfo["ok"] == True:
statusinfo['info'] = 'Works as expected, failed {} times'.format(user.apistatus.degraded_count) statusinfo["info"] = "Works as expected, failed {} times".format(
statusinfo['degraded_count'] = 0 user.apistatus.degraded_count
)
statusinfo["degraded_count"] = 0
if user.wantstatus: if user.wantstatus:
im.send_status_update(statusinfo['info']) im.send_status_update(statusinfo["info"])
user.apistatus.updateobj(statusinfo) user.apistatus.updateobj(statusinfo)
logger.info('New API status: %s', user.apistatus) logger.info("New API status: %s", user.apistatus)
session.commit() session.commit()
@@ -161,12 +179,12 @@ def main():
logtoconsole() logtoconsole()
else: else:
logtofile() logtofile()
logger = logging.getLogger('Infomentor Notifier') logger = logging.getLogger("Infomentor Notifier")
logger.info('STARTING-------------------- %s', os.getpid()) logger.info("STARTING-------------------- %s", os.getpid())
try: try:
lock = flock.flock() lock = flock.flock()
if not lock.aquire(): if not lock.aquire():
logger.info('EXITING - PREVIOUS IS RUNNING') logger.info("EXITING - PREVIOUS IS RUNNING")
raise Exception() raise Exception()
if args.addfake: if args.addfake:
add_fake(args.addfake) add_fake(args.addfake)
@@ -181,10 +199,11 @@ def main():
else: else:
notify_users() notify_users()
except Exception as e: except Exception as e:
logger.info('Exceptional exit') logger.info("Exceptional exit")
logger.exception('Info') logger.exception("Info")
finally: finally:
logger.info('EXITING--------------------- %s', os.getpid()) logger.info("EXITING--------------------- %s", os.getpid())
if __name__ == "__main__": if __name__ == "__main__":
main() main()

36
infomentor/config.py
View File

@@ -5,21 +5,22 @@ _config = None
def _set_defaults(config): def _set_defaults(config):
config.add_section('pushover') config.add_section("pushover")
config.add_section('general') config.add_section("general")
config.add_section('smtp') config.add_section("smtp")
config['pushover']['apikey'] = '' config["pushover"]["apikey"] = ""
config['general']['secretkey'] = '' config["general"]["secretkey"] = ""
config['general']['baseurl'] = '' config["general"]["baseurl"] = ""
config['general']['adminmail'] = '' config["general"]["adminmail"] = ""
config['general']['im1url'] = 'https://im1.infomentor.de/Germany/Germany/Production' config["general"]["im1url"] = "https://im1.infomentor.de/Germany/Germany/Production"
config['general']['mimurl'] = 'https://mein.infomentor.de' config["general"]["mimurl"] = "https://mein.infomentor.de"
config['smtp']['server'] = '' config["smtp"]["server"] = ""
config['smtp']['username'] = '' config["smtp"]["username"] = ""
config['smtp']['password'] = '' config["smtp"]["password"] = ""
def load(cfg_file='infomentor.ini'):
'''Load the config from the file''' def load(cfg_file="infomentor.ini"):
"""Load the config from the file"""
global _config global _config
if _config is None: if _config is None:
_config = configparser.ConfigParser() _config = configparser.ConfigParser()
@@ -29,8 +30,9 @@ def load(cfg_file='infomentor.ini'):
_config.read(cfg_file) _config.read(cfg_file)
return _config return _config
def save(cfg_file='infomentor.ini'):
'''Write config to file''' def save(cfg_file="infomentor.ini"):
"""Write config to file"""
global _config global _config
with open(cfg_file, 'w+') as f: with open(cfg_file, "w+") as f:
_config.write(f) _config.write(f)

361
infomentor/connector.py
View File

File diff suppressed because it is too large Load Diff

8
infomentor/db.py
View File

@@ -4,14 +4,14 @@ from sqlalchemy.orm import sessionmaker
_session = None _session = None
def get_db(filename='infomentor.db'):
'''Get the database session for infomentor''' def get_db(filename="infomentor.db"):
"""Get the database session for infomentor"""
global _session global _session
if _session is None: if _session is None:
engine = create_engine(f'sqlite:///{filename}') engine = create_engine(f"sqlite:///{filename}")
model.ModelBase.metadata.create_all(engine) model.ModelBase.metadata.create_all(engine)
model.ModelBase.metadata.bind = engine model.ModelBase.metadata.bind = engine
DBSession = sessionmaker(bind=engine) DBSession = sessionmaker(bind=engine)
_session = DBSession() _session = DBSession()
return _session return _session

29
infomentor/flock.py
View File

@@ -1,44 +1,46 @@
import os import os
class flock(object): class flock(object):
'''A simple filelocking mechanism to prevent execution at the same time''' """A simple filelocking mechanism to prevent execution at the same time"""
filename = '.im.lock'
filename = ".im.lock"
def __init__(self): def __init__(self):
'''Creates an object with the current pid''' """Creates an object with the current pid"""
self.pid = os.getpid() self.pid = os.getpid()
def aquire(self): def aquire(self):
'''Try to get the lock, if it fails it returns False''' """Try to get the lock, if it fails it returns False"""
if self.is_locked(): if self.is_locked():
return False return False
with open(self.filename, 'w+') as f: with open(self.filename, "w+") as f:
f.write('{}'.format(self.pid)) f.write("{}".format(self.pid))
return True return True
def release(self): def release(self):
'''Release the lock''' """Release the lock"""
if self.own_lock(): if self.own_lock():
os.unlink(self.filename) os.unlink(self.filename)
def __del__(self): def __del__(self):
'''Release on delete''' """Release on delete"""
self.release() self.release()
def own_lock(self): def own_lock(self):
'''Check if the lock is assigned to the current pid''' """Check if the lock is assigned to the current pid"""
lockinfo = self._get_lockinfo() lockinfo = self._get_lockinfo()
return lockinfo == self.pid return lockinfo == self.pid
def is_locked(self): def is_locked(self):
'''Check if it is currently locked''' """Check if it is currently locked"""
lockinfo = self._get_lockinfo() lockinfo = self._get_lockinfo()
if not lockinfo: if not lockinfo:
return False return False
return self._is_process_active(lockinfo) return self._is_process_active(lockinfo)
def _is_process_active(self, pid): def _is_process_active(self, pid):
'''Check if the processed having the lock is still running''' """Check if the processed having the lock is still running"""
try: try:
os.kill(pid, 0) os.kill(pid, 0)
return pid != self.pid return pid != self.pid
@@ -46,12 +48,11 @@ class flock(object):
return False return False
def _get_lockinfo(self): def _get_lockinfo(self):
'''Retrieve the information about the lock''' """Retrieve the information about the lock"""
try: try:
lock = {} lock = {}
with open(self.filename, 'r') as f: with open(self.filename, "r") as f:
pid = int(f.read().strip()) pid = int(f.read().strip())
return pid return pid
except Exception as e: except Exception as e:
return False return False

70
infomentor/icloudcalendar.py
View File

@@ -8,26 +8,27 @@ from lxml import etree
import requests import requests
from requests.auth import HTTPBasicAuth from requests.auth import HTTPBasicAuth
class iCloudConnector(object): class iCloudConnector(object):
icloud_url = "https://caldav.icloud.com" icloud_url = "https://caldav.icloud.com"
username = None username = None
password = None password = None
propfind_principal = ( propfind_principal = (
u'''<?xml version="1.0" encoding="utf-8"?><propfind xmlns='DAV:'>''' u"""<?xml version="1.0" encoding="utf-8"?><propfind xmlns='DAV:'>"""
u'''<prop><current-user-principal/></prop></propfind>''' u"""<prop><current-user-principal/></prop></propfind>"""
) )
propfind_calendar_home_set = ( propfind_calendar_home_set = (
u'''<?xml version="1.0" encoding="utf-8"?><propfind xmlns='DAV:' ''' u"""<?xml version="1.0" encoding="utf-8"?><propfind xmlns='DAV:' """
u'''xmlns:cd='urn:ietf:params:xml:ns:caldav'><prop>''' u"""xmlns:cd='urn:ietf:params:xml:ns:caldav'><prop>"""
u'''<cd:calendar-home-set/></prop></propfind>''' u"""<cd:calendar-home-set/></prop></propfind>"""
) )
def __init__(self, username, password, **kwargs): def __init__(self, username, password, **kwargs):
self.username = username self.username = username
self.password = password self.password = password
if 'icloud_url' in kwargs: if "icloud_url" in kwargs:
self.icloud_url = kwargs['icloud_url'] self.icloud_url = kwargs["icloud_url"]
self.discover() self.discover()
self.get_calendars() self.get_calendars()
@@ -42,45 +43,40 @@ class iCloudConnector(object):
def discover(self): def discover(self):
# Build and dispatch a request to discover the prncipal us for the # Build and dispatch a request to discover the prncipal us for the
# given credentials # given credentials
headers = { headers = {"Depth": "1"}
'Depth': '1',
}
auth = HTTPBasicAuth(self.username, self.password) auth = HTTPBasicAuth(self.username, self.password)
principal_response = requests.request( principal_response = requests.request(
'PROPFIND', "PROPFIND",
self.icloud_url, self.icloud_url,
auth=auth, auth=auth,
headers=headers, headers=headers,
data=self.propfind_principal.encode('utf-8') data=self.propfind_principal.encode("utf-8"),
) )
if principal_response.status_code != 207: if principal_response.status_code != 207:
print('Failed to retrieve Principal: ', print("Failed to retrieve Principal: ", principal_response.status_code)
principal_response.status_code)
exit(-1) exit(-1)
# Parse the resulting XML response # Parse the resulting XML response
soup = BeautifulSoup(principal_response.content, 'lxml') soup = BeautifulSoup(principal_response.content, "lxml")
self.principal_path = soup.find( self.principal_path = (
'current-user-principal' soup.find("current-user-principal").find("href").get_text()
).find('href').get_text() )
discovery_url = self.icloud_url + self.principal_path discovery_url = self.icloud_url + self.principal_path
# Next use the discovery URL to get more detailed properties - such as # Next use the discovery URL to get more detailed properties - such as
# the calendar-home-set # the calendar-home-set
home_set_response = requests.request( home_set_response = requests.request(
'PROPFIND', "PROPFIND",
discovery_url, discovery_url,
auth=auth, auth=auth,
headers=headers, headers=headers,
data=self.propfind_calendar_home_set.encode('utf-8') data=self.propfind_calendar_home_set.encode("utf-8"),
) )
if home_set_response.status_code != 207: if home_set_response.status_code != 207:
print('Failed to retrieve calendar-home-set', print("Failed to retrieve calendar-home-set", home_set_response.status_code)
home_set_response.status_code)
exit(-1) exit(-1)
# And then extract the calendar-home-set URL # And then extract the calendar-home-set URL
soup = BeautifulSoup(home_set_response.content, 'lxml') soup = BeautifulSoup(home_set_response.content, "lxml")
self.calendar_home_set_url = soup.find( self.calendar_home_set_url = soup.find(
'href', "href", attrs={"xmlns": "DAV:"}
attrs={'xmlns':'DAV:'}
).get_text() ).get_text()
# get_calendars # get_calendars
@@ -88,9 +84,9 @@ class iCloudConnector(object):
# we can create a local object to control calendars (thin wrapper around # we can create a local object to control calendars (thin wrapper around
# CALDAV library) # CALDAV library)
def get_calendars(self): def get_calendars(self):
self.caldav = caldav.DAVClient(self.calendar_home_set_url, self.caldav = caldav.DAVClient(
username=self.username, self.calendar_home_set_url, username=self.username, password=self.password
password=self.password) )
self.principal = self.caldav.principal() self.principal = self.caldav.principal()
self.calendars = self.principal.calendars() self.calendars = self.principal.calendars()
@@ -98,16 +94,16 @@ class iCloudConnector(object):
if len(self.calendars) > 0: if len(self.calendars) > 0:
for calendar in self.calendars: for calendar in self.calendars:
properties = calendar.get_properties([dav.DisplayName(), ]) properties = calendar.get_properties([dav.DisplayName()])
display_name = properties['{DAV:}displayname'] display_name = properties["{DAV:}displayname"]
if display_name == name: if display_name == name:
return calendar return calendar
return None return None
def create_calendar(self,name): def create_calendar(self, name):
return self.principal.make_calendar(name=name) return self.principal.make_calendar(name=name)
def delete_all_events(self,calendar): def delete_all_events(self, calendar):
for event in calendar.events(): for event in calendar.events():
event.delete() event.delete()
return True return True
@@ -116,12 +112,14 @@ class iCloudConnector(object):
# to do # to do
pass pass
def create_simple_timed_event(self,start_datetime, end_datetime, summary, def create_simple_timed_event(
description): self, start_datetime, end_datetime, summary, description
):
# to do # to do
pass pass
def create_simple_dated_event(self,start_datetime, end_datetime, summary, def create_simple_dated_event(
description): self, start_datetime, end_datetime, summary, description
):
# to do # to do
pass pass

18
infomentor/informer.py
View File

@@ -266,12 +266,17 @@ class Informer(object):
new_cal_hash = hashlib.sha1(new_cal_entry).hexdigest() new_cal_hash = hashlib.sha1(new_cal_entry).hexdigest()
session = db.get_db() session = db.get_db()
storedata = { storedata = {
'calendar_id': uid, "calendar_id": uid,
'ical': new_cal_entry, "ical": new_cal_entry,
'hash': new_cal_hash "hash": new_cal_hash,
} }
calendarentry = session.query(model.CalendarEntry).filter(model.CalendarEntry.calendar_id == uid) .with_parent(self.user, "calendarentries").one_or_none() calendarentry = (
if calendarentry is not None : session.query(model.CalendarEntry)
.filter(model.CalendarEntry.calendar_id == uid)
.with_parent(self.user, "calendarentries")
.one_or_none()
)
if calendarentry is not None:
if calendarentry.hash == new_cal_hash: if calendarentry.hash == new_cal_hash:
self.logger.info("no change for calendar entry {}".format(uid)) self.logger.info("no change for calendar entry {}".format(uid))
continue continue
@@ -286,6 +291,5 @@ class Informer(object):
self.user.calendarentries.append(calendarentry) self.user.calendarentries.append(calendarentry)
session.commit() session.commit()
self.logger.debug(new_cal_entry.decode('utf-8')) self.logger.debug(new_cal_entry.decode("utf-8"))
cal.add_event(calend.to_ical()) cal.add_event(calend.to_ical())

117
infomentor/model.py
View File

@@ -11,17 +11,23 @@ cfg = config.load()
ModelBase = declarative_base() ModelBase = declarative_base()
_PASSWORD_SECRET_KEY = cfg['general']['secretkey'] _PASSWORD_SECRET_KEY = cfg["general"]["secretkey"]
BS = 16 BS = 16
def pad(s): def pad(s):
diff = BS - len(s) % BS diff = BS - len(s) % BS
return (s + (diff) * chr(diff)).encode('utf8') return (s + (diff) * chr(diff)).encode("utf8")
def unpad(s): def unpad(s):
return s[0:-s[-1]].decode('utf8') return s[0 : -s[-1]].decode("utf8")
class User(ModelBase): class User(ModelBase):
'''The infomentor user.''' """The infomentor user."""
__tablename__ = 'users'
__tablename__ = "users"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
name = Column(String) name = Column(String)
@@ -31,7 +37,7 @@ class User(ModelBase):
icalendar = relationship("ICloudCalendar", back_populates="user", uselist=False) icalendar = relationship("ICloudCalendar", back_populates="user", uselist=False)
wantstatus = Column(Boolean) wantstatus = Column(Boolean)
homeworks = relationship("Homework", back_populates="user") homeworks = relationship("Homework", back_populates="user")
news = relationship("News",back_populates="user") news = relationship("News", back_populates="user")
calendarentries = relationship("CalendarEntry", back_populates="user", uselist=True) calendarentries = relationship("CalendarEntry", back_populates="user", uselist=True)
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
@@ -39,9 +45,9 @@ class User(ModelBase):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
def _setup_cipher(self): def _setup_cipher(self):
if not hasattr(self, 'cipher'): if not hasattr(self, "cipher"):
aeskey = hashlib.sha256(_PASSWORD_SECRET_KEY.encode()).digest() aeskey = hashlib.sha256(_PASSWORD_SECRET_KEY.encode()).digest()
self.cipher = AES.new(aeskey,AES.MODE_ECB) self.cipher = AES.new(aeskey, AES.MODE_ECB)
@property @property
def password(self): def password(self):
@@ -57,33 +63,37 @@ class User(ModelBase):
def __repr__(self): def __repr__(self):
return "<User(name='%s', password='%s')>" % ( return "<User(name='%s', password='%s')>" % (
self.name, '*' * len(self.password)) self.name,
"*" * len(self.password),
)
class Notification(ModelBase): class Notification(ModelBase):
'''This contains the information about the type of notification and additional the key to reach out to the user''' """This contains the information about the type of notification and additional the key to reach out to the user"""
__tablename__ = 'notifications'
__tablename__ = "notifications"
class Types(enum.Enum): class Types(enum.Enum):
'''Supported notification types''' """Supported notification types"""
PUSHOVER = 1 PUSHOVER = 1
EMAIL = 2 EMAIL = 2
FAKE = 3 FAKE = 3
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id')) user_id = Column(Integer, ForeignKey("users.id"))
ntype = Column(Enum(Types)) ntype = Column(Enum(Types))
info = Column(String) info = Column(String)
user = relationship("User", back_populates="notification") user = relationship("User", back_populates="notification")
def __repr__(self): def __repr__(self):
return "<Notification(type='{}', info='{}')>".format( return "<Notification(type='{}', info='{}')>".format(self.ntype, self.info)
self.ntype, self.info)
class Attachment(ModelBase): class Attachment(ModelBase):
'''General attachment type for homework and news''' """General attachment type for homework and news"""
__tablename__ = 'attachments'
__tablename__ = "attachments"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
attachment_id = Column(Integer) attachment_id = Column(Integer)
@@ -91,20 +101,21 @@ class Attachment(ModelBase):
url = Column(String) url = Column(String)
title = Column(String) title = Column(String)
localpath = Column(String) localpath = Column(String)
news_id = Column(Integer, ForeignKey('news.id')) news_id = Column(Integer, ForeignKey("news.id"))
homework_id = Column(Integer, ForeignKey('homework.id')) homework_id = Column(Integer, ForeignKey("homework.id"))
news = relationship("News", back_populates="attachments") news = relationship("News", back_populates="attachments")
homework = relationship("Homework", back_populates="attachments") homework = relationship("Homework", back_populates="attachments")
class News(ModelBase): class News(ModelBase):
'''A News entry''' """A News entry"""
__tablename__ = 'news'
__tablename__ = "news"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
news_id = Column(Integer) news_id = Column(Integer)
user_id = Column(Integer, ForeignKey('users.id')) user_id = Column(Integer, ForeignKey("users.id"))
title = Column(String) title = Column(String)
content = Column(String) content = Column(String)
category = Column(String) category = Column(String)
@@ -113,50 +124,62 @@ class News(ModelBase):
imagefile = Column(String) imagefile = Column(String)
notified = Column(Boolean, default=False) notified = Column(Boolean, default=False)
raw = Column(String) raw = Column(String)
attachments = relationship("Attachment", order_by=Attachment.id, back_populates="news", uselist=True) attachments = relationship(
"Attachment", order_by=Attachment.id, back_populates="news", uselist=True
)
user = relationship("User", back_populates="news") user = relationship("User", back_populates="news")
def __repr__(self): def __repr__(self):
return "<News(id='%d', title='%s')>" % ( return "<News(id='%d', title='%s')>" % (self.id, self.title)
self.id, self.title)
class CalendarEntry(ModelBase): class CalendarEntry(ModelBase):
'''A News entry''' """A News entry"""
__tablename__ = 'calendarentries'
__tablename__ = "calendarentries"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
calendar_id = Column(Integer) calendar_id = Column(Integer)
title = Column(String) title = Column(String)
user_id = Column(Integer, ForeignKey('users.id')) user_id = Column(Integer, ForeignKey("users.id"))
ical = Column(String) ical = Column(String)
hash = Column(String) hash = Column(String)
user = relationship("User", back_populates="calendarentries") user = relationship("User", back_populates="calendarentries")
def __repr__(self): def __repr__(self):
return "<CalendarEntry(id='%d', title='%s', hash='%s')>" % ( return "<CalendarEntry(id='%d', title='%s', hash='%s')>" % (
self.id, self.title, hash) self.id,
self.title,
hash,
)
class Homework(ModelBase): class Homework(ModelBase):
'''A homework entry''' """A homework entry"""
__tablename__ = 'homework'
__tablename__ = "homework"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
homework_id = Column(Integer) homework_id = Column(Integer)
user_id = Column(Integer, ForeignKey('users.id')) user_id = Column(Integer, ForeignKey("users.id"))
subject = Column(String) subject = Column(String)
courseElement = Column(String) courseElement = Column(String)
text = Column(String) text = Column(String)
date = Column(String) date = Column(String)
imageUrl = Column(String) imageUrl = Column(String)
attachments = relationship("Attachment", order_by=Attachment.id, back_populates="homework") attachments = relationship(
"Attachment", order_by=Attachment.id, back_populates="homework"
)
user = relationship("User", back_populates="homeworks") user = relationship("User", back_populates="homeworks")
class ApiStatus(ModelBase): class ApiStatus(ModelBase):
'''Representing the result of the last trys to access the api, represented as one status''' """Representing the result of the last trys to access the api, represented as one status"""
__tablename__ = 'api_status'
__tablename__ = "api_status"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id')) user_id = Column(Integer, ForeignKey("users.id"))
degraded_count = Column(Integer) degraded_count = Column(Integer)
datetime = Column(DateTime) datetime = Column(DateTime)
info = Column(String) info = Column(String)
@@ -169,14 +192,19 @@ class ApiStatus(ModelBase):
def __repr__(self): def __repr__(self):
return "<ApiStatus(ok='%s', NOKs='%d', info='%s')>" % ( return "<ApiStatus(ok='%s', NOKs='%d', info='%s')>" % (
self.ok, self.degraded_count, self.info) self.ok,
self.degraded_count,
self.info,
)
class ICloudCalendar(ModelBase): class ICloudCalendar(ModelBase):
'''An icloud account with a calendar name''' """An icloud account with a calendar name"""
__tablename__ = 'icloud_calendar'
__tablename__ = "icloud_calendar"
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id')) user_id = Column(Integer, ForeignKey("users.id"))
icloud_user = Column(String) icloud_user = Column(String)
icloud_pwd = Column(String) icloud_pwd = Column(String)
calendarname = Column(String) calendarname = Column(String)
@@ -187,9 +215,9 @@ class ICloudCalendar(ModelBase):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
def _setup_cipher(self): def _setup_cipher(self):
if not hasattr(self, 'cipher'): if not hasattr(self, "cipher"):
aeskey = hashlib.sha256(_PASSWORD_SECRET_KEY.encode()).digest() aeskey = hashlib.sha256(_PASSWORD_SECRET_KEY.encode()).digest()
self.cipher = AES.new(aeskey,AES.MODE_ECB) self.cipher = AES.new(aeskey, AES.MODE_ECB)
@property @property
def password(self): def password(self):
@@ -205,5 +233,6 @@ class ICloudCalendar(ModelBase):
def __repr__(self): def __repr__(self):
return "<ICloudCalendar(user='%s' cal='%s')>" % ( return "<ICloudCalendar(user='%s' cal='%s')>" % (
self.icloud_user, self.calendarname) self.icloud_user,
self.calendarname,
)

45
infomentor/web.py
View File

@@ -5,34 +5,47 @@ from flask_bootstrap import Bootstrap
app = Flask(__name__) app = Flask(__name__)
Bootstrap(app) Bootstrap(app)
@app.route('/')
@app.route("/")
def home(): def home():
return render_template('notfound.html') return render_template("notfound.html")
@app.route('/addlogin')
@app.route("/addlogin")
def extra(): def extra():
return render_template('addlogin.html') return render_template("addlogin.html")
@app.route('/create', methods=['POST'])
@app.route("/create", methods=["POST"])
def create(): def create():
if request.form['accesscode'] != 'fhKjzgV/BXWq4YRxUPO4qYlHWCDf': if request.form["accesscode"] != "fhKjzgV/BXWq4YRxUPO4qYlHWCDf":
return redirect(url_for('home')) return redirect(url_for("home"))
session = db.get_db() session = db.get_db()
username = request.form['username'] username = request.form["username"]
existing_user = session.query(model.User).filter(model.User.name == username).one_or_none() existing_user = (
session.query(model.User).filter(model.User.name == username).one_or_none()
)
if existing_user is not None: if existing_user is not None:
return redirect(url_for('home')) return redirect(url_for("home"))
password = request.form['password'] password = request.form["password"]
user = model.User(name=username, password=password) user = model.User(name=username, password=password)
if request.form['notify'] == 'mail': if request.form["notify"] == "mail":
user.notification = [model.Notification(ntype=model.Notification.Types.EMAIL, info=request.form['info'])] user.notification = [
model.Notification(
ntype=model.Notification.Types.EMAIL, info=request.form["info"]
)
]
else: else:
user.notification = [model.Notification(ntype=model.Notification.Types.PUSHOVER, info=request.form['info'])] user.notification = [
model.Notification(
ntype=model.Notification.Types.PUSHOVER, info=request.form["info"]
)
]
session.add(user) session.add(user)
session.commit() session.commit()
return "success" return "success"
if __name__ == '__main__':
app.run(debug=True)
if __name__ == "__main__":
app.run(debug=True)

27
setup.py
View File

@@ -1,12 +1,23 @@
from setuptools import setup, find_packages from setuptools import setup, find_packages
setup( setup(
name = 'infomentor', name="infomentor",
version = '1.0.0', version="1.0.0",
url = 'https://github.com/mypackage.git', url="https://github.com/mypackage.git",
author = 'Matthias Bilger', author="Matthias Bilger",
author_email = 'matthias@bilger.info', author_email="matthias@bilger.info",
description = 'grab infomentor news and push or mail them', description="grab infomentor news and push or mail them",
packages = find_packages(), packages=find_packages(),
install_requires = ['pycrypto', 'request', 'sqlalchemy', 'dateparser', 'python-pushover', 'flask', 'flask-bootstrap', 'caldav', 'bs4', 'icalendar' ], install_requires=[
"pycrypto",
"request",
"sqlalchemy",
"dateparser",
"python-pushover",
"flask",
"flask-bootstrap",
"caldav",
"bs4",
"icalendar",
],
) )