394 lines
15 KiB
Python
394 lines
15 KiB
Python
import time
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
import re
|
||
import argparse
|
||
from datetime import datetime
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Argument parsing
|
||
# ---------------------------------------------------------------------------
|
||
|
||
parser = argparse.ArgumentParser(description="CoronaNG Autojoin Script")
|
||
parser.add_argument("--cvid", type=int, required=True, nargs='+', help="One or more cvid values")
|
||
parser.add_argument("--jsessionid", type=str, help="The JSESSIONID cookie value (re-login not supported)")
|
||
parser.add_argument("--user", type=str, help="Username for auto-login")
|
||
parser.add_argument("--pass", dest="password", type=str, help="Password for auto-login")
|
||
parser.add_argument("--interval", type=float, default=30.0,
|
||
help="Long refresh time in seconds after the active window has been running for a while (default: 30)")
|
||
parser.add_argument("--presend", action="store_true",
|
||
help="Start hammering registration requests 5s before the window opens (tiny interval). "
|
||
"Without this flag, wakes up 2s early to avoid missing the window.")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not args.jsessionid and not (args.user and args.password):
|
||
parser.error("Either --jsessionid or both --user and --pass must be provided.")
|
||
if args.jsessionid:
|
||
print("Warning: --jsessionid provided; automatic re-login on session expiry is not supported.")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shared state
|
||
# ---------------------------------------------------------------------------
|
||
|
||
base_headers = {
|
||
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:124.0) Gecko/20100101 Firefox/124.0",
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
|
||
"Accept-Language": "en-GB,en;q=0.5",
|
||
"Accept-Encoding": "gzip, deflate, br",
|
||
"Connection": "keep-alive",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
"Sec-Fetch-Dest": "document",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "same-origin",
|
||
"Sec-Fetch-User": "?1",
|
||
"Pragma": "no-cache",
|
||
"Cache-Control": "no-cache",
|
||
}
|
||
|
||
current_jsessionid = args.jsessionid
|
||
registered_cvids = set()
|
||
|
||
# Seconds before the window opens to wake up
|
||
PRE_START_WINDOW_PRESEND = 5 # --presend: wake at T-5, hammer T-3 to T at 1/s
|
||
PRE_START_WINDOW_NORMAL = 2 # default: wake at T-2, sleep precisely to T
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Auth helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def login(username, password):
|
||
login_url = "https://campusonline.uni-ulm.de/CoronaNG/index.html"
|
||
login_headers = {
|
||
**base_headers,
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Origin": "https://campusonline.uni-ulm.de",
|
||
"Referer": login_url,
|
||
}
|
||
response = requests.post(
|
||
login_url,
|
||
headers=login_headers,
|
||
data={"uid": username, "password": password},
|
||
allow_redirects=False,
|
||
)
|
||
jsessionid = response.cookies.get("JSESSIONID")
|
||
if not jsessionid:
|
||
match = re.search(r"JSESSIONID=([A-Fa-f0-9]+)", response.headers.get("Set-Cookie", ""))
|
||
if match:
|
||
jsessionid = match.group(1)
|
||
return jsessionid
|
||
|
||
|
||
def is_logged_out(response):
|
||
return 'name="uid"' in response.text or 'name="password"' in response.text
|
||
|
||
|
||
def ensure_logged_in():
|
||
global current_jsessionid
|
||
if args.user and args.password:
|
||
print("Session expired — logging in again...")
|
||
jsessionid = login(args.user, args.password)
|
||
if jsessionid:
|
||
current_jsessionid = jsessionid
|
||
print("Re-login successful. Waiting 1s...")
|
||
time.sleep(1)
|
||
else:
|
||
print("Re-login failed: JSESSIONID not found in response.")
|
||
else:
|
||
print("Session appears expired and no credentials provided for re-login.")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parsing helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_DATE_RE = re.compile(r'(\d{2}\.\d{2}\.\d{4})\s+(\d{2}:\d{2})')
|
||
|
||
def _parse_date(td):
|
||
m = _DATE_RE.search(td.get_text())
|
||
if m:
|
||
return datetime.strptime(f"{m.group(1)} {m.group(2)}", "%d.%m.%Y %H:%M")
|
||
return None
|
||
|
||
def parse_anmeldezeit(soup):
|
||
"""Return (start_dt, end_dt) from the Anmeldezeiten table, or (None, None)."""
|
||
caption = soup.find("caption", string=re.compile(r"Anmeldezeiten"))
|
||
if not caption:
|
||
return None, None
|
||
table = caption.find_parent("table")
|
||
row = table.find("tr", class_="bgo")
|
||
if not row:
|
||
return None, None
|
||
tds = row.find_all("td")
|
||
if len(tds) < 2:
|
||
return None, None
|
||
return _parse_date(tds[0]), _parse_date(tds[1])
|
||
|
||
|
||
def get_participant_info(soup):
|
||
"""Return (current, max) participants, or (None, None)."""
|
||
for tr in soup.find_all("tr", class_="dbu"):
|
||
th = tr.find("th")
|
||
if th and "Max. Teilnehmer" in th.text:
|
||
td = tr.find("td")
|
||
match = re.search(r"(\d+) \(aktuell (\d+)\)", td.text.strip())
|
||
if match:
|
||
return int(match.group(2)), int(match.group(1)) # current, max
|
||
return None, None
|
||
|
||
|
||
def get_course_name(soup):
|
||
"""Return the course name from the Vorlesungsverzeichnis table, or None."""
|
||
caption = soup.find("caption", string=re.compile(r"Veranstaltung aus dem Vorlesungsverzeichnis"))
|
||
if not caption:
|
||
return None
|
||
table = caption.find_parent("table")
|
||
for tr in table.find_all("tr"):
|
||
th = tr.find("th")
|
||
if th and th.get_text(strip=True) == "Name":
|
||
td = tr.find("td")
|
||
if td:
|
||
return td.get_text(strip=True)
|
||
return None
|
||
|
||
|
||
def is_registered(soup):
|
||
"""Return True if already registered (person.gif appears twice — once for observer, once for participant)."""
|
||
return len(soup.find_all("img", src=lambda s: s and "person.gif" in s)) >= 2
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Registration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def registration_interval(elapsed_sec):
|
||
"""Sleep interval (seconds) based on elapsed time since the window opened."""
|
||
if elapsed_sec < 5:
|
||
return 0.5
|
||
elif elapsed_sec < 15: # 5–15 s
|
||
return 1.0
|
||
elif elapsed_sec < 45: # 15–45 s
|
||
return 3.0
|
||
elif elapsed_sec < 105: # 45 s – 1 m 45 s
|
||
return 5.0
|
||
else:
|
||
return args.interval
|
||
|
||
|
||
def try_register(cvid):
|
||
"""POST to register directly (no preceding GET). Returns True if confirmed via person.gif."""
|
||
post_url = "https://campusonline.uni-ulm.de/CoronaNG/user/userDetails.html"
|
||
referer = f"https://campusonline.uni-ulm.de/CoronaNG/user/userDetails.html?id={cvid}"
|
||
payload = f"id={cvid}&command=participate"
|
||
post_headers = {
|
||
**base_headers,
|
||
"Cookie": f"JSESSIONID={current_jsessionid}",
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Content-Length": str(len(payload)),
|
||
"Origin": "https://campusonline.uni-ulm.de",
|
||
"Referer": referer,
|
||
}
|
||
try:
|
||
response = requests.post(post_url, headers=post_headers, data=payload)
|
||
if response.status_code != 200:
|
||
return False
|
||
soup = BeautifulSoup(response.content, "html.parser")
|
||
return is_registered(soup)
|
||
except requests.exceptions.RequestException as e:
|
||
print(f" POST error: {e}")
|
||
return False
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Initial login
|
||
# ---------------------------------------------------------------------------
|
||
|
||
if args.user and args.password:
|
||
print("Logging in...")
|
||
current_jsessionid = login(args.user, args.password)
|
||
if not current_jsessionid:
|
||
print("Initial login failed.")
|
||
exit(1)
|
||
print("Login successful. Waiting 1s...")
|
||
time.sleep(1)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parallel fetch
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def fetch_course(cvid, jsessionid):
|
||
"""Fetch a single course page. Returns a result dict."""
|
||
url = f"https://campusonline.uni-ulm.de/CoronaNG/user/userDetails.html?id={cvid}"
|
||
headers = {**base_headers, "Cookie": f"JSESSIONID={jsessionid}", "Referer": url}
|
||
try:
|
||
response = requests.get(url, headers=headers)
|
||
if response.status_code == 403:
|
||
return {"cvid": cvid, "logged_out": True, "headers": headers}
|
||
response.raise_for_status()
|
||
fetched_at = datetime.now()
|
||
if is_logged_out(response):
|
||
return {"cvid": cvid, "logged_out": True, "headers": headers}
|
||
soup = BeautifulSoup(response.content, "html.parser")
|
||
|
||
error_span = soup.find("span", class_="Error")
|
||
if error_span:
|
||
print(f" cvid={cvid}: Server error — {error_span.get_text(strip=True)}")
|
||
return {"cvid": cvid, "error": error_span.get_text(strip=True)}
|
||
|
||
list_table = soup.find("div", class_="listTable")
|
||
if not list_table:
|
||
print(f" cvid={cvid}: WARNING — listTable not found, HTML parsing may have failed.")
|
||
print(soup.prettify())
|
||
|
||
start_dt, end_dt = parse_anmeldezeit(soup)
|
||
current_p, max_p = get_participant_info(soup)
|
||
registered = is_registered(soup)
|
||
name = get_course_name(soup)
|
||
return {
|
||
"cvid": cvid,
|
||
"logged_out": False,
|
||
"headers": headers,
|
||
"fetched_at": fetched_at,
|
||
"start_dt": start_dt,
|
||
"end_dt": end_dt,
|
||
"current_p": current_p,
|
||
"max_p": max_p,
|
||
"registered": registered,
|
||
"name": name,
|
||
}
|
||
except requests.exceptions.RequestException as e:
|
||
return {"cvid": cvid, "error": e}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Main loop
|
||
# ---------------------------------------------------------------------------
|
||
|
||
while True:
|
||
iteration_start = datetime.now()
|
||
min_sleep = 300.0
|
||
logged_out_detected = False
|
||
to_register = [] # cvids with open spots
|
||
remaining_cvids = []
|
||
|
||
if len(registered_cvids) == len(args.cvid):
|
||
print("All courses are registered. Exiting.")
|
||
exit(0)
|
||
|
||
for cvid in args.cvid:
|
||
if cvid in registered_cvids:
|
||
print(f" cvid={cvid}: Already registered!")
|
||
else:
|
||
remaining_cvids.append(cvid)
|
||
|
||
# Fetch all course pages in parallel
|
||
with ThreadPoolExecutor(max_workers=len(remaining_cvids)) as executor:
|
||
futures = {executor.submit(fetch_course, cvid, current_jsessionid): cvid
|
||
for cvid in remaining_cvids}
|
||
results = [future.result() for future in as_completed(futures)]
|
||
|
||
for result in results:
|
||
cvid = result["cvid"]
|
||
|
||
if "error" in result:
|
||
print(f" cvid={cvid}: GET error: {result['error']}")
|
||
continue
|
||
|
||
if result["logged_out"]:
|
||
logged_out_detected = True
|
||
continue
|
||
|
||
if result["registered"]:
|
||
print(f" cvid={cvid}: Already registered!")
|
||
registered_cvids.add(cvid)
|
||
continue
|
||
|
||
now = result["fetched_at"]
|
||
start_dt = result["start_dt"]
|
||
end_dt = result["end_dt"]
|
||
current_p = result["current_p"]
|
||
max_p = result["max_p"]
|
||
headers = result["headers"]
|
||
name = result.get("name") or "?"
|
||
print(f" cvid={cvid}: {name}")
|
||
|
||
# ---- No Anmeldezeiten found --------------------------------------------
|
||
if start_dt is None:
|
||
print(f" cvid={cvid}: No Anmeldezeit found — polling every 5 min.")
|
||
continue
|
||
|
||
# ---- Registration window already closed --------------------------------
|
||
if now > end_dt:
|
||
print(f" cvid={cvid}: Anmeldezeit expired ({end_dt:%d.%m.%Y %H:%M}). Skipping.")
|
||
continue
|
||
|
||
secs_to_start = (start_dt - now).total_seconds()
|
||
pre_window = PRE_START_WINDOW_PRESEND if args.presend else PRE_START_WINDOW_NORMAL
|
||
|
||
# ---- Far from start: sleep until pre-window ----------------------------
|
||
if secs_to_start > pre_window:
|
||
sleep_needed = min(secs_to_start - pre_window, 300.0)
|
||
print(f" cvid={cvid}: Opens {start_dt:%d.%m.%Y %H:%M} "
|
||
f"(in {secs_to_start:.0f}s). Next check in {sleep_needed:.0f}s.")
|
||
min_sleep = min(min_sleep, sleep_needed)
|
||
continue
|
||
|
||
# ---- Pre-start window --------------------------------------------------
|
||
if secs_to_start > 0:
|
||
if args.presend and secs_to_start > 3.0:
|
||
# T-5 to T-3: sleep precisely until T-3, then start hammering
|
||
interval = secs_to_start - 3.0
|
||
print(f" cvid={cvid}: Pre-start ({secs_to_start:.1f}s to go) — sleeping {interval:.2f}s to T-3.")
|
||
elif args.presend:
|
||
# T-3 to T: hammer at 1 req/s
|
||
interval = 1.0
|
||
print(f" cvid={cvid}: Pre-start ({secs_to_start:.1f}s to go) — hammering at 1 req/s.")
|
||
else:
|
||
# T-2 to T: sleep precisely until T
|
||
interval = secs_to_start
|
||
print(f" cvid={cvid}: Pre-start ({secs_to_start:.2f}s to go) — sleeping to T.")
|
||
# ---- Active window: T to end_dt ----------------------------------------
|
||
else:
|
||
elapsed = -secs_to_start
|
||
interval = registration_interval(elapsed)
|
||
print(f" cvid={cvid}: Active (+{elapsed:.1f}s) — interval {interval}s.")
|
||
|
||
min_sleep = min(min_sleep, interval)
|
||
|
||
if current_p is not None and max_p is not None:
|
||
print(f" Participants: {current_p}/{max_p}")
|
||
if current_p < max_p:
|
||
to_register.append(cvid)
|
||
else:
|
||
print(f" No spots available.")
|
||
else:
|
||
print(f" Could not read participant info.")
|
||
|
||
# Attempt registration in parallel
|
||
if to_register:
|
||
with ThreadPoolExecutor(max_workers=len(to_register)) as executor:
|
||
futures = {executor.submit(try_register, cvid): cvid for cvid in to_register}
|
||
for future in as_completed(futures):
|
||
cvid = futures[future]
|
||
print(f" cvid={cvid}: Spot available! Registering...")
|
||
if future.result():
|
||
print(f"Successfully registered for cvid={cvid}!")
|
||
registered_cvids.add(cvid)
|
||
if len(registered_cvids) == len(args.cvid):
|
||
print("All courses are registered. Exiting.")
|
||
exit(0)
|
||
else:
|
||
print(f" cvid={cvid}: Registration attempt failed.")
|
||
|
||
if len(registered_cvids) == len(args.cvid):
|
||
print("All courses are registered. Exiting.")
|
||
exit(0)
|
||
|
||
if logged_out_detected:
|
||
ensure_logged_in()
|
||
min_sleep = 1.0
|
||
|
||
# Sleep until the computed wake-up time, accounting for time already spent this iteration
|
||
wake_at = iteration_start.timestamp() + min_sleep
|
||
sleep_sec = max(wake_at - time.time(), 0.05)
|
||
if sleep_sec >= 1.0:
|
||
print(f'[{time.strftime("%H:%M:%S")}] Sleeping {sleep_sec:.1f}s...')
|
||
time.sleep(sleep_sec)
|