Yandrik/uulm-coronang-autojoin
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: good script and docker

Browse Source
This commit is contained in:
Yandrik
2026-04-13 18:23:46 +02:00
parent 90c5a42b47
commit 49b43b9598
7 changed files with 583 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

442
main.py
View File

@ -3,115 +3,391 @@ import requests
from bs4 import BeautifulSoup
import re
import argparse
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
# ---------------------------------------------------------------------------
# Argument parsing
# ---------------------------------------------------------------------------
# Create an ArgumentParser object
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.")
# Add arguments for cvid and jsessionid
parser.add_argument("--cvid", type=int, required=True, help="The cvid value")
parser.add_argument("--jsessionid", type=str, required=True, help="The JSESSIONID cookie value")
# Parse the command-line arguments
args = parser.parse_args()
# URL to scrape
url = f"https://campusonline.uni-ulm.de/CoronaNG/user/userDetails.html?cvid={args.cvid}"
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.")
# Cookie to include in the request
cookie = f"JSESSIONID={args.jsessionid}"
# ---------------------------------------------------------------------------
# Shared state
# ---------------------------------------------------------------------------
# Headers for the GET request
headers = {
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",
"Referer": url,
"Connection": "keep-alive",
"Cookie": cookie,
"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"
"Cache-Control": "no-cache",
}
def make_post_request(url, headers, payload, max_retries=3):
retries = 0
while retries < max_retries:
try:
response = requests.post(url, headers=headers, data=payload)
response.raise_for_status() # Raise an exception for 4xx or 5xx status codes
return response
except requests.exceptions.RequestException as e:
print(f"Error occurred during POST request: {str(e)}")
retries += 1
if retries < max_retries:
print(f"Retrying in 5 seconds... (Attempt {retries+1}/{max_retries})")
time.sleep(5)
else:
print("Max retries reached. Exiting.")
exit(1)
current_jsessionid = args.jsessionid
registered_cvids = set()
while True:
# 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:
print("Scraping...")
# Send GET request to the URL with the specified headers
response = requests.get(url, headers=headers)
response.raise_for_status() # Raise an exception for 4xx or 5xx status codes
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
# Parse the HTML content using BeautifulSoup
# ---------------------------------------------------------------------------
# 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")
# Find the <tr> element with class "dbu"
tr_elements = soup.find_all("tr", class_="dbu")
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)}
for tr_element in tr_elements:
th_element = tr_element.find("th")
if th_element and "Max. Teilnehmer" in th_element.text:
# Extract the max and current participant numbers
td_element = tr_element.find("td")
participant_info = td_element.text.strip()
# regex to find the numbers in a string like "10 (aktuell 10)"
regex = r"(\d+) \(aktuell (\d+)\)"
match = re.search(regex, participant_info)
if match:
max_participants = int(match.group(1))
current_participants = int(match.group(2))
print("Max participants:", max_participants, "; Current participants:", current_participants)
else:
print("Failed to parse participant info:", participant_info)
continue
# Check if there is a free spot
if current_participants < max_participants:
# Send POST request to participate
post_url = "https://campusonline.uni-ulm.de/CoronaNG/user/userDetails.html"
payload = f"id={args.cvid}&command=participate"
post_headers = headers.copy()
post_headers["Content-Type"] = "application/x-www-form-urlencoded"
post_headers["Content-Length"] = str(len(payload))
post_headers["Origin"] = "https://campusonline.uni-ulm.de"
post_response = make_post_request(post_url, post_headers, payload)
if post_response.status_code == 200:
print("Successfully participated!")
exit(0)
else:
print("Failed to participate. Status code:", post_response.status_code)
exit(1)
else:
print("No free spots available.")
break
else:
print("Participant information not found.")
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:
print(f"Error occurred during GET request: {str(e)}")
return {"cvid": cvid, "error": e}
print(f'Current Time: {time.strftime("%H:%M:%S")}. Sleeping for 30 seconds...')
time.sleep(30)
# ---------------------------------------------------------------------------
# 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)