288 lines
9.3 KiB
Python
288 lines
9.3 KiB
Python
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 password, 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(password, 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:
|
|
# noinspection DuplicatedCode
|
|
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)
|
|
# noinspection DuplicatedCode
|
|
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]]
|
|
# noinspection DuplicatedCode
|
|
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)
|