1 Commits

Author SHA1 Message Date
8be441bab9 init turbo 2024-04-05 11:22:14 +02:00
15 changed files with 255 additions and 613 deletions

View File

@ -1,6 +1,3 @@
import time
import asyncio
import icalendar import icalendar
import datetime import datetime
import recurring_ical_events import recurring_ical_events
@ -13,7 +10,7 @@ months = ["Januar", "Februar", "März", "April", "Mai", "Juni", "Juli", "August"
"November", "Dezember"] "November", "Dezember"]
async def getWeek(weekstart: datetime, file: str, showsat: bool): def getWeek(weekstart: datetime, file: str, showsat: bool):
""" """
Liefert alle Events einer Woche zurück. \n Liefert alle Events einer Woche zurück. \n
Wochenstart wird automatisch auf den Montag der Woche gelegt. \n Wochenstart wird automatisch auf den Montag der Woche gelegt. \n
@ -70,10 +67,10 @@ async def getWeek(weekstart: datetime, file: str, showsat: bool):
"day": estart.day "day": estart.day
} }
eventl += [eventdict] eventl += [eventdict]
return eventl, await daylist(start_date, showsat), prevw, nextw, mon return eventl, daylist(start_date, showsat), prevw, nextw, mon
async def daylist(weekstart: datetime, showsat: bool): def daylist(weekstart: datetime, showsat: bool):
""" """
Gibt die Essen einer Woche zurück. Gibt die Essen einer Woche zurück.
:param weekstart: :param weekstart:
@ -86,19 +83,15 @@ async def daylist(weekstart: datetime, showsat: bool):
r = 6 r = 6
else: else:
r = 5 r = 5
essen = []
for i in range(r): for i in range(r):
essen += [getMeals(weekday)] essen = getMeals(weekday)
dayl += [{ dayl += [{
"day": weekday.day, "day": weekday.day,
"short": shortnames[i], "short": shortnames[i],
"long": longnames[i], "long": longnames[i],
"mensa": i "mensa": essen
}] }]
weekday += datetime.timedelta(days=1) weekday += datetime.timedelta(days=1)
essenl = await asyncio.gather(*essen, return_exceptions=True)
for day in range(r):
dayl[day]["mensa"] = essenl[day]
return dayl return dayl

View File

@ -1,2 +1,6 @@
{ {
"TINF22B3": [
"raplaTINF22B3.ical",
"https://rapla.dhbw-karlsruhe.de/rapla?page=ical&user=vollmer&file=tinf22b3"
]
} }

View File

@ -1,10 +1,9 @@
import requests
import urllib.parse import urllib.parse
import time import time
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from flask import redirect, url_for from flask import redirect, url_for
from init import Dualis from init import Dualis
import asyncio
import httpx
headers = { headers = {
'Cookie': 'cnsc=0', 'Cookie': 'cnsc=0',
@ -17,32 +16,31 @@ headers = {
url = "https://dualis.dhbw.de/scripts/mgrqispi.dll" url = "https://dualis.dhbw.de/scripts/mgrqispi.dll"
async def checkUser(email: str, password: str): def checkUser(email: str, password: str):
""" """
Erhält von Dualis den Token und Cookie für User. Erhält von Dualis den Token und Cookie für User.
:param email: :param email:
:param password: :param password:
:return (Token, Session): :return (Token, Session):
""" """
async with httpx.AsyncClient() as s: s = requests.Session()
fpw = urllib.parse.quote(password, safe='', encoding=None, errors=None) fpw = urllib.parse.quote(password, safe='', encoding=None, errors=None)
fmail = urllib.parse.quote(email, safe='', encoding=None, errors=None) fmail = urllib.parse.quote(email, safe='', encoding=None, errors=None)
content = (f'usrname={fmail}&pass={fpw}&ARGUMENTS=clino%2Cusrname%2Cpass%2Cmenuno%2Cmenu_type%2Cbrowser' payload = 'usrname=' + fmail + '&pass=' + fpw + ('&ARGUMENTS=clino%2Cusrname%2Cpass%2Cmenuno%2Cmenu_type%2Cbrowser'
f'%2Cplatform&APPNAME=CampusNet&PRGNAME=LOGINCHECK') '%2Cplatform&APPNAME=CampusNet&PRGNAME=LOGINCHECK')
response = await s.post(url=url, headers=headers, content=content) response = s.post(url, headers=headers, data=payload)
header = response.headers header = response.headers
try: try:
refresh = header["REFRESH"] refresh = header["REFRESH"]
arg = refresh.find("=-N") + 3 arg = refresh.find("=-N") + 3
komma = refresh[arg:].find(",") komma = refresh[arg:].find(",")
cookie = s.cookies.get("cnsc") except KeyError:
except KeyError: return -2, s
return -2, s token = refresh[arg:komma + arg]
token = refresh[arg:komma + arg] return token, s
return token, cookie
async def getKurs(token: int, cookie: str): def getKurs(token: int, cookie: str):
""" """
Bestimmt aus der ersten Prüfung den Kursbezeichner des Users. Bestimmt aus der ersten Prüfung den Kursbezeichner des Users.
TODO: Umstellen auf Bezeichner INKL. Standort TODO: Umstellen auf Bezeichner INKL. Standort
@ -53,34 +51,34 @@ async def getKurs(token: int, cookie: str):
try: try:
headers["Cookie"] = "cnsc=" + cookie headers["Cookie"] = "cnsc=" + cookie
token = str(token) token = str(token)
async with httpx.AsyncClient as s: response = requests.request("GET", url +
response = await s.get(url=f"{url}?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N{token},-N000307,", "?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token + ",-N000307,",
headers=headers) headers=headers, data={})
html = BeautifulSoup(response.text, 'lxml') html = BeautifulSoup(response.text, 'lxml')
link = html.body.find('a', attrs={'id': "Popup_details0001"})['href'] link = html.body.find('a', attrs={'id': "Popup_details0001"})['href']
response = await s.get(url=f"{url}{link[21:]}", headers=headers) response = requests.request("GET", url + link[21:], headers=headers, data={})
html = BeautifulSoup(response.text, 'lxml') html = BeautifulSoup(response.text, 'lxml')
content = html.body.find('td', attrs={'class': 'level02'}).text content = html.body.find('td', attrs={'class': 'level02'}).text
start = content.find(" ") + 4 start = content.find(" ") + 4
end = start + (content[start:].find(" ")) end = start + (content[start:].find(" "))
kurs = content[start:end] kurs = content[start:end]
except AttributeError: except AttributeError:
kurs = 0 kurs = 0
return kurs return kurs
async def logOut(token: int, cookie: str): def logOut(token: int, cookie: str):
""" """
Invalidiert Token und Cookie bei Dualis. Invalidiert Token und Cookie bei Dualis.
:param token: :param token:
:param cookie: :param cookie:
""" """
headers["Cookie"] = "cnsc=" + cookie headers["Cookie"] = "cnsc=" + cookie
async with httpx.AsyncClient() as s: requests.request("GET", url + "?APPNAME=CampusNet&PRGNAME=LOGOUT&ARGUMENTS=-N" + str(token)
await s.get(url=f"{url}?APPNAME=CampusNet&PRGNAME=LOGOUT&ARGUMENTS=-N{token}, -N001", headers=headers) + ", -N001", headers=headers, data={})
async def getSem(token: int, cookie: str): def getSem(token: int, cookie: str):
""" """
Liefert die Liste aller auf Dualis verfügbaren Semester. Liefert die Liste aller auf Dualis verfügbaren Semester.
:param token: :param token:
@ -89,21 +87,21 @@ async def getSem(token: int, cookie: str):
""" """
headers["Cookie"] = "cnsc=" + cookie headers["Cookie"] = "cnsc=" + cookie
token = str(token) token = str(token)
async with httpx.AsyncClient() as s: response = requests.request("GET", url +
response = await s.get(url=f"{url}?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N{token},-N000307,", "?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token + ",-N000307,",
headers=headers) headers=headers, data={})
html = BeautifulSoup(response.text, 'lxml') html = BeautifulSoup(response.text, 'lxml')
select = html.find('select') select = html.find('select')
select = select.find_all(value=True) select = select.find_all(value=True)
optlist = [] optlist = []
for i in select: for i in select:
t = i.text.replace("Wi", "Winter").replace("So", "Sommer") t = i.text.replace("Wi", "Winter").replace("So", "Sommer")
t = t.replace("Se", "semester") t = t.replace("Se", "semester")
optlist += [[t, i['value']]] optlist += [[t, i['value']]]
return optlist return optlist
async def getResults(token, cookie: str, resl: str): def getResults(token, cookie: str, resl: str):
""" """
Liefert die Liste aller Prüfungsergebnisse eines Semesters. Liefert die Liste aller Prüfungsergebnisse eines Semesters.
:param token: :param token:
@ -112,43 +110,31 @@ async def getResults(token, cookie: str, resl: str):
:return [[Name, Note, Credits], ...]: :return [[Name, Note, Credits], ...]:
""" """
headers["Cookie"] = "cnsc=" + cookie headers["Cookie"] = "cnsc=" + cookie
async with httpx.AsyncClient() as s: response = requests.request("GET", url + "?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token +
response = await s.get( ",-N000307," + ",-N" + resl, headers=headers, data={})
url=f"{url}?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N{token},-N000307,,-N{resl}", html = BeautifulSoup(response.content.decode("utf-8"), 'lxml')
headers=headers) table = html.find('table', attrs={"class": "nb list"})
html = BeautifulSoup(response.content.decode("utf-8"), 'lxml') body = table.find("tbody")
table = html.find('table', attrs={"class": "nb list"}) vorl = body.find_all("tr")
body = table.find("tbody") vorlist = []
vorl = body.find_all("tr") for row in vorl:
vorlist = [] cols = row.find_all("td")
tasks = [] col = [[e.text.strip()] for e in cols]
i = 0 if len(col) != 0:
for row in vorl: if len(col[4][0]) == 0:
cols = row.find_all("td") col[2] = getPruefung(row.find("a")["href"])
col = [[e.text.strip()] for e in cols]
if len(col) != 0:
if len(col[4][0]) == 0:
tasks += [getPruefung(s, row.find("a")["href"])]
col[2] = i
i += 1
vorlist += [col[1:4]] vorlist += [col[1:4]]
notlisted = await asyncio.gather(*tasks, return_exceptions=True) return vorlist
for i in vorlist:
for e in range(0, len(i)):
if isinstance(i[e], int):
i[e] = notlisted[i[e]]
return vorlist[:-1]
async def getPruefung(s, url): def getPruefung(url):
""" """
Ermittelt Noten "geschachtelter" Prüfungen, die nicht auf der Hauptseite angezeigt werden. Ermittelt Noten "geschachtelter" Prüfungen, die nicht auf der Hauptseite angezeigt werden.
TODO: Namen der spezifischen Prüfungen auch zurückgeben, um Zusammensetzung zu spezifizieren. TODO: Namen der spezifischen Prüfungen auch zurückgeben, um Zusammensetzung zu spezifizieren.
:param s:
:param url: :param url:
:return list: :return list:
""" """
response = await s.get("https://dualis.dhbw.de" + url, headers=headers) response = requests.request("GET", "https://dualis.dhbw.de" + url, headers=headers, data={})
html = BeautifulSoup(response.content.decode("utf-8"), 'lxml') html = BeautifulSoup(response.content.decode("utf-8"), 'lxml')
table = html.find('table') table = html.find('table')
pruefung = table.find_all("tr") pruefung = table.find_all("tr")

View File

@ -1,16 +1,13 @@
import json import json
from init import db, Meals, scheduler
import asyncio
from init import db, Meals, scheduler, app
import datetime import datetime
import requests
import time import time
import httpx
nomeal = 'Essen nicht (mehr) verfügbar' nomeal = 'Essen nicht (mehr) verfügbar'
async def getMeals(day: datetime): def getMeals(day: datetime):
""" """
Liefert alle Mahlzeiten eines Tages. \n Liefert alle Mahlzeiten eines Tages. \n
Befinden sie sich schon in der Datenbank, werden diese zurückgegeben. \n Befinden sie sich schon in der Datenbank, werden diese zurückgegeben. \n
@ -26,10 +23,10 @@ async def getMeals(day: datetime):
essen += [i.name] essen += [i.name]
essen.sort(key=len, reverse=True) essen.sort(key=len, reverse=True)
return essen return essen
return await getMealsFromAPI(day, dbentry=True) return getMealsFromAPI(day, dbentry=True)
async def getMealsFromAPI(day: str, dbentry: bool = False): def getMealsFromAPI(day: str, dbentry: bool = False):
""" """
Fragt die Mensa-API nach den Mahlzeiten eines Tages ab. \n Fragt die Mensa-API nach den Mahlzeiten eines Tages ab. \n
Wenn dbentry: Schreibt die Ergebnisse in die Datenbank. \n Wenn dbentry: Schreibt die Ergebnisse in die Datenbank. \n
@ -38,46 +35,46 @@ async def getMealsFromAPI(day: str, dbentry: bool = False):
:param dbentry: :param dbentry:
:return [Name1, Name2, ...]: :return [Name1, Name2, ...]:
""" """
async with httpx.AsyncClient() as s: url = "https://dh-api.paulmartin.cloud/plans/" + day + "?canteens=erzberger"
response = await s.get(url=f"https://dh-api.paulmartin.cloud/plans/{day}?canteens=erzberger") response = requests.request("GET", url)
response = response.content response = response.content
jres = json.loads(response.decode("utf-8")) jres = json.loads(response.decode("utf-8"))
essen = [] essen = []
try: try:
num = len(jres["data"][0]["lines"]) num = len(jres["data"][0]["lines"])
for i in range(num): for i in range(num):
try: try:
jsmeal = jres["data"][0]["lines"][i]["meals"] jsmeal = jres["data"][0]["lines"][i]["meals"]
cont = True cont = True
except IndexError: except IndexError:
essen = [] essen = []
cont = False cont = False
if cont: if cont:
for e in range(len(jsmeal)): for e in range(len(jsmeal)):
ji = jsmeal[e] ji = jsmeal[e]
name = ji["name"] name = ji["name"]
if pricetofloat(ji["price"]) >= 1.1: if pricetofloat(ji["price"]) >= 1.1:
vegan = ji["classifiers"].count("VG") == 1 vegan = ji["classifiers"].count("VG") == 1
schwein = ji["classifiers"].count("S") == 1 schwein = ji["classifiers"].count("S") == 1
if vegan: if vegan:
veget = True veget = True
else: else:
veget = ji["classifiers"].count("VEG") == 1 veget = ji["classifiers"].count("VEG") == 1
if veget: if veget:
if name.count("Reibekäse") > 0: if name.count("Reibekäse") > 0:
vegan = True vegan = True
essen += [name] essen += [name]
if dbentry: if dbentry:
mid = int(time.time() * 1000) % 100000 mid = int(time.time() * 1000) % 100000
neu = Meals(date=day, name=name, id=mid, vegan=vegan, vegetarian=veget, schwein=schwein) neu = Meals(date=day, name=name, id=mid, vegan=vegan, vegetarian=veget, schwein=schwein)
db.session.add(neu) db.session.add(neu)
db.session.commit() db.session.commit()
if not essen: if not essen:
essen = [nomeal]
except KeyError:
essen = [nomeal] essen = [nomeal]
return essen except KeyError:
essen = [nomeal]
return essen
def pricetofloat(price: str): def pricetofloat(price: str):
@ -113,35 +110,29 @@ def formatDay(day: datetime):
return day return day
async def refreshMeals(): @scheduler.task('cron', id="refreshMeals", hour='8-11', day_of_week='*', minute='15', week='*', second='30')
def refreshMeals():
""" """
Aktualisiert immer vormittags alle Mahlzeiten in der Datenbank. \n Aktualisiert immer vormittags alle Mahlzeiten in der Datenbank. \n
Datenbankeinträge werden ersetzt, wenn die API andere Mahlzeiten liefert. Datenbankeinträge werden ersetzt, wenn die API andere Mahlzeiten liefert.
""" """
print("Aktualisiere Essenspläne...\n") print("Aktualisiere Essenspläne...\n")
with scheduler.app.app_context():
table = Meals.query.all() table = Meals.query.all()
dates = [] dates = []
for i in table: for i in table:
if i.date not in dates: if i.date not in dates:
dates += [i.date] dates += [i.date]
for i in range(len(dates)): for i in range(len(dates)):
dates[i] = formatDay(dates[i]) dates[i] = formatDay(dates[i])
for i in dates: for i in dates:
apinames = await getMealsFromAPI(i) apinames = getMealsFromAPI(i)
apinames = await getMealsFromAPI(i) dbmeals = Meals.query.filter_by(date=i).all()
dbmeals = Meals.query.filter_by(date=i).all() dbnames = []
dbnames = [] for m in dbmeals:
for m in dbmeals: dbnames += [m.name]
dbnames += [m.name] if set(dbnames) != set(apinames) and nomeal not in apinames:
if set(dbnames) != set(apinames) and nomeal not in apinames: for n in dbnames:
for n in dbnames: db.session.delete(Meals.query.filter_by(date=i, name=n).first())
db.session.delete(Meals.query.filter_by(date=i, name=n).first()) db.session.commit()
db.session.commit() getMealsFromAPI(i, True)
await getMealsFromAPI(i, True)
@scheduler.task('cron', id="mensaschedule", hour='8-11', day_of_week='*', minute='*/15', week='*', second='5')
def mensaschedule():
with app.app_context():
asyncio.run(refreshMeals())

View File

@ -1,24 +1,9 @@
import urllib.error import urllib.error
from urllib.request import urlretrieve from urllib.request import urlretrieve
import asyncio
import httpx
import icalendar import icalendar
import json import json
import recurring_ical_events import recurring_ical_events
from init import scheduler
from init import scheduler, app
async def fetchPlan(session, url):
return await session.get(url=url)
def writeToFile(filename, data):
with open(filename, 'w+') as f:
f.write(data.text)
f.close()
def parseURL(url: str): def parseURL(url: str):
@ -49,7 +34,7 @@ def parseURL(url: str):
return 0 return 0
async def getNewRapla(url: str): def getNewRapla(url: str):
""" """
Speichert den iCal eines Raplas auf dem Server. \n Speichert den iCal eines Raplas auf dem Server. \n
Gibt Namen der Datei zurück. \n Gibt Namen der Datei zurück. \n
@ -65,18 +50,16 @@ async def getNewRapla(url: str):
try: try:
urlretrieve(url, "calendars/rapla" + kurs + ".ical") urlretrieve(url, "calendars/rapla" + kurs + ".ical")
async with httpx.AsyncClient() as s:
response = await fetchPlan(s, url)
writeToFile(f"calendars/rapla{kurs}.ical", response)
except urllib.error.URLError: except urllib.error.URLError:
return -1 return -1
file = open("calendars/list.json", "r+") file = open("calendars/list.json", "r+")
jsoncal = json.load(file) jsoncal = json.load(file)
jsoncal.update({kurs: [f"rapla{kurs}.ical", url]}) jsoncal.update({kurs: ["rapla" + kurs + ".ical", url]})
file.close() file.close()
file = open("calendars/list.json", "w") file = open("calendars/list.json", "w")
json.dump(jsoncal, file, indent=4) json.dump(jsoncal, file, indent=4)
return f"rapla{kurs}.ical" return "rapla" + kurs + ".ical"
def getIcal(kurs: str): def getIcal(kurs: str):
@ -110,23 +93,13 @@ def getRaplas():
return sorted(kursl), sorted(filel), sorted(urll) return sorted(kursl), sorted(filel), sorted(urll)
async def refreshRapla(): @scheduler.task("interval", id="refreshRapla", minutes=5)
def refreshRapla():
""" """
Aktualisiert alle 5 Minuten alle gespeicherten Raplas. Aktualisiert alle 5 Minuten alle gespeicherten Raplas.
""" """
filel = getRaplas()[1] filel = getRaplas()[1]
urll = getRaplas()[2] urll = getRaplas()[2]
jobl = [] for i in range(len(filel)):
async with httpx.AsyncClient() as s: print("Update Rapla: " + filel[i][:-5])
for i in range(len(filel)): urlretrieve(urll[i], "calendars/" + filel[i])
print(f"Update Rapla: {filel[i][:-5]}")
jobl += [fetchPlan(s, urll[i])]
callist = await asyncio.gather(*jobl, return_exceptions=True)
for cal in range(len(callist)):
writeToFile(f"calendars/{filel[cal]}", callist[cal])
@scheduler.task('cron', id="raplaschedule", hour='*', day_of_week='*', minute='*/15', week='*')
def raplaschedule():
with app.app_context():
asyncio.run(refreshRapla())

22
init.py
View File

@ -1,12 +1,11 @@
import apscheduler
from flask import Flask from flask import Flask
from flask_login import LoginManager, UserMixin from flask_login import LoginManager, UserMixin
from flask_sqlalchemy import SQLAlchemy from flask_sqlalchemy import SQLAlchemy
from flask_talisman import Talisman from flask_talisman import Talisman
from get_mysql import get_mysql from get_mysql import get_mysql
import atexit import atexit
from flask_apscheduler import APScheduler from flask_apscheduler import APScheduler
from turbo_flask import Turbo
def create(): def create():
@ -15,6 +14,7 @@ def create():
:return app: :return app:
""" """
app = Flask(__name__) app = Flask(__name__)
turbo.init_app(app)
dbpw = get_mysql()[1] dbpw = get_mysql()[1]
dbun = get_mysql()[0] dbun = get_mysql()[0]
@ -56,17 +56,7 @@ class Dualis(db.Model):
token = db.Column(db.String(255), unique=True) token = db.Column(db.String(255), unique=True)
uid = db.Column(db.Integer, primary_key=True) uid = db.Column(db.Integer, primary_key=True)
token_created = db.Column(db.Integer) token_created = db.Column(db.Integer)
semester = db.Column(db.String(15)) result_list = db.Column(db.String(15))
class Semesterlist(db.Model):
"""
Datenbank-Modell für Semester-Liste.
"""
uid = db.Column(db.Integer)
semestername = db.Column(db.String(25))
semesterid = db.Column(db.String(15))
itemid = db.Column(db.Integer, primary_key=True)
class Meals(db.Model): class Meals(db.Model):
@ -81,11 +71,11 @@ class Meals(db.Model):
schwein = db.Column(db.Boolean) schwein = db.Column(db.Boolean)
scheduler = APScheduler () scheduler = APScheduler()
turbo = Turbo()
app = create() app = create()
def_src = ["*.paulmartin.cloud", '\'self\''] def_src = ["*.paulmartin.cloud", '\'self\'']
Talisman(app, content_security_policy={"default-src": def_src, "script-src": def_src # + ["'unsafe-inline'"] # Talisman(app, content_security_policy={"default-src": def_src, "script-src": def_src # + ["'unsafe-inline'"]})
})
scheduler.init_app(app) scheduler.init_app(app)
scheduler.start() scheduler.start()
scheduler.api_enabled = True scheduler.api_enabled = True

View File

@ -1,6 +1,3 @@
from init import Semesterlist, User
def getCookie(cookies): def getCookie(cookies):
""" """
Liefert (letzten) Cookie der Cookies-Liste zurück. Liefert (letzten) Cookie der Cookies-Liste zurück.
@ -11,16 +8,3 @@ def getCookie(cookies):
for c in cookies: for c in cookies:
cookie = c.value cookie = c.value
return cookie return cookie
def getSemesterList(uid):
semesterlist = Semesterlist.query.filter_by(uid=uid).all()
semester = []
for s in semesterlist:
semester += [[s.semestername, s.semesterid]]
semester.sort(key=lambda x: x[-1], reverse=True)
return semester
def loadUser(uid):
return User.query.filter_by(id=uid).first()

View File

@ -1,17 +1,13 @@
beautifulsoup4~=4.12.2 beautifulsoup4
Flask~=3.0.3 Flask
Flask_APScheduler Flask_APScheduler
Flask_Login Flask_Login
flask_sqlalchemy flask_sqlalchemy
icalendar~=5.0.11 icalendar
recurring_ical_events recurring_ical_events
Requests~=2.31.0 Requests
talisman talisman
Werkzeug~=3.0.0 Werkzeug
lxml lxml
bs4~=0.0.1 bs4
pytz~=2023.3.post1 pytz
asyncio~=3.4.3
httpx~=1.0.0b0
celery~=5.4.0rc2

View File

@ -10,9 +10,8 @@ import time
import random import random
import fetchDUALIS import fetchDUALIS
import fetchMENSA
import fetchRAPLA import fetchRAPLA
from requesthelpers import * import requesthelpers
from fetchRAPLA import * from fetchRAPLA import *
from calendar_generation import getWeek from calendar_generation import getWeek
from init import * from init import *
@ -24,10 +23,14 @@ def index():
Leitet den normalen Website-Aufruf zum Login weiter. Leitet den normalen Website-Aufruf zum Login weiter.
:return HTML: :return HTML:
""" """
return redirect(url_for("login")) if current_user.is_authenticated:
return welcome()
else:
return login()
@app.route("/dashboard")
@app.route("/dashboard", methods=["GET", "POST"])
@login_required @login_required
def welcome(): def welcome():
""" """
@ -47,36 +50,40 @@ def welcome():
else: else:
t = "hidden" t = "hidden"
p = "" p = ""
return render_template('dashboard.html', kurs=kurs, name=name, theorie=t, praxis=p) if turbo.can_stream():
print ("Streaming...")
return turbo.stream([
turbo.update(
render_template('dashboard.html', kurs=kurs, name=name, theorie=t, praxis=p), target="content")
])
else:
print ("Not streaming!")
return render_template('dashboard.html', kurs=kurs, name=name, theorie=t, praxis=p)
@app.route("/theorie/noten", methods=["GET", "POST"]) @app.route("/theorie/noten")
@login_required @login_required
async def displayNoten(): def displayNoten():
""" """
Zeigt die Noten aus Dualis an. Hierfür ist ein aktives Token nötig. Zeigt die Noten aus Dualis an. Hierfür ist ein aktives Token nötig.
:return HTML: :return HTML:
""" """
d = Dualis.query.filter_by(uid=current_user.id).first() d = Dualis.query.filter_by(uid=current_user.id).first()
if request.method == "POST": if not d.result_list:
d.semester = request.form.get("sem")
db.session.commit()
if not d.semester:
return redirect(url_for("getSemester", next=url_for(request.endpoint))) return redirect(url_for("getSemester", next=url_for(request.endpoint)))
t = d.token t = d.token
chosensemester = d.semester sem = d.result_list
c = request.cookies.get("cnsc") c = request.cookies.get("cnsc")
timeout = fetchDUALIS.timeOut(d, c, "displayNoten") timeout = fetchDUALIS.timeOut(d, c, "displayNoten")
if timeout: if timeout:
return timeout return timeout
semester = getSemesterList(current_user.id) res = fetchDUALIS.getResults(t, c, sem)
noten = await fetchDUALIS.getResults(t, c, chosensemester) return render_template("noten.html", noten=res, semester=fetchDUALIS.getSem(t, c), sel=sem, s="n", praxis="hidden")
return render_template("noten.html", noten=noten, semester=semester, sel=chosensemester, s="n", praxis="hidden")
@app.route("/plan", methods=["GET"]) @app.route("/plan", methods=["GET"])
@login_required @login_required
async def displayRapla(): def displayRapla():
""" """
Zeigt den Stundenplan für eingeloggte User an. \n Zeigt den Stundenplan für eingeloggte User an. \n
TODO: Persönliche Filter, Notizen, Essensvorlieben etc. berücksichtigen. TODO: Persönliche Filter, Notizen, Essensvorlieben etc. berücksichtigen.
@ -92,14 +99,14 @@ async def displayRapla():
samstag = request.args.get("samstag") samstag = request.args.get("samstag")
if not samstag: if not samstag:
samstag = False samstag = False
events = await getWeek(week, fetchRAPLA.getIcal(current_user.kurs), samstag) events = getWeek(week, fetchRAPLA.getIcal(current_user.kurs), samstag)
return render_template("plan-user.html", events=events[0], eventdays=events[1], return render_template("plan-user.html", events=events[0], eventdays=events[1],
name=current_user.name, prev=str(events[2])[:10], next=str(events[3])[:10], mon=events[4], name=current_user.name, prev=str(events[2])[:10], next=str(events[3])[:10], mon=events[4],
s="p", praxis="hidden") s="p", praxis="hidden")
@app.route("/plan/<string:kurs>") @app.route("/plan/<string:kurs>")
async def displayPlan(kurs): def displayPlan(kurs):
""" """
Zeigt den Stundenplan ohne Login an. \n Zeigt den Stundenplan ohne Login an. \n
Präferenzen werden nicht berücksichtigt. Präferenzen werden nicht berücksichtigt.
@ -122,7 +129,7 @@ async def displayPlan(kurs):
samstag = request.args.get("samstag") samstag = request.args.get("samstag")
if not samstag: if not samstag:
samstag = False samstag = False
events = await getWeek(week, plan, samstag) events = getWeek(week, plan, samstag)
return render_template("plan-anon.html", events=events[0], eventdays=events[1], kurs=kurs, return render_template("plan-anon.html", events=events[0], eventdays=events[1], kurs=kurs,
prev=str(events[2])[:10], next=str(events[3])[:10], mon=events[4], praxis="hidden") prev=str(events[2])[:10], next=str(events[3])[:10], mon=events[4], praxis="hidden")
else: else:
@ -140,7 +147,7 @@ def redKurs():
@app.route("/set-up/kurs") @app.route("/set-up/kurs")
@login_required @login_required
async def getKurs(): def getKurs():
""" """
Automatische Kurs-Auswahl. \n Automatische Kurs-Auswahl. \n
Aktives Dualis-Token benötigt. Aktives Dualis-Token benötigt.
@ -154,7 +161,7 @@ async def getKurs():
return timeout return timeout
e = False e = False
if not current_user.kurs: if not current_user.kurs:
kurs = await fetchDUALIS.getKurs(d.token, cookie) kurs = fetchDUALIS.getKurs(d.token, cookie)
if kurs != 0: if kurs != 0:
if not fetchRAPLA.getIcal(kurs): if not fetchRAPLA.getIcal(kurs):
return render_template('kurs.html', detected=(kurs, e), s="s", theorie="hidden", praxis="hidden", return render_template('kurs.html', detected=(kurs, e), s="s", theorie="hidden", praxis="hidden",
@ -175,23 +182,15 @@ async def getKurs():
@app.route("/set-up/semester") @app.route("/set-up/semester")
@login_required @login_required
async def getSemester(): def getSemester():
""" """
Manuelle Semester-Auswahl. Manuelle Semester-Auswahl.
:return HTML: :return HTML:
""" """
t = Dualis.query.filter_by(uid=current_user.id).first().token t = Dualis.query.filter_by(uid=current_user.id).first().token
c = request.cookies.get("cnsc") c = request.cookies.get("cnsc")
semesterlist = Semesterlist.query.filter_by(uid=current_user.id).all()
if not semesterlist: return render_template("semester.html", semester=fetchDUALIS.getSem(t, c), s="s", theorie="hidden", praxis="hidden")
semester = await fetchDUALIS.getSem(t, c)
for i in semester:
semitem = Semesterlist(semestername=i[0], semesterid=i[1], uid=current_user.id, itemid=current_user.id*int(i[1][-7:])//1000000)
db.session.add(semitem)
db.session.commit()
else:
semester = getSemesterList(current_user.id)
return render_template("semester.html", semester=semester, s="s", theorie="hidden", praxis="hidden")
@app.route("/set-up/semester", methods=["POST"]) @app.route("/set-up/semester", methods=["POST"])
@ -205,7 +204,7 @@ def setSemester():
if not n: if not n:
n = url_for("welcome") n = url_for("welcome")
d = Dualis.query.filter_by(uid=current_user.id).first() d = Dualis.query.filter_by(uid=current_user.id).first()
d.semester = request.form.get("sem") d.result_list = request.form.get("sem")
db.session.commit() db.session.commit()
return redirect(n) return redirect(n)
@ -223,7 +222,7 @@ def chooseRaplas():
@app.route("/set-up/rapla", methods=["POST"]) @app.route("/set-up/rapla", methods=["POST"])
@login_required @login_required
async def getRapla(): def getRapla():
""" """
Verarbeitet die Eingabe von chooseRaplas(). Verarbeitet die Eingabe von chooseRaplas().
:return HTML: :return HTML:
@ -233,10 +232,10 @@ async def getRapla():
if file == url == "None": if file == url == "None":
return redirect(url_for("chooseRaplas")) return redirect(url_for("chooseRaplas"))
if file != "None": if file != "None":
loadUser(current_user.id).kurs = file[5:-5] User.query.filter_by(id=current_user.id).first().kurs = file[5:-5]
db.session.commit() db.session.commit()
elif url != "None": elif url != "None":
file = await getNewRapla(url) file = getNewRapla(url)
if type(file) is not int: if type(file) is not int:
User.query.filter_by(id=current_user.id).first().kurs = file[5:-5] User.query.filter_by(id=current_user.id).first().kurs = file[5:-5]
db.session.commit() db.session.commit()
@ -255,7 +254,7 @@ def login():
@app.route("/log-in", methods=["POST"]) @app.route("/log-in", methods=["POST"])
async def login_post(): def login_post():
""" """
Verarbeitet die Eingabe von login(). \n Verarbeitet die Eingabe von login(). \n
Falls der User schon angelegt ist, wird das Passwort verglichen. \n Falls der User schon angelegt ist, wird das Passwort verglichen. \n
@ -271,18 +270,18 @@ async def login_post():
success = make_response(redirect(url_for("getKurs"))) success = make_response(redirect(url_for("getKurs")))
user = User.query.filter_by(email=email).first() user = User.query.filter_by(email=email).first()
t = await fetchDUALIS.checkUser(email, password) t = fetchDUALIS.checkUser(email, password)
if t[0] == -2: if t[0] == -2:
return redirect(url_for("login", code=-2)) return redirect(url_for("login", code=-2))
if user: if user:
dualis = Dualis.query.filter_by(uid=user.id).first() dualis = Dualis.query.filter_by(uid=user.id).first()
dualis.token = t[0] dualis.token = t[0]
newcookie = t[1] newcookie = requesthelpers.getCookie(t[1].cookies)
dualis.token_created = time.time() dualis.token_created = time.time()
db.session.commit() db.session.commit()
login_user(user) login_user(user)
if user.kurs: if user.kurs:
if not dualis.semester: if not dualis.result_list:
success = make_response(redirect(url_for("getSemester"))) success = make_response(redirect(url_for("getSemester")))
elif not n: elif not n:
success = make_response(redirect(url_for("welcome"))) success = make_response(redirect(url_for("welcome")))
@ -308,14 +307,14 @@ async def login_post():
@app.route("/log-out") @app.route("/log-out")
@login_required @login_required
async def logout(): def logout():
""" """
Loggt den User aus. Loggt den User aus.
:return Empty Token: :return Empty Token:
""" """
cookie = request.cookies.get("cnsc") cookie = request.cookies.get("cnsc")
dualis = Dualis.query.filter_by(uid=current_user.id).first() dualis = Dualis.query.filter_by(uid=current_user.id).first()
await fetchDUALIS.logOut(dualis.token, cookie) fetchDUALIS.logOut(dualis.token, cookie)
dualis.token = None dualis.token = None
db.session.commit() db.session.commit()
logout_user() logout_user()

View File

@ -5,37 +5,38 @@
<link rel="stylesheet" type="text/css" href={{ url_for("static", filename="style.css") }}> <link rel="stylesheet" type="text/css" href={{ url_for("static", filename="style.css") }}>
<meta http-equiv="refresh" content="510"> <meta http-equiv="refresh" content="510">
<script async src="https://analytics.paulmartin.cloud/script.js" data-website-id="459fa66e-e255-4393-8e89-ead8b1572d0d"></script> <script async src="https://analytics.paulmartin.cloud/script.js" data-website-id="459fa66e-e255-4393-8e89-ead8b1572d0d"></script>
{{ turbo() }}
{% block head %} {% block head %}
{% endblock %} {% endblock %}
</head> </head>
<body> <body>
<nav> <nav>
<ul> <ul id = "navbar-items">
{% if theorie=="hidden" and praxis=="hidden" %} {% if theorie=="hidden" and praxis=="hidden" %}
<li class="top"><a class="top" href={{ url_for("welcome", sel="theorie") }}>Theorie</a></li> <li class="top"><a class="top" href={{ url_for("index", sel="theorie") }}>Theorie</a></li>
<li class="top"><a class="top" href={{ url_for("welcome", sel="praxis") }}>Praxis</a></li> <li class="top"><a class="top" href={{ url_for("index", sel="praxis") }}>Praxis</a></li>
{% elif theorie == "hidden" %} {% elif theorie == "hidden" %}
<li class="top"><a class="top" href={{ url_for("welcome", sel="theorie") }}>Theorie</a></li> <li class="top"><a class="top" href={{ url_for("index", sel="theorie") }}>Theorie</a></li>
<li class="top selected"><a class="top selected" href={{ url_for("welcome", sel="praxis") }}>Praxis</a></li> <li class="top selected"><a class="top selected" href={{ url_for("index", sel="praxis") }}>Praxis</a></li>
{% else %} {% else %}
<li class="top selected"><a class="top selected" href={{ url_for("welcome", sel="theorie") }}>Theorie</a></li> <li class="top selected"><a class="top selected" href={{ url_for("index", sel="theorie") }}>Theorie</a></li>
<li class="top"><a class="top" href={{ url_for("welcome", sel="praxis") }}>Praxis</a></li> <li class="top"><a class="top" href={{ url_for("index", sel="praxis") }}>Praxis</a></li>
{% endif %}
<li {{ theorie }}><a {% if s == "n" %} class="selected" {% endif %} href={{ url_for("displayNoten") }}>Noten</a></li>
<li {{ theorie }}><a {% if s == "p" %} class="selected" {% endif %} href={{ url_for("displayRapla") }}>Stundenplan</a></li>
<li {{ praxis }}><a {% if s == "t" %} class="selected" {% endif %} href="">To Dos</a></li>
<li><a {% if s == "s" %} class="selected" {% endif %} href={{ url_for("redKurs") }}>Konfiguration</a></li>
{% if request.endpoint %}
{% if request.endpoint[:7] != "login" %}
<li><a class="bottom" href={{ url_for("logout") }}>Log-Out</a></li>
{% endif %} {% endif %}
{% endif %} <li {{ theorie }}><a {% if s == "n" %} class="selected" {% endif %} href={{ url_for("displayNoten") }}>Noten</a></li>
</ul> <li {{ theorie }}><a {% if s == "p" %} class="selected" {% endif %} href={{ url_for("displayRapla") }}>Stundenplan</a></li>
<li {{ praxis }}><a {% if s == "t" %} class="selected" {% endif %} href="">To Dos</a></li>
<li><a {% if s == "s" %} class="selected" {% endif %} href={{ url_for("redKurs") }}>Konfiguration</a></li>
<li {% block logout %}{% endblock %}><a class="bottom" href={{ url_for("logout") }}>Log-Out</a></li>
</ul>
</nav> </nav>
<div class="cs1"> <turbo-frame>
{% block content %} <div class="cs1" id = 'page'>
{% endblock %} <div id="content">
</div> {% block content %}
{% endblock %}
</div>
</div>
</turbo-frame>
</body> </body>
</html> </html>

View File

@ -1,4 +1,8 @@
{% extends "index.html" %} {% extends "index.html" %}
{% block logout %}
hidden
{% endblock %}
{% block content %} {% block content %}
<div class="cs"> <div class="cs">
<h1>Einloggen</h1> <h1>Einloggen</h1>

21
templates/noten-list.html Normal file
View File

@ -0,0 +1,21 @@
{% for i in range (semester|length) %}
{% if semester[i][1]==sel %}
<h1>Deine Noten im {{ semester[i][0] }}</h1>
{% endif %}
{% endfor %}
<br>
<br>
{% for i in noten %}
{% if i[1]|length == 1 %}
{% if i[1][0][0].isnumeric() or i[1][0][-1].isalpha()%}
<h2>{{ i[0][0] }}: {{ i[1][0] }} (Credits: {{ i[2][0] }})</h2>
{% else %}
<h2>{{ i[0][0] }}: {{ i[1][0][-3:] }} (Credits: {{ i[2][0] }})</h2>
{% endif %}
{% else %}
<h2>{{ i[0][0] }} (Credits: {{ i[2][0] }}):</h2>
{% for e in i[1] %}
<h2>&emsp;&emsp;&emsp;&emsp;&emsp;&emsp;{{ e }}</h2>
{% endfor %}
{% endif %}
{% endfor %}

View File

@ -4,30 +4,14 @@
<meta http-equiv="Refresh" content="600"> <meta http-equiv="Refresh" content="600">
{% endblock %} {% endblock %}
{% block content %} {% block content %}
{% for i in range (semester|length) %} <turbo-frame>
{% if semester[i][1]==sel %} <div id="noten">
<h1>Deine Noten im {{ semester[i][0] }}</h1> {% include "noten-list.html" %}
{% endif %} </div>
{% endfor %} </turbo-frame>
<br> <br>
<br> <br>
{% for i in noten %} <form id= "dropdown" method="post" action={{ url_for ("setSemester", next=url_for("displayNoten")) }}>
{% if i[1]|length == 1 %}
{% if i[1][0][0].isnumeric() or i[1][0][-1].isalpha()%}
<h2>{{ i[0][0] }}: {{ i[1][0] }} (Credits: {{ i[2][0] }})</h2>
{% else %}
<h2>{{ i[0][0] }}: {{ i[1][0][-3:] }} (Credits: {{ i[2][0] }})</h2>
{% endif %}
{% else %}
<h2>{{ i[0][0] }} (Credits: {{ i[2][0] }}):</h2>
{% for e in i[1] %}
<h2>&emsp;&emsp;&emsp;&emsp;&emsp;&emsp;{{ e }}</h2>
{% endfor %}
{% endif %}
{% endfor %}
<br>
<br>
<form id= "dropdown" method="post" action={{ url_for ("displayNoten") }}>
<label for="sem">Semester wählen! </label> <label for="sem">Semester wählen! </label>
<select name="sem" id="sem"> <select name="sem" id="sem">
{% for i in range (semester|length) %} {% for i in range (semester|length) %}

View File

@ -1,284 +0,0 @@
import asyncio
import httpx
import time
import urllib.parse
import requests
from flask import Flask
from bs4 import BeautifulSoup
from celery import Celery
import fetchDUALIS
from login_data import passwort, email #CREATE LOCAL login_data.py!!!
app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
headers = {
'Cookie': 'cnsc=0',
'Content-Type': 'application/x-www-form-urlencoded',
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/58.0.3029.110 Safari/537.36',
'Accept-Encoding': 'utf-8'
}
url = "https://dualis.dhbw.de/scripts/mgrqispi.dll"
fpw = urllib.parse.quote(passwort, safe='', encoding=None, errors=None)
fmail = urllib.parse.quote(email, safe='', encoding=None, errors=None)
@celery.task(bind=True)
def celery_requests(self):
with httpx.Client() as s:
content = (f'usrname={fmail}&pass={fpw}&ARGUMENTS=clino%2Cusrname%2Cpass%2Cmenuno%2Cmenu_type%2Cbrowser'
f'%2Cplatform&APPNAME=CampusNet&PRGNAME=LOGINCHECK')
response = s.post(url=url, headers=headers, content=content)
return response, s
# noinspection DuplicatedCode
async def checkUser_celery():
req = celery_requests.apply_async()
response = req[0]
s = req[1]
header = response.headers
try:
refresh = header["REFRESH"]
arg = refresh.find("=-N") + 3
komma = refresh[arg:].find(",")
cookie = s.cookies.get("cnsc")
except KeyError:
return -2, s
token = refresh[arg:komma + arg]
return token, cookie
# noinspection DuplicatedCode
def checkUser_normal():
"""
Erhält von Dualis den Token und Cookie für User.
:param email:
:param password:
:return (Token, Session):
"""
s = requests.Session()
payload = 'usrname=' + fmail + '&pass=' + fpw + ('&ARGUMENTS=clino%2Cusrname%2Cpass%2Cmenuno%2Cmenu_type%2Cbrowser'
'%2Cplatform&APPNAME=CampusNet&PRGNAME=LOGINCHECK')
response = s.post(url, headers=headers, data=payload)
header = response.headers
try:
refresh = header["REFRESH"]
arg = refresh.find("=-N") + 3
komma = refresh[arg:].find(",")
except KeyError:
return -2, s
token = refresh[arg:komma + arg]
cookies = s.cookies
cookie = 0
for c in cookies:
cookie = c.value
return token, cookie
# noinspection DuplicatedCode
async def checkUser_async():
"""
Erhält von Dualis den Token und Cookie für User.
:param email:
:param password:
:return (Token, Session):
"""
# noinspection DuplicatedCode
async with httpx.AsyncClient() as s:
content = (f'usrname={fmail}&pass={fpw}&ARGUMENTS=clino%2Cusrname%2Cpass%2Cmenuno%2Cmenu_type%2Cbrowser'
f'%2Cplatform&APPNAME=CampusNet&PRGNAME=LOGINCHECK')
response = await s.post(url=url, headers=headers, content=content)
header = response.headers
try:
refresh = header["REFRESH"]
arg = refresh.find("=-N") + 3
komma = refresh[arg:].find(",")
cookie = s.cookies.get("cnsc")
except KeyError:
return -2, s
token = refresh[arg:komma + arg]
return token, cookie
# noinspection DuplicatedCode
def getSem_normal(token: int, cookie: str):
"""
Liefert die Liste aller auf Dualis verfügbaren Semester.
:param token:
:param cookie:
:return Liste [[Semester, Semester-ID], ...]:
"""
headers["Cookie"] = "cnsc=" + cookie
token = str(token)
response = requests.request("GET", url +
"?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token + ",-N000307,",
headers=headers, data={})
html = BeautifulSoup(response.text, 'lxml')
select = html.find('select')
select = select.find_all(value=True)
optlist = []
for i in select:
t = i.text.replace("Wi", "Winter").replace("So", "Sommer")
t = t.replace("Se", "semester")
optlist += [[t, i['value']]]
return optlist
# noinspection DuplicatedCode
async def getSem_async(token, cookie):
"""
Liefert die Liste aller auf Dualis verfügbaren Semester.
:param token:
:param cookie:
:return Liste [[Semester, Semester-ID], ...]:
"""
headers["Cookie"] = "cnsc=" + cookie
token = str(token)
async with httpx.AsyncClient() as s:
response = await s.get(url=url +
"?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token + ",-N000307,",
headers=headers)
html = BeautifulSoup(response.text, 'lxml')
select = html.find('select')
select = select.find_all(value=True)
optlist = []
for i in select:
t = i.text.replace("Wi", "Winter").replace("So", "Sommer")
t = t.replace("Se", "semester")
optlist += [[t, i['value']]]
return optlist
async def getSem_celery(token, cookie):
pass
# noinspection DuplicatedCode
async def getResults_async(token, cookie, resl):
headers["Cookie"] = "cnsc=" + cookie
async with httpx.AsyncClient() as s:
response = await s.get(url=url + "?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token +
",-N000307," + ",-N" + resl, headers=headers)
html = BeautifulSoup(response.content.decode("utf-8"), 'lxml')
table = html.find('table', attrs={"class": "nb list"})
body = table.find("tbody")
vorl = body.find_all("tr")
vorlist = []
tasks = []
i = 0
for row in vorl:
cols = row.find_all("td")
col = [[e.text.strip()] for e in cols]
if len(col) != 0:
if len(col[4][0]) == 0:
tasks += [getPruefung_async(s, row.find("a")["href"])]
col[2] = i
i += 1
vorlist += [col[1:4]]
extrakurse = await asyncio.gather(*tasks, return_exceptions=True)
for i in vorlist:
for e in range(0, len(i)):
if isinstance(i[e], int):
i[e] = extrakurse[i[e]]
return vorlist[:-1]
# noinspection DuplicatedCode
async def getPruefung_async(s, url):
response = await s.get("https://dualis.dhbw.de" + url, headers=headers)
html = BeautifulSoup(response.content.decode("utf-8"), 'lxml')
table = html.find('table')
pruefung = table.find_all("tr")
ret = []
for row in pruefung:
cols = row.find_all("td")
col = [e.text.strip() for e in cols]
if len(col) == 6 and len(col[3]) <= 13:
if len(col[3]) != 0:
ret += [col[0] + ": " + col[3][:3]]
if ret[-1][0] == ':':
ret[-1] = "Gesamt" + ret[-1]
if len(ret) == 0:
ret = ["Noch nicht gesetzt"]
return ret
def normal():
login = checkUser_normal()
token = login[0]
cookie = str(login[1])
semlist = getSem_normal(token, cookie)
for i in range(1, len(semlist)):
results = fetchDUALIS.getResults(token, cookie, semlist[i][1])
return [token, cookie, results]
async def async_normal():
login = await checkUser_async()
token = login[0]
cookie = str(login[1])
semlist = await getSem_async(token, cookie)
for i in range(1, len(semlist)):
results = await getResults_async(token, cookie, semlist[i][1])
return [token, cookie, results]
@app.route('/c')
async def celery_normal():
login = await checkUser_celery()
token = login[0]
cookie = str(login[1])
#print(token, cookie)
#semlist = await getSem_celery(token, cookie)
#print(semlist)
#for i in range(0, len(semlist)):
# results = await getResults_celery(token, cookie, semlist[i][1])
# print(results)
return [token, cookie]
@app.route('/')
async def tests():
normaltime = 0
asynctime = 0
iter = 20
print("Starting...")
for i in range(iter):
start = time.perf_counter()
n = normal()
end = time.perf_counter()
normaltimeloop = end - start
await fetchDUALIS.logOut(n[0], n[1])
start = time.perf_counter()
a = await async_normal()
end = time.perf_counter()
asynctimeloop = end - start
await fetchDUALIS.logOut(a[0], a[1])
if a[2] == n[2]:
normaltime += normaltimeloop
asynctime += asynctimeloop
print(str(((i + 1) / iter) * 100) + '%')
#vergl = "Gleicher Inhalt!"
#else:
# vergl = "ACHTUNG! Ungleicher Inhalt!"
#return "<br><h2>" + vergl + "</h2><br> <h1> Normal: " + str(normaltime) + "</h2><br><br> <h1> Async: " + str(asynctime) + "</h1>"
delta = normaltime / iter - asynctime / iter
return f"<br><h2> Durchschnitt normal: {normaltime / iter} </h2><br><h2> Durchschnitt asynchron: {asynctime / iter} </h2><br><h1> Durchschnittliches Delta: {delta} </h1><br>"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=1024, debug=True)