feat: working coronang

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uulm-utils"
-version = "0.1.0"
+version = "1.0"
 description = "Collection of helpers for Ulm University"
 readme = "README.md"
 requires-python = ">=3.12"

src/uulm_utils/main.py

@@ -1,6 +1,6 @@
 import asyncclick as click
 import questionary
-from playwright.async_api import async_playwright, Playwright
+from playwright.async_api import async_playwright, Playwright, expect
 import pandas as pd
 from dotenv import load_dotenv
 
@@ -9,7 +9,7 @@ from enum import Enum
 import asyncio
 from time import sleep
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta
 
 
 load_dotenv()  # take environment variables
@@ -54,6 +54,8 @@ async def cli(ctx, username, password, headful, debug):
     if(debug): logger.setLevel(logging.DEBUG)
     ctx.ensure_object(dict)
     ctx.obj['HEADLESS'] = not headful
+    ctx.obj['USERNAME'] = username
+    ctx.obj['PASSWORD'] = password
 
 @cli.command()
 @click.pass_context
@@ -73,7 +75,7 @@ async def campusonline(ctx):
         selection = await selection_or_walk(options)
         print(selection)
         sleep(2)
-        raise NotImplementedError
+    return
 
 @cli.command()
 @click.argument('target_times', nargs=-1, type=click.DateTime( ['%H:%M:%S']), required=True)
@@ -85,31 +87,41 @@ async def coronang(ctx, target_times, before):
     Please beware that CoronaNG only allows one active session at all times.
     '''
     CORONANG_VERSION='v1.8.00'
+    CORONANG_URL="https://campusonline.uni-ulm.de/CoronaNG/user/mycorona.html"
     logger.info('Parsed input times as %s', target_times)
+    before_seconds = timedelta(seconds=before)
+    target_times = sorted(list(target_times))
     async for page, browser, context in run_playwright(ctx.obj['HEADLESS']):
-        await page.goto("https://campusonline.uni-ulm.de/CoronaNG/user/mycorona.html")
+        await page.goto(CORONANG_URL)
         server_version = await page.locator("css=#mblock_innen > a:nth-child(1)").inner_text()
         if(server_version != CORONANG_VERSION):
             logger.warning('Read CoronaNG version %s. Last tested version is %s. Please use --headful flag to ensure that everything is working.',
                            server_version, CORONANG_VERSION)
 
+        # iterate over staggered login
         for target_time in target_times:
+            # wait for execution
             while True:
                 server_str = await page.locator("css=#mblock_innen").inner_text()
                 server_time = datetime.strptime(server_str.split().pop(), "%H:%M:%S")
-                break
+                dtime = target_time -server_time
-
+                dtime_before = dtime - before_seconds
-        raise NotImplementedError
+                logger.debug('Server Time: %s, delta: %s', server_time, dtime_before)
-        await page.locator("input[name=\"uid\"]").click()
+                # execute
-        await page.locator("input[name=\"uid\"]").fill(ctx.obj['USERNAME'])
+                if dtime_before < timedelta(0):
-        await page.locator("input[name=\"password\"]").click()
+                    # login necessary?
-        await page.locator("input[name=\"password\"]").fill(ctx.obj['PASSWORD'])
+                    if (await page.locator("input[name=\"uid\"]").count()) >0:
-        await page.get_by_role("button", name="Anmelden").click()
+                        await page.locator("input[name=\"uid\"]").click()
-        await page.get_by_role("link", name="Beobachtungen & Teilnahmen").click()
+                        await page.locator("input[name=\"uid\"]").fill(ctx.obj['USERNAME'])
-        await page.get_by_role("table", name="Ihre Beobachtungen. Sie kö").get_by_role("button").click()
+                        await page.locator("input[name=\"password\"]").click()
-        await page.get_by_role("table", name="Ihre Beobachtungen. Sie kö").get_by_role("combobox").select_option("5")
+                        await page.locator("input[name=\"password\"]").fill(ctx.obj['PASSWORD'])
-        await page.get_by_role("cell", name="An Markierten teilnehmen Ausf").get_by_role("button").click()
+                        await page.get_by_role("button", name="Anmelden").click()
-        await page.reload()
+                        await page.goto(CORONANG_URL)
+                        await page.get_by_role("table", name="Ihre Beobachtungen. Sie kö").get_by_role("button").click()
+                        await page.get_by_role("table", name="Ihre Beobachtungen. Sie kö").get_by_role("combobox").select_option("5")
+                        await page.get_by_role("cell", name="An Markierten teilnehmen Ausf").get_by_role("button").click()
+                    await page.reload()
+    return
 
 @cli.command()
 @click.argument('target_times', nargs=-1, type=click.DateTime( ["%H:%M:%S"]), required=True)
@@ -124,9 +136,11 @@ async def sport(ctx, target_times, target_course, before):
     print(target_course)
     # TODO Check Version in HTML Head
     logger.info('Parsed input times as %s', target_times)
+    before_seconds = timedelta(seconds=before)
+    target_times = sorted(list(target_times))
     async for page, browser, context in run_playwright(ctx.obj['HEADLESS']):
         pass
-    raise NotImplementedError
+    return
 
 @cli.command()
 @click.argument('filename', type=click.Path(exists=True))
@@ -152,7 +166,7 @@ def grades(filename, target_lp:int):
         acc_note = acc_note + weight * row['grade']
         logger.debug('Added "%s" with %d/%d credits and grade %.1f', row['name'], weight, row['credits'], row['grade'])
     acc_note: float = acc_note / acc_lp
-    print(f'Final Grade: {acc_note}')
+    print(f'Final Grade: {acc_note} with {acc_lp}/{target_lp} credits')
 
 if __name__ == "__main__":
     asyncio.run(cli.main())