"""CIS API client for FH-Burgenland (basic auth, HTML/iCal scraping).""" import json import re import threading import time from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from functools import lru_cache from pathlib import Path from typing import Optional import requests from bs4 import BeautifulSoup DATA_DIR = Path(__file__).resolve().parent.parent / "data" PERSON_ID_CACHE = DATA_DIR / "person_ids.json" PHOTO_DIR = DATA_DIR / "photos" _FILE_LOCK = threading.Lock() TZ = timezone(timedelta(hours=1)) # Europe/Vienna winter; events carry their own tz, we just parse naive SLOTS = [ "08:30", "09:30", "10:15", "11:15", "12:00", "13:00", "14:00", "14:45", "15:45", "16:30", "17:45", "18:30", "19:30", "20:15", "21:00", ] SLOT_ENDS = [ "09:15", "10:15", "11:00", "12:00", "12:45", "13:45", "14:45", "15:30", "16:30", "17:15", "18:30", "19:15", "20:15", "21:00", "21:45", ] @dataclass class Event: lv: str lektor: str gruppe: str ort: str start: datetime # naive, Vienna wall time end: datetime uid: str = "" @property def date(self): return self.start.date() @property def weekday(self): return self.start.weekday() # 0=Mon @property def hours(self): return round((self.end - self.start).total_seconds() / 3600, 2) @dataclass class Person: uid: str name: str stg: str = "" semester: str = "" verband: str = "" gruppe: str = "" kind: str = "student" # student | lektor @dataclass class Student: nachname: str vorname: str geschlecht: str mail: str matrikel: str = "" @property def name(self): return f"{self.nachname}, {self.vorname}" def parse_ical(text: str) -> list[Event]: events = [] for block in re.findall(r"BEGIN:VEVENT(.*?)END:VEVENT", text, re.S): fields = {} for line in block.strip().splitlines(): if ":" in line: key, _, val = line.partition(":") fields[key.split(";")[0].upper()] = val desc = fields.get("DESCRIPTION", "").replace("\\n", "\n").split("\n") desc = [d.strip() for d in desc if d.strip()] lv, lektor, gruppe, ort = (desc + ["", "", "", ""])[:4] if not lv: m = re.match(r"^\s*(\S+)\s+(.*)$", fields.get("SUMMARY", "")) lv = m.group(1) if m else fields.get("SUMMARY", "") ort = m.group(2).strip() if m else "" if " - " in fields.get("SUMMARY", ""): gruppe = fields["SUMMARY"].rsplit(" - ", 1)[1].strip() dtstart = fields.get("DTSTART", "") dtend = fields.get("DTEND", "") if not dtstart or not dtend: continue start = datetime.strptime(dtstart, "%Y%m%dT%H%M%S") end = datetime.strptime(dtend, "%Y%m%dT%H%M%S") if end <= start: end = start + timedelta(hours=1) events.append(Event(lv=lv, lektor=lektor, gruppe=gruppe, ort=ort, start=start, end=end, uid=fields.get("UID", ""))) events.sort(key=lambda e: e.start) return events class CISClient: def __init__(self, user: str, password: str, base_url: str): self.base = base_url.rstrip("/") self.s = requests.Session() self.s.auth = (user, password) self.s.headers["User-Agent"] = "Mozilla/5.0 (cis-enhanced)" self._meta_cache: Optional[dict] = None self._meta_ts = 0.0 self._benutzer: Optional[list[dict]] = None self._benutzer_ts = 0.0 self._profile_cache: dict = {} # ---------- metadata (stg list, semesters, rooms) ---------- def meta(self, max_age: int = 3600) -> dict: if self._meta_cache and time.time() - self._meta_ts < max_age: return self._meta_cache html = self._get(f"{self.base}/private/lvplan/index.php") soup = BeautifulSoup(html, "html.parser") selects: dict[str, list[tuple[str, str]]] = {} for s_ in soup.find_all("select"): name = s_.get("name") or "" opts = [(o.get("value") or "", o.get_text(" ", strip=True)) for o in s_.find_all("option")] if name not in selects or len(opts) > len(selects[name]): selects[name] = opts stgs = [] for vz, label in selects.get("stg_kz", []): if not vz: continue m = re.match(r"^(\S+)\s+\((.*)\)$", label) stgs.append({"kz": vz, "code": m.group(1) if m else vz, "name": m.group(2) if m else label}) semesters = [] for raw, label in selects.get("begin", []): m = re.match(r"^(\d+)&ende=(\d+)$", raw or "") if not m: continue semesters.append({"label": label, "begin": int(m.group(1)), "ende": int(m.group(2))}) semesters.sort(key=lambda s: s["begin"], reverse=True) rooms = [] for raw, label in selects.get("select", []): if "ort_kurzbz=" not in raw: continue ort = raw.split("ort_kurzbz=")[1].split("&")[0] m = re.match(r"^(\S+)\s+\((.*)\)$", label) rooms.append({"kurzbz": ort, "code": m.group(1) if m else ort, "name": m.group(2) if m else label}) self._meta_cache = { "stgs": stgs, "semesters": semesters, "rooms": rooms, "verbands": ["V", "B", "A", "C", "D", "E", "F"], "grps": [str(i) for i in range(1, 7)], } self._meta_ts = time.time() return self._meta_cache # ---------- low level ---------- def _get(self, url: str, **kw) -> str: r = self.s.get(url, timeout=30, **kw) r.raise_for_status() return r.text def _ical_url(self, params: dict) -> str: q = "&".join(f"{k}={v}" for k, v in params.items()) return f"{self.base}/private/lvplan/stpl_kalender.php?{q}" def _ical_student_params(self, uid: str, begin: int, ende: int) -> dict: ptype = "student" if uid.isdigit() else "lektor" return {"type": ptype, "pers_uid": uid, "ort_kurzbz": "", "stg_kz": "", "sem": "", "ver": "", "grp": "", "gruppe_kurzbz": "", "lva": "", "begin": begin, "ende": ende, "format": "ical", "version": "2", "target": "ical"} def _ical_stg_params(self, kz: str, sem: str, ver: str, grp: str, begin: int, ende: int) -> dict: return {"type": "verband", "stg_kz": kz, "sem": sem, "ver": ver, "grp": grp, "begin": begin, "ende": ende, "format": "ical", "version": "2", "target": "ical"} # ---------- events ---------- def student_events(self, uid: str, begin: int, ende: int) -> list[Event]: return parse_ical(self._get(self._ical_url(self._ical_student_params(uid, begin, ende)))) def stg_events(self, kz: str, sem: str, ver: str, grp: str, begin: int, ende: int) -> list[Event]: return parse_ical(self._get(self._ical_url(self._ical_stg_params(kz, sem, ver, grp, begin, ende)))) def ical_raw(self, kind: str, **kw) -> str: if kind in ("student", "lektor"): return self._get(self._ical_url(self._ical_student_params(kw["uid"], kw["begin"], kw["ende"]))) return self._get(self._ical_url(self._ical_stg_params(kw["kz"], kw["sem"], kw["ver"], kw.get("grp", ""), kw["begin"], kw["ende"]))) # ---------- person ---------- @lru_cache(maxsize=256) def person(self, uid: str) -> Optional[Person]: uid = (uid or "").strip() if not uid: return None is_student = uid.isdigit() try: url = f"{self.base}/private/lvplan/stpl_week.php?pers_uid={uid}" if not is_student: url += "&type=lektor" html = self._get(url) except requests.HTTPError: return None t = re.sub(r"<[^>]+>", " ", html) t = re.sub(r"\s+", " ", t) if is_student: m = re.search(r"Person:\s*(.+?)\s*-\s*(\d{9,10}).*?Studiengang:\s*(.+?)\s+Semester:\s*(\S+?)\s+Verband:\s*(\S+?)\s+Gruppe:", t) if not m: return None name, u, stg, sem, ver = m.groups() return Person(uid=u, name=name.strip(), stg=stg.strip(), semester=sem.strip(), verband=ver.strip(), gruppe="", kind="student") m = re.search(r"Person:\s*(.+?)\s*-\s*([A-Za-z][\w.]*)", t) if not m: return None name, u = m.groups() return Person(uid=u, name=name.strip(), kind="lektor") # ---------- person_id + photos ---------- def person_id_for(self, uid: str) -> str: """Matrikel/username → CIS person_id (persistent JSON cache).""" uid = (uid or "").strip() if not uid: return "" with _FILE_LOCK: cache: dict = {} if PERSON_ID_CACHE.exists(): try: cache = json.loads(PERSON_ID_CACHE.read_text()) except Exception: cache = {} if uid in cache: return cache[uid] try: html = self._get(f"{self.base}/private/profile/index.php?uid={uid}") except requests.HTTPError: return "" m = re.search(r"bild\.php\?src=person&person_id=(\d+)", html) if not m: return "" pid = m.group(1) with _FILE_LOCK: if PERSON_ID_CACHE.exists(): try: cache = json.loads(PERSON_ID_CACHE.read_text()) except Exception: cache = {} cache[uid] = pid try: PERSON_ID_CACHE.parent.mkdir(parents=True, exist_ok=True) PERSON_ID_CACHE.write_text(json.dumps(cache, indent=0)) except OSError: pass return pid def photo_bytes(self, person_id: str, max_age: int = 7 * 86400) -> Optional[tuple[bytes, str]]: """Profilfoto holen (7-Tage-Disk-Cache). (bytes, content_type) oder None.""" person_id = (person_id or "").strip() if not person_id.isdigit(): return None f = PHOTO_DIR / f"{person_id}.bin" if f.exists() and time.time() - f.stat().st_mtime < max_age: try: return f.read_bytes(), "image/jpeg" except OSError: pass try: r = self.s.get(f"{self.base}/public/bild.php?src=person&person_id={person_id}", timeout=30) r.raise_for_status() except requests.RequestException: return None data = r.content if not data: return None ct = (r.headers.get("Content-Type") or "image/jpeg").split(";")[0].strip() if ct not in ("image/jpeg", "image/png", "image/gif"): return None try: PHOTO_DIR.mkdir(parents=True, exist_ok=True) f.write_bytes(data) except OSError: pass return data, ct # ---------- Namensliste ---------- def namensliste(self, kz: str, sem: str, ver: str) -> list[Student]: html = self._get(f"{self.base}/private/stud_in_grp.php?kz={kz}&sem={sem}&verband={ver}") soup = BeautifulSoup(html, "html.parser") tbl = soup.find("table", class_="tablesorter") out = [] if not tbl: return out # columns: [checkbox?, nachname, vorname, geschlecht, mail] for tr in tbl.find_all("tr")[1:]: cells = [c.get_text(" ", strip=True) for c in tr.find_all(["td", "th"])] if not cells or not any(cells): continue # drop leading checkbox column if present if len(cells) == 5 and not cells[0]: cells = cells[1:] if len(cells) < 4: continue nachname, vorname, geschlecht, mail = cells[0], cells[1], cells[2], cells[3] if not nachname and not vorname: continue out.append(Student(nachname=nachname, vorname=vorname, geschlecht=geschlecht, mail=mail, matrikel=mail.split("@")[0] if "@" in mail else "")) return out # ---------- Studien-Semester pro StG ---------- @lru_cache(maxsize=64) def study_semesters(self, kz: str) -> list[int]: """Von CIS bekannte Studien-Semester eines StG (z.B. [1..6] BSc, [1..4] MSc).""" try: r = self.s.post(f"{self.base}/private/lvplan/lvplan_autocomplete.php", data={"autocomplete": "getSemester", "stg_kz": kz}, timeout=30) data = r.json() vals = data.values() if isinstance(data, dict) else data return sorted({int(x) for x in vals if str(x).strip().isdigit()}) except Exception: return [] # ---------- Benutzer (alle Studierenden + Lektoren) ---------- def benutzer_all(self, max_age: int = 6 * 3600) -> list[dict]: """Alle aktiven Benutzer (Studierende + Lektoren) via CIS-Autocomplete. Substring-Suche über Vokale liefert die Vollmenge.""" if self._benutzer and time.time() - self._benutzer_ts < max_age: return self._benutzer merged: dict[str, dict] = {} for ch in "aeiou": try: r = self.s.get(f"{self.base}/private/lvplan/lvplan_autocomplete.php", params={"autocomplete": "benutzer", "term": ch}, timeout=60) t = r.text.strip() if not t: continue for d in r.json(): uid = str(d.get("uid") or "").strip() if not uid: continue merged[uid] = {"uid": uid, "vorname": (d.get("vorname") or "").strip(), "nachname": (d.get("nachname") or "").strip(), "mitarbeiter_uid": (d.get("mitarbeiter_uid") or "").strip()} except Exception: continue self._benutzer = list(merged.values()) self._benutzer_ts = time.time() return self._benutzer # ---------- Profilseite ---------- def profile(self, uid: str, max_age: int = 3600) -> Optional[dict]: """Profilseite parsen (StudentIn oder MitarbeiterIn).""" uid = (uid or "").strip() if not uid: return None hit = self._profile_cache.get(uid) if hit and time.time() - hit[0] < max_age: return hit[1] try: html = self._get(f"{self.base}/private/profile/index.php?uid={uid}") except requests.HTTPError: return None t = re.sub(r"<[^>]+>", " ", html) t = re.sub(r"\s+", " ", t) def _pick(pattern: str, group: int = 1) -> str: m = re.search(pattern, t) return m.group(group).strip() if m else "" role = "" for cand in ("StudentIn", "MitarbeiterIn Extern", "MitarbeiterIn Intern", "MitarbeiterIn"): if re.search(rf"\b{cand}\b", t): role = cand break m = re.search(r"bild\.php\?src=person&person_id=(\d+)", html) prof = { "uid": uid, "role": role, "person_id": m.group(1) if m else "", "username": _pick(r"Username:\s*(\S+)"), "anrede": _pick(r"Anrede:\s*(.*?)\s*(?:Titel:|Vorname:)"), "matrikel_nr": _pick(r"Matrikelnummer:\s*(\d+)"), "titel": _pick(r"Titel:\s*(.*?)\s*Vorname:").strip(), "vorname": _pick(r"Vorname:\s*(.*?)\s*Nachname:"), "nachname": _pick(r"Nachname:\s*(.*?)\s*Postnomen:"), "postnomen": _pick(r"Postnomen:\s*(.*?)\s*(?:Geburtsdatum:|Kurzzeichen:|Studiengang:|eMail)"), "geburtsdatum": _pick(r"Geburtsdatum:\s*([\d.]+)"), "geburtsort": _pick(r"Geburtsort:\s*(.*?)\s*(?:Hauptwohnsitz:|Studiengang:)"), "wohnsitz": _pick(r"Hauptwohnsitz:\s*(.*?)\s*Studiengang:"), "fh_ausweis": _pick(r"FH-Ausweis Status\s*(.*?)\s*eMail"), "privat_mail": _pick(r"Privat\s+E-Mail\s+(\S+@\S+)"), "privat_tel": _pick(r"Telefonnummer\s+([+\d][\d\s/-]*)"), "telefon": _pick(r"Telefon:\s*(.*?)\s*(?:B\xfcro:|eMail|$)"), "buero": _pick(r"B\xfcro:\s*(\S+)"), "kurzzeichen": _pick(r"Kurzzeichen:\s*(\S+)"), "mail": _pick(r"Intern:\s*([^\s]+@[^\s]+)"), "mail_alias": _pick(r"Alias:\s*([^\s]+@[^\s]+)"), "stg": _pick(r"Studiengang:\s*(.*?)\s*Semester:"), "semester": _pick(r"Semester:\s*(\d+)"), "verband": _pick(r"Verband:\s*(\S+)"), } if not prof["nachname"] and not prof["vorname"]: prof = None self._profile_cache[uid] = (time.time(), prof) return prof # ---------- Reichweiten-Check (Security) ---------- PRIVATE_MARKERS = ("Geburtsdatum:", "Geburtsort:", "Hauptwohnsitz:", "Telefonnummer", "FH-Ausweis Status") def privacy_probe(self, uid: str, own_uid: str, semester: dict) -> list[dict]: """Access-Control-Check gegen EINE fremde Person. Returns nur Booleans/Anzahlen - niemals Feldwerte.""" checks: list[dict] = [] def add(name: str, ok, info: str = ""): checks.append({"name": name, "ok": bool(ok), "info": info}) users = self.benutzer_all() add("Verzeichnis Autocomplete", len(users) > 500, f"{len(users)} Personen (Name+UID) mit 5 Requests") prof = self.profile(uid) or {} nachname = prof.get("nachname") or "" html = "" try: html = self._get(f"{self.base}/private/profile/index.php?uid={uid}") except requests.HTTPError: pass t = re.sub(r"<[^>]+>", " ", html) if html else "" fremd = bool(nachname) and nachname in t and uid != own_uid priv = [m.rstrip(":") for m in self.PRIVATE_MARKERS if m in t] if fremd else [] add("Fremde Privinfelder im Profil", priv, ("LEAK: " + ", ".join(priv)) if priv else "gefiltert (nur oeffentliche Felder)" if fremd else "Ziel nicht ladbar") pid = self.person_id_for(uid) photo_ok, photo_info = False, "kein person_id" if pid: try: r = self.s.get(f"{self.base}/public/bild.php?src=person&person_id={pid}", timeout=30) ct = (r.headers.get("Content-Type") or "").split(";")[0] photo_ok = r.status_code == 200 and ct.startswith("image/") photo_info = f"HTTP {r.status_code}, {ct}, {len(r.content)} Bytes" except requests.RequestException as e: photo_info = str(e) add("Profilfoto via person_id (IDOR)", photo_ok, photo_info) trap_ok, trap_info = False, "-" if pid: try: h2 = self._get(f"{self.base}/private/profile/index.php?person_id={pid}") t2 = re.sub(r"<[^>]+>", " ", h2) trap_ok = bool(nachname) and nachname in t2 and own_uid not in t2 trap_info = ("LEAK: fremdes Profil" if trap_ok else "Parameter ignoriert -> eigenes Profil") except requests.HTTPError as e: trap_info = f"HTTP {e}" add("Falle: person_id statt uid uebergeben", trap_ok, trap_info) b, e = semester["begin"], semester["ende"] try: evs = self.student_events(uid, b, e) except Exception: evs = [] add("Fremder LV-Plan (HTML/iCal)", len(evs) > 0, f"{len(evs)} Termine im aktuellen Semester") try: r = requests.get(self._ical_url(self._ical_student_params(uid, b, e)), timeout=20, allow_redirects=False) body = r.text if r.status_code == 200 else "" anon = r.status_code == 200 and "BEGIN:VCALENDAR" in body add("iCal OHNE Login (anonym)", anon, f"HTTP {r.status_code} {'-> VCALENDAR' if anon else '(Auth noetig)'}") except requests.RequestException as ex: add("iCal OHNE Login (anonym)", False, str(ex)) sem, ver = prof.get("semester", ""), prof.get("verband", "") kz = next((x["kz"] for x in self.meta()["stgs"] if x["name"] == prof.get("stg")), "") if kz and sem and ver: try: n = len(self.namensliste(kz, sem, ver)) except Exception: n = 0 add("Klassenliste Name+Mail", n > 0, f"{n} Personen (1 Request)") else: add("Klassenliste Name+Mail", False, "n/a (Ziel ohne StG-Badaten)") return checks # ---------- Verbände pro StG/Semester ---------- def verbands_for(self, kz: str, sem: str) -> list[str]: fallback = ["V", "B", "G", "A", "C", "D", "E", "F"] try: r = self.s.post(f"{self.base}/private/lvplan/lvplan_autocomplete.php", data={"autocomplete": "getVerband", "stg_kz": kz, "sem": sem}, timeout=30) data = r.json() if isinstance(data, dict): vals = list(data.values()) else: vals = data vals = [str(v).strip() for v in vals if str(v).strip()] return vals or fallback except Exception: return fallback # ---------- detail (LV details for one slot) ---------- def detail(self, kind: str, datum: str, stunde: int, uid: str = "", kz: str = "", sem: str = "", ver: str = "", grp: str = "", ort: str = "") -> dict: if kind == "lektor": url = (f"{self.base}/private/lvplan/stpl_detail.php?type=lektor&datum={datum}" f"&stunde={stunde}&pers_uid={uid}&stg_kz=&sem=&ver=&grp=&ort_kurzbz={ort}") elif kind == "student": url = (f"{self.base}/private/lvplan/stpl_detail.php?type=student&datum={datum}" f"&stunde={stunde}&pers_uid={uid}&stg_kz={kz}&sem={sem}&ver={ver}" f"&grp= {grp}&ort_kurzbz={ort}") else: url = (f"{self.base}/private/lvplan/stpl_detail.php?type=verband&datum={datum}" f"&stunde={stunde}&stg_kz={kz}&sem={sem}&ver={ver}" f"&grp={grp}&ort_kurzbz={ort}") html = self._get(url) soup = BeautifulSoup(html, "html.parser") headers, data_row = [], None for tr in soup.find_all("tr"): ths = tr.find_all("th") tds = tr.find_all("td") if ths and not tds: headers = [th.get_text(strip=True) for th in ths] elif tds and not ths: data_row = [td.get_text(" ", strip=True) for td in tds] if not headers or not data_row: return {} return {h: data_row[i] for i, h in enumerate(headers) if i < len(data_row)}