SoftwareEngineering/DualHub
4
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

async test

Browse Source
This commit is contained in:
Paul Martin
2024-04-08 20:48:34 +02:00
parent 56ec2cfb15
commit 211ec18887
9 changed files with 344 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
fetchDUALIS.py
View File

@ -4,6 +4,8 @@ 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,28 +18,29 @@ 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):

42
fetchMENSA.py
View File

@ -1,5 +1,5 @@
import json
from init import db, Meals, scheduler
from init import db, Meals#, scheduler
import datetime
import requests
import time
@ -110,29 +110,29 @@ def formatDay(day: datetime):
return day
@scheduler.task('cron', id="refreshMeals", hour='8-11', day_of_week='*', minute='15', week='*', second='30')
#@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
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)
#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)

4
fetchRAPLA.py
View File

@ -3,7 +3,7 @@ from urllib.request import urlretrieve
import icalendar
import json
import recurring_ical_events
from init import scheduler
#from init import scheduler
def parseURL(url: str):
@ -93,7 +93,7 @@ def getRaplas():
return sorted(kursl), sorted(filel), sorted(urll)
@scheduler.task("interval", id="refreshRapla", minutes=5)
#@scheduler.task("interval", id="refreshRapla", minutes=5)
def refreshRapla():
"""
Aktualisiert alle 5 Minuten alle gespeicherten Raplas.

12
init.py
View File

@ -4,7 +4,7 @@ from flask_sqlalchemy import SQLAlchemy
from flask_talisman import Talisman
from get_mysql import get_mysql
import atexit
from flask_apscheduler import APScheduler
#from flask_apscheduler import APScheduler
def create():
@ -25,7 +25,7 @@ def create():
login_manager.login_view = "login"
# Shut down the scheduler when exiting the app
atexit.register(lambda: scheduler.shutdown())
#atexit.register(lambda: scheduler.shutdown())
@login_manager.user_loader
def load_user(uid: int):
@ -69,11 +69,11 @@ class Meals(db.Model):
schwein = db.Column(db.Boolean)
scheduler = APScheduler()
#scheduler = APScheduler()
app = create()
def_src = ["*.paulmartin.cloud", '\'self\'']
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
#scheduler.init_app(app)
#scheduler.start()
#scheduler.api_enabled = True

17
requirements.txt
View File

@ -1,13 +1,16 @@
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

14
routing.py
View File

@ -49,7 +49,7 @@ 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():
"""
@ -57,6 +57,9 @@ def displayNoten():
:return HTML:
"""
d = Dualis.query.filter_by(uid=current_user.id).first()
if request.method == "POST":
d.result_list = request.form.get("sem")
db.session.commit()
if not d.result_list:
return redirect(url_for("getSemester", next=url_for(request.endpoint)))
t = d.token
@ -65,8 +68,7 @@ def displayNoten():
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")
return render_template("noten.html", noten=fetchDUALIS.getResults(t, c, sem), semester=fetchDUALIS.getSem(t, c), sel=sem, s="n", praxis="hidden")
@app.route("/plan", methods=["GET"])
@ -242,7 +244,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
@ -258,13 +260,13 @@ 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)

2
templates/noten.html
View File

@ -27,7 +27,7 @@
{% endfor %}
<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) %}

0
testpages.py → tests_examples/testpages.py
View File

277
tests_examples/vergleich.py Normal file
View File

@ -0,0 +1,277 @@
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
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
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
async def checkUser_async():
"""
Erhält von Dualis den Token und Cookie für User.
:param email:
:param password:
:return (Token, Session):
"""
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
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
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
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]
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
fetchDUALIS.logOut(n[0], n[1])
start = time.perf_counter()
a = await async_normal()
end = time.perf_counter()
asynctimeloop = end - start
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)