458 lines
18 KiB
Python
458 lines
18 KiB
Python
"""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+)"),
|
|
"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)"),
|
|
"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
|
|
|
|
# ---------- 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)}
|