Merge branch 'refs/heads/main' into turbo
# Conflicts: # init.py
This commit is contained in:
@ -1,3 +1,6 @@
|
||||
import time
|
||||
|
||||
import asyncio
|
||||
import icalendar
|
||||
import datetime
|
||||
import recurring_ical_events
|
||||
@ -10,7 +13,7 @@ months = ["Januar", "Februar", "März", "April", "Mai", "Juni", "Juli", "August"
|
||||
"November", "Dezember"]
|
||||
|
||||
|
||||
def getWeek(weekstart: datetime, file: str, showsat: bool):
|
||||
async def getWeek(weekstart: datetime, file: str, showsat: bool):
|
||||
"""
|
||||
Liefert alle Events einer Woche zurück. \n
|
||||
Wochenstart wird automatisch auf den Montag der Woche gelegt. \n
|
||||
@ -67,10 +70,10 @@ def getWeek(weekstart: datetime, file: str, showsat: bool):
|
||||
"day": estart.day
|
||||
}
|
||||
eventl += [eventdict]
|
||||
return eventl, daylist(start_date, showsat), prevw, nextw, mon
|
||||
return eventl, await daylist(start_date, showsat), prevw, nextw, mon
|
||||
|
||||
|
||||
def daylist(weekstart: datetime, showsat: bool):
|
||||
async def daylist(weekstart: datetime, showsat: bool):
|
||||
"""
|
||||
Gibt die Essen einer Woche zurück.
|
||||
:param weekstart:
|
||||
@ -83,15 +86,19 @@ def daylist(weekstart: datetime, showsat: bool):
|
||||
r = 6
|
||||
else:
|
||||
r = 5
|
||||
essen = []
|
||||
for i in range(r):
|
||||
essen = getMeals(weekday)
|
||||
essen += [getMeals(weekday)]
|
||||
dayl += [{
|
||||
"day": weekday.day,
|
||||
"short": shortnames[i],
|
||||
"long": longnames[i],
|
||||
"mensa": essen
|
||||
"mensa": i
|
||||
}]
|
||||
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
|
||||
|
||||
|
||||
|
||||
@ -1,6 +1,2 @@
|
||||
{
|
||||
"TINF22B3": [
|
||||
"raplaTINF22B3.ical",
|
||||
"https://rapla.dhbw-karlsruhe.de/rapla?page=ical&user=vollmer&file=tinf22b3"
|
||||
]
|
||||
}
|
||||
138
fetchDUALIS.py
138
fetchDUALIS.py
@ -1,9 +1,10 @@
|
||||
import requests
|
||||
import urllib.parse
|
||||
import time
|
||||
from bs4 import BeautifulSoup
|
||||
from flask import redirect, url_for
|
||||
from init import Dualis
|
||||
import asyncio
|
||||
import httpx
|
||||
|
||||
headers = {
|
||||
'Cookie': 'cnsc=0',
|
||||
@ -16,31 +17,32 @@ headers = {
|
||||
url = "https://dualis.dhbw.de/scripts/mgrqispi.dll"
|
||||
|
||||
|
||||
def checkUser(email: str, password: str):
|
||||
async def checkUser(email: str, password: str):
|
||||
"""
|
||||
Erhält von Dualis den Token und Cookie für User.
|
||||
:param email:
|
||||
:param password:
|
||||
:return (Token, Session):
|
||||
"""
|
||||
s = requests.Session()
|
||||
fpw = urllib.parse.quote(password, safe='', encoding=None, errors=None)
|
||||
fmail = urllib.parse.quote(email, safe='', encoding=None, errors=None)
|
||||
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]
|
||||
return token, s
|
||||
async with httpx.AsyncClient() as s:
|
||||
fpw = urllib.parse.quote(password, 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'
|
||||
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
|
||||
|
||||
|
||||
def getKurs(token: int, cookie: str):
|
||||
async def getKurs(token: int, cookie: str):
|
||||
"""
|
||||
Bestimmt aus der ersten Prüfung den Kursbezeichner des Users.
|
||||
TODO: Umstellen auf Bezeichner INKL. Standort
|
||||
@ -51,34 +53,34 @@ def getKurs(token: int, cookie: str):
|
||||
try:
|
||||
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')
|
||||
link = html.body.find('a', attrs={'id': "Popup_details0001"})['href']
|
||||
response = requests.request("GET", url + link[21:], headers=headers, data={})
|
||||
html = BeautifulSoup(response.text, 'lxml')
|
||||
content = html.body.find('td', attrs={'class': 'level02'}).text
|
||||
start = content.find(" ") + 4
|
||||
end = start + (content[start:].find(" "))
|
||||
kurs = content[start:end]
|
||||
async with httpx.AsyncClient as s:
|
||||
response = await s.get(url=f"{url}?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N{token},-N000307,",
|
||||
headers=headers)
|
||||
html = BeautifulSoup(response.text, 'lxml')
|
||||
link = html.body.find('a', attrs={'id': "Popup_details0001"})['href']
|
||||
response = await s.get(url=f"{url}{link[21:]}", headers=headers)
|
||||
html = BeautifulSoup(response.text, 'lxml')
|
||||
content = html.body.find('td', attrs={'class': 'level02'}).text
|
||||
start = content.find(" ") + 4
|
||||
end = start + (content[start:].find(" "))
|
||||
kurs = content[start:end]
|
||||
except AttributeError:
|
||||
kurs = 0
|
||||
return kurs
|
||||
|
||||
|
||||
def logOut(token: int, cookie: str):
|
||||
async def logOut(token: int, cookie: str):
|
||||
"""
|
||||
Invalidiert Token und Cookie bei Dualis.
|
||||
:param token:
|
||||
:param cookie:
|
||||
"""
|
||||
headers["Cookie"] = "cnsc=" + cookie
|
||||
requests.request("GET", url + "?APPNAME=CampusNet&PRGNAME=LOGOUT&ARGUMENTS=-N" + str(token)
|
||||
+ ", -N001", headers=headers, data={})
|
||||
async with httpx.AsyncClient() as s:
|
||||
await s.get(url=f"{url}?APPNAME=CampusNet&PRGNAME=LOGOUT&ARGUMENTS=-N{token}, -N001", headers=headers)
|
||||
|
||||
|
||||
def getSem(token: int, cookie: str):
|
||||
async def getSem(token: int, cookie: str):
|
||||
"""
|
||||
Liefert die Liste aller auf Dualis verfügbaren Semester.
|
||||
:param token:
|
||||
@ -87,21 +89,21 @@ def getSem(token: int, cookie: str):
|
||||
"""
|
||||
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
|
||||
async with httpx.AsyncClient() as s:
|
||||
response = await s.get(url=f"{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
|
||||
|
||||
|
||||
def getResults(token, cookie: str, resl: str):
|
||||
async def getResults(token, cookie: str, resl: str):
|
||||
"""
|
||||
Liefert die Liste aller Prüfungsergebnisse eines Semesters.
|
||||
:param token:
|
||||
@ -110,31 +112,43 @@ def getResults(token, cookie: str, resl: str):
|
||||
:return [[Name, Note, Credits], ...]:
|
||||
"""
|
||||
headers["Cookie"] = "cnsc=" + cookie
|
||||
response = requests.request("GET", url + "?APPNAME=CampusNet&PRGNAME=COURSERESULTS&ARGUMENTS=-N" + token +
|
||||
",-N000307," + ",-N" + resl, headers=headers, data={})
|
||||
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 = []
|
||||
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:
|
||||
col[2] = getPruefung(row.find("a")["href"])
|
||||
async with httpx.AsyncClient() as s:
|
||||
response = await s.get(
|
||||
url=f"{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(s, row.find("a")["href"])]
|
||||
col[2] = i
|
||||
i += 1
|
||||
vorlist += [col[1:4]]
|
||||
return vorlist
|
||||
notlisted = 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] = notlisted[i[e]]
|
||||
return vorlist[:-1]
|
||||
|
||||
|
||||
def getPruefung(url):
|
||||
async def getPruefung(s, url):
|
||||
"""
|
||||
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.
|
||||
:param s:
|
||||
:param url:
|
||||
:return list:
|
||||
"""
|
||||
response = requests.request("GET", "https://dualis.dhbw.de" + url, headers=headers, data={})
|
||||
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")
|
||||
|
||||
137
fetchMENSA.py
137
fetchMENSA.py
@ -1,13 +1,16 @@
|
||||
import json
|
||||
from init import db, Meals, scheduler
|
||||
|
||||
import asyncio
|
||||
|
||||
from init import db, Meals, scheduler, app
|
||||
import datetime
|
||||
import requests
|
||||
import time
|
||||
import httpx
|
||||
|
||||
nomeal = 'Essen nicht (mehr) verfügbar'
|
||||
|
||||
|
||||
def getMeals(day: datetime):
|
||||
async def getMeals(day: datetime):
|
||||
"""
|
||||
Liefert alle Mahlzeiten eines Tages. \n
|
||||
Befinden sie sich schon in der Datenbank, werden diese zurückgegeben. \n
|
||||
@ -23,10 +26,10 @@ def getMeals(day: datetime):
|
||||
essen += [i.name]
|
||||
essen.sort(key=len, reverse=True)
|
||||
return essen
|
||||
return getMealsFromAPI(day, dbentry=True)
|
||||
return await getMealsFromAPI(day, dbentry=True)
|
||||
|
||||
|
||||
def getMealsFromAPI(day: str, dbentry: bool = False):
|
||||
async def getMealsFromAPI(day: str, dbentry: bool = False):
|
||||
"""
|
||||
Fragt die Mensa-API nach den Mahlzeiten eines Tages ab. \n
|
||||
Wenn dbentry: Schreibt die Ergebnisse in die Datenbank. \n
|
||||
@ -35,46 +38,46 @@ def getMealsFromAPI(day: str, dbentry: bool = False):
|
||||
:param dbentry:
|
||||
:return [Name1, Name2, ...]:
|
||||
"""
|
||||
url = "https://dh-api.paulmartin.cloud/plans/" + day + "?canteens=erzberger"
|
||||
response = requests.request("GET", url)
|
||||
response = response.content
|
||||
jres = json.loads(response.decode("utf-8"))
|
||||
essen = []
|
||||
try:
|
||||
num = len(jres["data"][0]["lines"])
|
||||
for i in range(num):
|
||||
try:
|
||||
jsmeal = jres["data"][0]["lines"][i]["meals"]
|
||||
cont = True
|
||||
except IndexError:
|
||||
essen = []
|
||||
cont = False
|
||||
if cont:
|
||||
for e in range(len(jsmeal)):
|
||||
ji = jsmeal[e]
|
||||
name = ji["name"]
|
||||
if pricetofloat(ji["price"]) >= 1.1:
|
||||
vegan = ji["classifiers"].count("VG") == 1
|
||||
schwein = ji["classifiers"].count("S") == 1
|
||||
if vegan:
|
||||
veget = True
|
||||
else:
|
||||
veget = ji["classifiers"].count("VEG") == 1
|
||||
if veget:
|
||||
if name.count("Reibekäse") > 0:
|
||||
vegan = True
|
||||
async with httpx.AsyncClient() as s:
|
||||
response = await s.get(url=f"https://dh-api.paulmartin.cloud/plans/{day}?canteens=erzberger")
|
||||
response = response.content
|
||||
jres = json.loads(response.decode("utf-8"))
|
||||
essen = []
|
||||
try:
|
||||
num = len(jres["data"][0]["lines"])
|
||||
for i in range(num):
|
||||
try:
|
||||
jsmeal = jres["data"][0]["lines"][i]["meals"]
|
||||
cont = True
|
||||
except IndexError:
|
||||
essen = []
|
||||
cont = False
|
||||
if cont:
|
||||
for e in range(len(jsmeal)):
|
||||
ji = jsmeal[e]
|
||||
name = ji["name"]
|
||||
if pricetofloat(ji["price"]) >= 1.1:
|
||||
vegan = ji["classifiers"].count("VG") == 1
|
||||
schwein = ji["classifiers"].count("S") == 1
|
||||
if vegan:
|
||||
veget = True
|
||||
else:
|
||||
veget = ji["classifiers"].count("VEG") == 1
|
||||
if veget:
|
||||
if name.count("Reibekäse") > 0:
|
||||
vegan = True
|
||||
|
||||
essen += [name]
|
||||
if dbentry:
|
||||
mid = int(time.time() * 1000) % 100000
|
||||
neu = Meals(date=day, name=name, id=mid, vegan=vegan, vegetarian=veget, schwein=schwein)
|
||||
db.session.add(neu)
|
||||
db.session.commit()
|
||||
if not essen:
|
||||
essen += [name]
|
||||
if dbentry:
|
||||
mid = int(time.time() * 1000) % 100000
|
||||
neu = Meals(date=day, name=name, id=mid, vegan=vegan, vegetarian=veget, schwein=schwein)
|
||||
db.session.add(neu)
|
||||
db.session.commit()
|
||||
if not essen:
|
||||
essen = [nomeal]
|
||||
except KeyError:
|
||||
essen = [nomeal]
|
||||
except KeyError:
|
||||
essen = [nomeal]
|
||||
return essen
|
||||
return essen
|
||||
|
||||
|
||||
def pricetofloat(price: str):
|
||||
@ -110,29 +113,35 @@ def formatDay(day: datetime):
|
||||
return day
|
||||
|
||||
|
||||
@scheduler.task('cron', id="refreshMeals", hour='8-11', day_of_week='*', minute='15', week='*', second='30')
|
||||
def refreshMeals():
|
||||
async def refreshMeals():
|
||||
"""
|
||||
Aktualisiert immer vormittags alle Mahlzeiten in der Datenbank. \n
|
||||
Datenbankeinträge werden ersetzt, wenn die API andere Mahlzeiten liefert.
|
||||
"""
|
||||
print("Aktualisiere Essenspläne...\n")
|
||||
with scheduler.app.app_context():
|
||||
table = Meals.query.all()
|
||||
dates = []
|
||||
for i in table:
|
||||
if i.date not in dates:
|
||||
dates += [i.date]
|
||||
for i in range(len(dates)):
|
||||
dates[i] = formatDay(dates[i])
|
||||
for i in dates:
|
||||
apinames = getMealsFromAPI(i)
|
||||
dbmeals = Meals.query.filter_by(date=i).all()
|
||||
dbnames = []
|
||||
for m in dbmeals:
|
||||
dbnames += [m.name]
|
||||
if set(dbnames) != set(apinames) and nomeal not in apinames:
|
||||
for n in dbnames:
|
||||
db.session.delete(Meals.query.filter_by(date=i, name=n).first())
|
||||
db.session.commit()
|
||||
getMealsFromAPI(i, True)
|
||||
|
||||
table = Meals.query.all()
|
||||
dates = []
|
||||
for i in table:
|
||||
if i.date not in dates:
|
||||
dates += [i.date]
|
||||
for i in range(len(dates)):
|
||||
dates[i] = formatDay(dates[i])
|
||||
for i in dates:
|
||||
apinames = await getMealsFromAPI(i)
|
||||
apinames = await getMealsFromAPI(i)
|
||||
dbmeals = Meals.query.filter_by(date=i).all()
|
||||
dbnames = []
|
||||
for m in dbmeals:
|
||||
dbnames += [m.name]
|
||||
if set(dbnames) != set(apinames) and nomeal not in apinames:
|
||||
for n in dbnames:
|
||||
db.session.delete(Meals.query.filter_by(date=i, name=n).first())
|
||||
db.session.commit()
|
||||
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())
|
||||
|
||||
@ -1,9 +1,24 @@
|
||||
import urllib.error
|
||||
from urllib.request import urlretrieve
|
||||
|
||||
import asyncio
|
||||
|
||||
import httpx
|
||||
import icalendar
|
||||
import json
|
||||
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):
|
||||
@ -34,7 +49,7 @@ def parseURL(url: str):
|
||||
return 0
|
||||
|
||||
|
||||
def getNewRapla(url: str):
|
||||
async def getNewRapla(url: str):
|
||||
"""
|
||||
Speichert den iCal eines Raplas auf dem Server. \n
|
||||
Gibt Namen der Datei zurück. \n
|
||||
@ -50,16 +65,18 @@ def getNewRapla(url: str):
|
||||
|
||||
try:
|
||||
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:
|
||||
return -1
|
||||
|
||||
file = open("calendars/list.json", "r+")
|
||||
jsoncal = json.load(file)
|
||||
jsoncal.update({kurs: ["rapla" + kurs + ".ical", url]})
|
||||
jsoncal.update({kurs: [f"rapla{kurs}.ical", url]})
|
||||
file.close()
|
||||
file = open("calendars/list.json", "w")
|
||||
json.dump(jsoncal, file, indent=4)
|
||||
return "rapla" + kurs + ".ical"
|
||||
return f"rapla{kurs}.ical"
|
||||
|
||||
|
||||
def getIcal(kurs: str):
|
||||
@ -93,13 +110,23 @@ def getRaplas():
|
||||
return sorted(kursl), sorted(filel), sorted(urll)
|
||||
|
||||
|
||||
@scheduler.task("interval", id="refreshRapla", minutes=5)
|
||||
def refreshRapla():
|
||||
async def refreshRapla():
|
||||
"""
|
||||
Aktualisiert alle 5 Minuten alle gespeicherten Raplas.
|
||||
"""
|
||||
filel = getRaplas()[1]
|
||||
urll = getRaplas()[2]
|
||||
for i in range(len(filel)):
|
||||
print("Update Rapla: " + filel[i][:-5])
|
||||
urlretrieve(urll[i], "calendars/" + filel[i])
|
||||
jobl = []
|
||||
async with httpx.AsyncClient() as s:
|
||||
for i in range(len(filel)):
|
||||
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
22
init.py
@ -1,11 +1,12 @@
|
||||
import apscheduler
|
||||
from flask import Flask
|
||||
from flask_login import LoginManager, UserMixin
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from flask_talisman import Talisman
|
||||
|
||||
from get_mysql import get_mysql
|
||||
import atexit
|
||||
from flask_apscheduler import APScheduler
|
||||
from turbo_flask import Turbo
|
||||
|
||||
|
||||
def create():
|
||||
@ -14,7 +15,6 @@ def create():
|
||||
:return app:
|
||||
"""
|
||||
app = Flask(__name__)
|
||||
turbo.init_app(app)
|
||||
dbpw = get_mysql()[1]
|
||||
dbun = get_mysql()[0]
|
||||
|
||||
@ -56,7 +56,17 @@ class Dualis(db.Model):
|
||||
token = db.Column(db.String(255), unique=True)
|
||||
uid = db.Column(db.Integer, primary_key=True)
|
||||
token_created = db.Column(db.Integer)
|
||||
result_list = db.Column(db.String(15))
|
||||
semester = 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):
|
||||
@ -71,11 +81,11 @@ class Meals(db.Model):
|
||||
schwein = db.Column(db.Boolean)
|
||||
|
||||
|
||||
scheduler = APScheduler()
|
||||
turbo = Turbo()
|
||||
scheduler = APScheduler ()
|
||||
app = create()
|
||||
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.start()
|
||||
scheduler.api_enabled = True
|
||||
|
||||
@ -1,3 +1,6 @@
|
||||
from init import Semesterlist, User
|
||||
|
||||
|
||||
def getCookie(cookies):
|
||||
"""
|
||||
Liefert (letzten) Cookie der Cookies-Liste zurück.
|
||||
@ -8,3 +11,16 @@ def getCookie(cookies):
|
||||
for c in cookies:
|
||||
cookie = c.value
|
||||
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()
|
||||
|
||||
@ -1,13 +1,17 @@
|
||||
beautifulsoup4
|
||||
Flask
|
||||
beautifulsoup4~=4.12.2
|
||||
Flask~=3.0.3
|
||||
Flask_APScheduler
|
||||
Flask_Login
|
||||
flask_sqlalchemy
|
||||
icalendar
|
||||
icalendar~=5.0.11
|
||||
recurring_ical_events
|
||||
Requests
|
||||
Requests~=2.31.0
|
||||
talisman
|
||||
Werkzeug
|
||||
Werkzeug~=3.0.0
|
||||
lxml
|
||||
bs4
|
||||
pytz
|
||||
bs4~=0.0.1
|
||||
pytz~=2023.3.post1
|
||||
|
||||
asyncio~=3.4.3
|
||||
httpx~=1.0.0b0
|
||||
celery~=5.4.0rc2
|
||||
65
routing.py
65
routing.py
@ -10,8 +10,9 @@ import time
|
||||
import random
|
||||
|
||||
import fetchDUALIS
|
||||
import fetchMENSA
|
||||
import fetchRAPLA
|
||||
import requesthelpers
|
||||
from requesthelpers import *
|
||||
from fetchRAPLA import *
|
||||
from calendar_generation import getWeek
|
||||
from init import *
|
||||
@ -61,29 +62,33 @@ def welcome():
|
||||
return render_template('dashboard.html', kurs=kurs, name=name, theorie=t, praxis=p)
|
||||
|
||||
|
||||
@app.route("/theorie/noten")
|
||||
@app.route("/theorie/noten", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def displayNoten():
|
||||
async def displayNoten():
|
||||
"""
|
||||
Zeigt die Noten aus Dualis an. Hierfür ist ein aktives Token nötig.
|
||||
:return HTML:
|
||||
"""
|
||||
d = Dualis.query.filter_by(uid=current_user.id).first()
|
||||
if not d.result_list:
|
||||
if request.method == "POST":
|
||||
d.semester = request.form.get("sem")
|
||||
db.session.commit()
|
||||
if not d.semester:
|
||||
return redirect(url_for("getSemester", next=url_for(request.endpoint)))
|
||||
t = d.token
|
||||
sem = d.result_list
|
||||
chosensemester = d.semester
|
||||
c = request.cookies.get("cnsc")
|
||||
timeout = fetchDUALIS.timeOut(d, c, "displayNoten")
|
||||
if timeout:
|
||||
return timeout
|
||||
res = fetchDUALIS.getResults(t, c, sem)
|
||||
return render_template("noten.html", noten=res, semester=fetchDUALIS.getSem(t, c), sel=sem, s="n", praxis="hidden")
|
||||
semester = getSemesterList(current_user.id)
|
||||
noten = await fetchDUALIS.getResults(t, c, chosensemester)
|
||||
return render_template("noten.html", noten=noten, semester=semester, sel=chosensemester, s="n", praxis="hidden")
|
||||
|
||||
|
||||
@app.route("/plan", methods=["GET"])
|
||||
@login_required
|
||||
def displayRapla():
|
||||
async def displayRapla():
|
||||
"""
|
||||
Zeigt den Stundenplan für eingeloggte User an. \n
|
||||
TODO: Persönliche Filter, Notizen, Essensvorlieben etc. berücksichtigen.
|
||||
@ -99,14 +104,14 @@ def displayRapla():
|
||||
samstag = request.args.get("samstag")
|
||||
if not samstag:
|
||||
samstag = False
|
||||
events = getWeek(week, fetchRAPLA.getIcal(current_user.kurs), samstag)
|
||||
events = await getWeek(week, fetchRAPLA.getIcal(current_user.kurs), samstag)
|
||||
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],
|
||||
s="p", praxis="hidden")
|
||||
|
||||
|
||||
@app.route("/plan/<string:kurs>")
|
||||
def displayPlan(kurs):
|
||||
async def displayPlan(kurs):
|
||||
"""
|
||||
Zeigt den Stundenplan ohne Login an. \n
|
||||
Präferenzen werden nicht berücksichtigt.
|
||||
@ -129,7 +134,7 @@ def displayPlan(kurs):
|
||||
samstag = request.args.get("samstag")
|
||||
if not samstag:
|
||||
samstag = False
|
||||
events = getWeek(week, plan, samstag)
|
||||
events = await getWeek(week, plan, samstag)
|
||||
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")
|
||||
else:
|
||||
@ -147,7 +152,7 @@ def redKurs():
|
||||
|
||||
@app.route("/set-up/kurs")
|
||||
@login_required
|
||||
def getKurs():
|
||||
async def getKurs():
|
||||
"""
|
||||
Automatische Kurs-Auswahl. \n
|
||||
Aktives Dualis-Token benötigt.
|
||||
@ -161,7 +166,7 @@ def getKurs():
|
||||
return timeout
|
||||
e = False
|
||||
if not current_user.kurs:
|
||||
kurs = fetchDUALIS.getKurs(d.token, cookie)
|
||||
kurs = await fetchDUALIS.getKurs(d.token, cookie)
|
||||
if kurs != 0:
|
||||
if not fetchRAPLA.getIcal(kurs):
|
||||
return render_template('kurs.html', detected=(kurs, e), s="s", theorie="hidden", praxis="hidden",
|
||||
@ -182,15 +187,23 @@ def getKurs():
|
||||
|
||||
@app.route("/set-up/semester")
|
||||
@login_required
|
||||
def getSemester():
|
||||
async def getSemester():
|
||||
"""
|
||||
Manuelle Semester-Auswahl.
|
||||
:return HTML:
|
||||
"""
|
||||
t = Dualis.query.filter_by(uid=current_user.id).first().token
|
||||
c = request.cookies.get("cnsc")
|
||||
|
||||
return render_template("semester.html", semester=fetchDUALIS.getSem(t, c), s="s", theorie="hidden", praxis="hidden")
|
||||
semesterlist = Semesterlist.query.filter_by(uid=current_user.id).all()
|
||||
if not semesterlist:
|
||||
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"])
|
||||
@ -204,7 +217,7 @@ def setSemester():
|
||||
if not n:
|
||||
n = url_for("welcome")
|
||||
d = Dualis.query.filter_by(uid=current_user.id).first()
|
||||
d.result_list = request.form.get("sem")
|
||||
d.semester = request.form.get("sem")
|
||||
db.session.commit()
|
||||
return redirect(n)
|
||||
|
||||
@ -222,7 +235,7 @@ def chooseRaplas():
|
||||
|
||||
@app.route("/set-up/rapla", methods=["POST"])
|
||||
@login_required
|
||||
def getRapla():
|
||||
async def getRapla():
|
||||
"""
|
||||
Verarbeitet die Eingabe von chooseRaplas().
|
||||
:return HTML:
|
||||
@ -232,10 +245,10 @@ def getRapla():
|
||||
if file == url == "None":
|
||||
return redirect(url_for("chooseRaplas"))
|
||||
if file != "None":
|
||||
User.query.filter_by(id=current_user.id).first().kurs = file[5:-5]
|
||||
loadUser(current_user.id).kurs = file[5:-5]
|
||||
db.session.commit()
|
||||
elif url != "None":
|
||||
file = getNewRapla(url)
|
||||
file = await getNewRapla(url)
|
||||
if type(file) is not int:
|
||||
User.query.filter_by(id=current_user.id).first().kurs = file[5:-5]
|
||||
db.session.commit()
|
||||
@ -254,7 +267,7 @@ def login():
|
||||
|
||||
|
||||
@app.route("/log-in", methods=["POST"])
|
||||
def login_post():
|
||||
async def login_post():
|
||||
"""
|
||||
Verarbeitet die Eingabe von login(). \n
|
||||
Falls der User schon angelegt ist, wird das Passwort verglichen. \n
|
||||
@ -270,18 +283,18 @@ def login_post():
|
||||
success = make_response(redirect(url_for("getKurs")))
|
||||
|
||||
user = User.query.filter_by(email=email).first()
|
||||
t = fetchDUALIS.checkUser(email, password)
|
||||
t = await fetchDUALIS.checkUser(email, password)
|
||||
if t[0] == -2:
|
||||
return redirect(url_for("login", code=-2))
|
||||
if user:
|
||||
dualis = Dualis.query.filter_by(uid=user.id).first()
|
||||
dualis.token = t[0]
|
||||
newcookie = requesthelpers.getCookie(t[1].cookies)
|
||||
newcookie = t[1]
|
||||
dualis.token_created = time.time()
|
||||
db.session.commit()
|
||||
login_user(user)
|
||||
if user.kurs:
|
||||
if not dualis.result_list:
|
||||
if not dualis.semester:
|
||||
success = make_response(redirect(url_for("getSemester")))
|
||||
elif not n:
|
||||
success = make_response(redirect(url_for("welcome")))
|
||||
@ -307,14 +320,14 @@ def login_post():
|
||||
|
||||
@app.route("/log-out")
|
||||
@login_required
|
||||
def logout():
|
||||
async def logout():
|
||||
"""
|
||||
Loggt den User aus.
|
||||
:return Empty Token:
|
||||
"""
|
||||
cookie = request.cookies.get("cnsc")
|
||||
dualis = Dualis.query.filter_by(uid=current_user.id).first()
|
||||
fetchDUALIS.logOut(dualis.token, cookie)
|
||||
await fetchDUALIS.logOut(dualis.token, cookie)
|
||||
dualis.token = None
|
||||
db.session.commit()
|
||||
logout_user()
|
||||
|
||||
@ -11,7 +11,7 @@
|
||||
</turbo-frame>
|
||||
<br>
|
||||
<br>
|
||||
<form id= "dropdown" method="post" action={{ url_for ("setSemester", next=url_for("displayNoten")) }}>
|
||||
<form id= "dropdown" method="post" action={{ url_for ("displayNoten") }}>
|
||||
<label for="sem">Semester wählen! </label>
|
||||
<select name="sem" id="sem">
|
||||
{% for i in range (semester|length) %}
|
||||
|
||||
284
tests_examples/vergleich.py
Normal file
284
tests_examples/vergleich.py
Normal file
@ -0,0 +1,284 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user