|
#!/usr/bin/env python3 |
|
"""Resolve a Moses Studiengang and fetch its module tree. |
|
|
|
Requirements: |
|
pip install beautifulsoup4 |
|
|
|
Examples: |
|
python3 get_studiengang_modules.py 180 --pretty |
|
python3 get_studiengang_modules.py "Computer Engineering" --pretty |
|
python3 get_studiengang_modules.py "Architektur" --degree "Master of Science" --pretty |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import sys |
|
import time |
|
import xml.etree.ElementTree as ET |
|
from dataclasses import dataclass, field |
|
from http.cookiejar import CookieJar |
|
from typing import Iterable |
|
from urllib.error import HTTPError |
|
from urllib.parse import parse_qs, urlencode, urljoin, urlparse |
|
from urllib.request import HTTPCookieProcessor, Request, build_opener |
|
|
|
from bs4 import BeautifulSoup |
|
|
|
SEARCH_URL = "https://moseskonto.tu-berlin.de/moses/modultransfersystem/studiengaenge/suchen.html" |
|
DETAIL_URL = "https://moseskonto.tu-berlin.de/moses/modultransfersystem/studiengaenge/anzeigen.html?studiengang={studiengang_id}" |
|
USER_AGENT = "Mozilla/5.0" |
|
|
|
|
|
class ResolutionError(RuntimeError): |
|
pass |
|
|
|
|
|
class ScrapeError(RuntimeError): |
|
pass |
|
|
|
|
|
REQUEST_SPACING_SECONDS = 0.35 |
|
MAX_RETRIES = 4 |
|
|
|
|
|
class MosesClient: |
|
def __init__(self) -> None: |
|
self.opener = build_opener(HTTPCookieProcessor(CookieJar())) |
|
|
|
def get(self, url: str, headers: dict[str, str] | None = None) -> str: |
|
request = Request(url, headers=headers or {}) |
|
return self.opener.open(request).read().decode("utf-8", errors="replace") |
|
|
|
def post(self, url: str, payload: dict[str, str], headers: dict[str, str] | None = None) -> str: |
|
request = Request( |
|
url, |
|
data=urlencode(payload).encode("utf-8"), |
|
headers=headers or {"Content-Type": "application/x-www-form-urlencoded"}, |
|
) |
|
return self.opener.open(request).read().decode("utf-8", errors="replace") |
|
|
|
|
|
def normalize(text: str) -> str: |
|
return " ".join(text.split()) |
|
|
|
|
|
|
|
def normalize_key(text: str) -> str: |
|
return normalize(text).casefold() |
|
|
|
|
|
|
|
def extract_studiengang_id(href: str | None) -> str | None: |
|
if not href: |
|
return None |
|
parsed = urlparse(href) |
|
values = parse_qs(parsed.query).get("studiengang") |
|
return values[0] if values else None |
|
|
|
|
|
|
|
def fetch_program_results(client: MosesClient) -> list[dict[str, str | None]]: |
|
initial_html = client.get(SEARCH_URL) |
|
soup = BeautifulSoup(initial_html, "html.parser") |
|
form = soup.find("form", id="j_idt56") |
|
if form is None: |
|
raise ScrapeError("Could not find the Studiengänge search form") |
|
|
|
action = urljoin(SEARCH_URL, form.get("action", SEARCH_URL)) |
|
view_state = form.find("input", {"name": "javax.faces.ViewState"}) |
|
client_window = form.find("input", {"name": "javax.faces.ClientWindow"}) |
|
if view_state is None or client_window is None: |
|
raise ScrapeError("Could not extract JSF state fields from the search form") |
|
|
|
payload = { |
|
"j_idt56": "j_idt56", |
|
"j_idt56:j_idt58": "", |
|
"j_idt56:j_idt60": "", |
|
"j_idt56:j_idt64": "", |
|
"j_idt56:j_idt70": "j_idt56:j_idt70", |
|
"javax.faces.ViewState": view_state.get("value", ""), |
|
"javax.faces.ClientWindow": client_window.get("value", ""), |
|
} |
|
results_html = client.post(action, payload) |
|
results_soup = BeautifulSoup(results_html, "html.parser") |
|
table = next( |
|
( |
|
table |
|
for table in results_soup.find_all("table") |
|
if table.find("th") and "Kurzname" in table.get_text(" ", strip=True) |
|
), |
|
None, |
|
) |
|
if table is None: |
|
raise ScrapeError("Could not find the Studiengänge result table") |
|
|
|
programs: list[dict[str, str | None]] = [] |
|
for row in table.find_all("tr")[1:]: |
|
cells = row.find_all("td") |
|
if len(cells) < 4: |
|
continue |
|
link = cells[0].find("a", href=True) |
|
href = urljoin(SEARCH_URL, link["href"]) if link else None |
|
programs.append( |
|
{ |
|
"studiengang_id": extract_studiengang_id(href), |
|
"name": normalize(cells[0].get_text(" ", strip=True)), |
|
"short": normalize(cells[1].get_text(" ", strip=True)), |
|
"degree": normalize(cells[2].get_text(" ", strip=True)), |
|
"provider": normalize(cells[3].get_text(" ", strip=True)), |
|
"href": href, |
|
} |
|
) |
|
return programs |
|
|
|
|
|
|
|
def filter_programs( |
|
programs: Iterable[dict[str, str | None]], |
|
*, |
|
degree: str | None, |
|
provider: str | None, |
|
short: str | None, |
|
) -> list[dict[str, str | None]]: |
|
filtered = list(programs) |
|
if degree: |
|
degree_key = normalize_key(degree) |
|
filtered = [program for program in filtered if normalize_key(program.get("degree") or "") == degree_key] |
|
if provider: |
|
provider_key = normalize_key(provider) |
|
filtered = [program for program in filtered if normalize_key(program.get("provider") or "") == provider_key] |
|
if short: |
|
short_key = normalize_key(short) |
|
filtered = [program for program in filtered if normalize_key(program.get("short") or "") == short_key] |
|
return filtered |
|
|
|
|
|
|
|
def rank_matches(query: str, programs: Iterable[dict[str, str | None]]) -> list[tuple[int, dict[str, str | None]]]: |
|
query_key = normalize_key(query) |
|
ranked: list[tuple[int, dict[str, str | None]]] = [] |
|
for program in programs: |
|
studiengang_id = program.get("studiengang_id") or "" |
|
name = program.get("name") or "" |
|
short = program.get("short") or "" |
|
name_key = normalize_key(name) |
|
short_key = normalize_key(short) |
|
combined_key = normalize_key(" ".join([studiengang_id, name, short])) |
|
|
|
score = -1 |
|
if query.isdigit() and studiengang_id == query: |
|
score = 100 |
|
elif name_key == query_key: |
|
score = 90 |
|
elif short_key == query_key: |
|
score = 85 |
|
elif query_key and query_key in name_key: |
|
score = 70 |
|
elif query_key and query_key in short_key: |
|
score = 65 |
|
elif query_key and query_key in combined_key: |
|
score = 50 |
|
|
|
if score >= 0: |
|
ranked.append((score, program)) |
|
|
|
return sorted( |
|
ranked, |
|
key=lambda item: ( |
|
-item[0], |
|
normalize_key(item[1].get("name") or ""), |
|
normalize_key(item[1].get("degree") or ""), |
|
normalize_key(item[1].get("provider") or ""), |
|
), |
|
) |
|
|
|
|
|
|
|
def describe_program(program: dict[str, str | None]) -> str: |
|
return ( |
|
f"id={program.get('studiengang_id')} | {program.get('name')} | {program.get('short')} | " |
|
f"{program.get('degree')} | {program.get('provider')}" |
|
) |
|
|
|
|
|
|
|
def resolve_program( |
|
client: MosesClient, |
|
query: str, |
|
*, |
|
degree: str | None, |
|
provider: str | None, |
|
short: str | None, |
|
) -> dict[str, str | None]: |
|
programs = fetch_program_results(client) |
|
filtered = filter_programs(programs, degree=degree, provider=provider, short=short) |
|
if not filtered: |
|
raise ResolutionError("No Studiengänge remain after applying the provided filters") |
|
|
|
ranked = rank_matches(query, filtered) |
|
if not ranked: |
|
raise ResolutionError(f"No Studiengang matches query: {query!r}") |
|
|
|
best_score = ranked[0][0] |
|
best_matches = [program for score, program in ranked if score == best_score] |
|
if len(best_matches) != 1: |
|
candidates = "\n".join(f"- {describe_program(program)}" for program in best_matches[:20]) |
|
raise ResolutionError( |
|
"Studiengang resolution is ambiguous. Narrow it down with --degree, --provider, or --short.\n" |
|
f"Top matches:\n{candidates}" |
|
) |
|
|
|
program = dict(best_matches[0]) |
|
detail_url = DETAIL_URL.format(studiengang_id=program["studiengang_id"]) |
|
detail_html = client.get(detail_url, headers={"User-Agent": USER_AGENT, "Referer": detail_url}) |
|
soup = BeautifulSoup(detail_html, "html.parser") |
|
stupo_select = soup.find("select", id="j_idt57:j_idt116") |
|
semester_select = soup.find("select", id="j_idt57:j_idt120") |
|
selected_stupo = stupo_select.find("option", selected=True) if stupo_select else None |
|
selected_semester = semester_select.find("option", selected=True) if semester_select else None |
|
mkg = selected_stupo.get("value") if selected_stupo else None |
|
semester = selected_semester.get("value") if selected_semester else None |
|
|
|
program.update( |
|
{ |
|
"detail_url": detail_url, |
|
"resolved_url": f"{detail_url}&mkg={mkg}&semester={semester}" if mkg and semester else detail_url, |
|
"mkg": mkg, |
|
"mkg_label": normalize(selected_stupo.get_text(" ", strip=True)) if selected_stupo else None, |
|
"semester": semester, |
|
"semester_label": normalize(selected_semester.get_text(" ", strip=True)) if selected_semester else None, |
|
"page_title": normalize(soup.title.get_text(" ", strip=True)) if soup.title else None, |
|
} |
|
) |
|
return program |
|
|
|
|
|
@dataclass |
|
class TreeSession: |
|
client: MosesClient |
|
base_url: str |
|
action_url: str |
|
form_id: str |
|
tree_id: str |
|
details_id: str |
|
view_state: str |
|
client_window: str |
|
last_request_at: float = field(default=0.0) |
|
|
|
@classmethod |
|
def from_program(cls, client: MosesClient, program: dict[str, str | None]) -> "TreeSession": |
|
base_url = program.get("resolved_url") or program.get("detail_url") |
|
if not base_url: |
|
raise ScrapeError("Resolved program does not contain a detail URL") |
|
|
|
html = client.get(base_url, headers={"User-Agent": USER_AGENT, "Referer": base_url}) |
|
soup = BeautifulSoup(html, "html.parser") |
|
form = soup.find("form") |
|
if form is None or not form.get("id"): |
|
raise ScrapeError("Could not find the Studiengang detail form") |
|
form_id = form["id"] |
|
action_url = urljoin(base_url, form.get("action", base_url)) |
|
|
|
view_state = form.find("input", {"name": "javax.faces.ViewState"}) |
|
client_window = form.find("input", {"name": "javax.faces.ClientWindow"}) |
|
if view_state is None or client_window is None: |
|
raise ScrapeError("Could not extract JSF state fields from the detail page") |
|
|
|
details_container = soup.find(id=lambda value: isinstance(value, str) and value.endswith("studiengangsbereich")) |
|
if details_container is None or not details_container.get("id"): |
|
raise ScrapeError("Could not locate the Studiengang detail panel") |
|
|
|
tree_root = soup.find( |
|
lambda tag: tag.has_attr("id") |
|
and tag.has_attr("class") |
|
and any(cls_name == "ui-treetable" for cls_name in tag.get("class", [])) |
|
) |
|
if tree_root is None: |
|
raise ScrapeError("Could not locate the Studiengang tree table") |
|
|
|
return cls( |
|
client=client, |
|
base_url=base_url, |
|
action_url=action_url, |
|
form_id=form_id, |
|
tree_id=tree_root["id"], |
|
details_id=details_container["id"], |
|
view_state=view_state.get("value", ""), |
|
client_window=client_window.get("value", ""), |
|
) |
|
|
|
def request(self, payload: dict[str, str], *, faces_ajax: bool = False) -> str: |
|
merged = dict(payload) |
|
merged[self.form_id] = self.form_id |
|
merged["javax.faces.ViewState"] = self.view_state |
|
merged["javax.faces.ClientWindow"] = self.client_window |
|
headers = { |
|
"User-Agent": USER_AGENT, |
|
"Referer": self.base_url, |
|
"Content-Type": "application/x-www-form-urlencoded", |
|
} |
|
if faces_ajax: |
|
headers["Faces-Request"] = "partial/ajax" |
|
|
|
attempts = MAX_RETRIES if faces_ajax else 1 |
|
for attempt in range(1, attempts + 1): |
|
now = time.monotonic() |
|
wait_time = REQUEST_SPACING_SECONDS - (now - self.last_request_at) |
|
if wait_time > 0: |
|
time.sleep(wait_time) |
|
try: |
|
response = self.client.post(self.action_url, merged, headers=headers) |
|
self.last_request_at = time.monotonic() |
|
return response |
|
except HTTPError as exc: |
|
self.last_request_at = time.monotonic() |
|
if not faces_ajax or exc.code not in {403, 429} or attempt == attempts: |
|
raise |
|
time.sleep(REQUEST_SPACING_SECONDS * attempt) |
|
raise ScrapeError("AJAX request retry loop exited unexpectedly") |
|
|
|
def parse_partial_updates(self, xml_text: str) -> dict[str, str]: |
|
root = ET.fromstring(xml_text) |
|
updates: dict[str, str] = {} |
|
for update in root.findall(".//update"): |
|
update_id = update.get("id") or "" |
|
updates[update_id] = update.text or "" |
|
if "javax.faces.ViewState" in update_id and update.text: |
|
self.view_state = update.text |
|
if "javax.faces.ClientWindow" in update_id and update.text: |
|
self.client_window = update.text |
|
return updates |
|
|
|
|
|
|
|
def sort_rowkey(value: str) -> list[int]: |
|
return [int(part) for part in value.split("_")] |
|
|
|
|
|
|
|
def row_to_node(row) -> dict[str, object]: |
|
cells = row.find_all("td") |
|
toggler = row.find("span", class_="ui-treetable-toggler") |
|
return { |
|
"rowkey": row.get("data-rk"), |
|
"parent_rowkey": row.get("data-prk"), |
|
"name": normalize(cells[0].get_text(" ", strip=True)), |
|
"child_area_count": int("".join(cells[1].stripped_strings) or "0"), |
|
"module_count": int("".join(cells[2].stripped_strings) or "0"), |
|
"lp": int("".join(cells[3].stripped_strings) or "0"), |
|
"expandable": bool(toggler and "visibility:hidden" not in (toggler.get("style") or "")), |
|
"entries": [], |
|
"children": [], |
|
} |
|
|
|
|
|
|
|
def extract_initial_nodes(session: TreeSession) -> dict[str, dict[str, object]]: |
|
html = session.client.get(session.base_url, headers={"User-Agent": USER_AGENT, "Referer": session.base_url}) |
|
soup = BeautifulSoup(html, "html.parser") |
|
tbody = soup.find(id=f"{session.tree_id}_data") |
|
if tbody is None: |
|
raise ScrapeError("Could not locate the initial tree rows") |
|
nodes: dict[str, dict[str, object]] = {} |
|
for row in tbody.find_all("tr", attrs={"data-rk": True}): |
|
node = row_to_node(row) |
|
nodes[str(node["rowkey"])] = node |
|
return nodes |
|
|
|
|
|
|
|
def expand_node(session: TreeSession, rowkey: str) -> list[dict[str, object]]: |
|
xml_text = session.request( |
|
{ |
|
"javax.faces.partial.ajax": "true", |
|
"javax.faces.source": session.tree_id, |
|
"javax.faces.partial.execute": session.tree_id, |
|
"javax.faces.partial.render": session.tree_id, |
|
f"{session.tree_id}_expand": rowkey, |
|
}, |
|
faces_ajax=True, |
|
) |
|
updates = session.parse_partial_updates(xml_text) |
|
fragment = updates.get(session.tree_id, "") |
|
soup = BeautifulSoup(fragment, "html.parser") |
|
return [row_to_node(row) for row in soup.find_all("tr", attrs={"data-rk": True})] |
|
|
|
|
|
|
|
def select_node_entries(session: TreeSession, rowkey: str) -> list[dict[str, str | None]]: |
|
try: |
|
xml_text = session.request( |
|
{ |
|
"javax.faces.partial.ajax": "true", |
|
"javax.faces.source": session.tree_id, |
|
"javax.faces.partial.execute": session.tree_id, |
|
"javax.faces.partial.render": session.details_id, |
|
"javax.faces.behavior.event": "select", |
|
"javax.faces.partial.event": "select", |
|
f"{session.tree_id}_selection": rowkey, |
|
f"{session.tree_id}_instantSelection": rowkey, |
|
}, |
|
faces_ajax=True, |
|
) |
|
except HTTPError as exc: |
|
raise ScrapeError(f"Failed to load module entries for rowkey {rowkey}: HTTP {exc.code}") from exc |
|
updates = session.parse_partial_updates(xml_text) |
|
fragment = updates.get(session.details_id, "") |
|
soup = BeautifulSoup(fragment, "html.parser") |
|
|
|
table = next( |
|
( |
|
table |
|
for table in soup.find_all("table") |
|
if "Name:" in {normalize(th.get_text(" ", strip=True)) for th in table.find_all("th")} |
|
and {normalize(th.get_text(" ", strip=True)) for th in table.find_all("th")} & {"#M", "Nummer:"} |
|
), |
|
None, |
|
) |
|
if table is None: |
|
return [] |
|
|
|
entries: list[dict[str, str | None]] = [] |
|
for row in table.find_all("tr")[1:]: |
|
cells = row.find_all("td") |
|
if len(cells) < 8: |
|
continue |
|
name_link = cells[0].find("a", href=True) |
|
number_link = cells[1].find("a", href=True) |
|
entries.append( |
|
{ |
|
"name": normalize(cells[0].get_text(" ", strip=True)), |
|
"module_number": normalize(cells[1].get_text(" ", strip=True)), |
|
"version": normalize(cells[2].get_text(" ", strip=True)), |
|
"lp": normalize(cells[3].get_text(" ", strip=True)), |
|
"grading": normalize(cells[4].get_text(" ", strip=True)), |
|
"exam_type": normalize(cells[5].get_text(" ", strip=True)), |
|
"turnus": normalize(cells[6].get_text(" ", strip=True)), |
|
"weight": normalize(cells[7].get_text(" ", strip=True)), |
|
"module_href": urljoin(session.base_url, name_link["href"]) if name_link else None, |
|
"number_href": urljoin(session.base_url, number_link["href"]) if number_link else None, |
|
} |
|
) |
|
return entries |
|
|
|
|
|
|
|
def build_module_tree(session: TreeSession) -> dict[str, object]: |
|
nodes = extract_initial_nodes(session) |
|
queue = [rowkey for rowkey, node in nodes.items() if node["expandable"]] |
|
expanded: set[str] = set() |
|
while queue: |
|
rowkey = queue.pop(0) |
|
if rowkey in expanded: |
|
continue |
|
expanded.add(rowkey) |
|
for child in expand_node(session, rowkey): |
|
child_rowkey = str(child["rowkey"]) |
|
nodes[child_rowkey] = child |
|
if child["expandable"]: |
|
queue.append(child_rowkey) |
|
|
|
for rowkey in sorted(nodes, key=sort_rowkey): |
|
if int(nodes[rowkey]["module_count"]) > 0: |
|
nodes[rowkey]["entries"] = select_node_entries(session, rowkey) |
|
else: |
|
nodes[rowkey]["entries"] = [] |
|
|
|
for node in nodes.values(): |
|
node["children"] = [] |
|
for rowkey, node in sorted(nodes.items(), key=lambda item: sort_rowkey(item[0])): |
|
parent_rowkey = node["parent_rowkey"] |
|
if isinstance(parent_rowkey, str) and parent_rowkey != "root" and parent_rowkey in nodes: |
|
nodes[parent_rowkey]["children"].append(node) |
|
|
|
root = next((node for node in nodes.values() if node["parent_rowkey"] == "root"), None) |
|
if root is None: |
|
raise ScrapeError("Could not determine the root module tree node") |
|
return root |
|
|
|
|
|
|
|
def compact_tree(node: dict[str, object]) -> dict[str, object]: |
|
return { |
|
"name": node["name"], |
|
"rowkey": node["rowkey"], |
|
"child_area_count": node["child_area_count"], |
|
"module_count": node["module_count"], |
|
"lp": node["lp"], |
|
"entries": node["entries"], |
|
"children": [compact_tree(child) for child in sorted(node["children"], key=lambda item: sort_rowkey(str(item["rowkey"])))], |
|
} |
|
|
|
|
|
|
|
def main() -> int: |
|
parser = argparse.ArgumentParser(description=__doc__) |
|
parser.add_argument("studiengang", help="Studiengang id, name, or short name") |
|
parser.add_argument("--degree", help="Optional exact degree filter for disambiguation") |
|
parser.add_argument("--provider", help="Optional exact provider filter for disambiguation") |
|
parser.add_argument("--short", help="Optional exact short-name filter for disambiguation") |
|
parser.add_argument("--pretty", action="store_true", help="Pretty-print JSON output") |
|
args = parser.parse_args() |
|
|
|
client = MosesClient() |
|
try: |
|
program = resolve_program( |
|
client, |
|
args.studiengang, |
|
degree=args.degree, |
|
provider=args.provider, |
|
short=args.short, |
|
) |
|
session = TreeSession.from_program(client, program) |
|
tree = compact_tree(build_module_tree(session)) |
|
except (ResolutionError, ScrapeError, ET.ParseError) as exc: |
|
print(str(exc), file=sys.stderr) |
|
return 2 |
|
|
|
payload = {"program": program, "tree": tree} |
|
json.dump(payload, sys.stdout, ensure_ascii=False, indent=2 if args.pretty else None) |
|
sys.stdout.write("\n") |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
raise SystemExit(main()) |