342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""CIS API client for FH-Burgenland (basic auth, HTML/iCal scraping)."""
|
|
import json
|
|
import re
|
|
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"
|
|
|
|
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
|
|
|
|
|
|
@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
|
|
|
|
# ---------- 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:
|
|
return {"type": "student", "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 == "student":
|
|
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]:
|
|
try:
|
|
html = self._get(f"{self.base}/private/lvplan/stpl_week.php?pers_uid={uid}")
|
|
except requests.HTTPError:
|
|
return None
|
|
t = re.sub(r"<[^>]+>", " ", html)
|
|
t = re.sub(r"\s+", " ", t)
|
|
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="")
|
|
|
|
# ---------- 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 ""
|
|
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)
|
|
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
|
|
|
|
# ---------- 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 == "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)}
|