"""CIS Enhanced - FastAPI app.""" import os from datetime import datetime, timedelta from pathlib import Path from dotenv import load_dotenv from fastapi import FastAPI, Query, Request from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.templating import Jinja2Templates load_dotenv(Path(__file__).resolve().parent.parent / ".env") for _var in ("CIS_USER", "CIS_PASSWORD"): if not os.environ.get(_var) or "HIER_" in os.environ[_var]: raise RuntimeError( f"Umweltvariable {_var} fehlt. .env ausfüllen (siehe .env.example) " f"bzw. beim Docker-Start über --env-file / env_file übergeben.") from fastapi.staticfiles import StaticFiles # noqa: E402 from .cis import CISClient, SLOTS # noqa: E402 from .util import (current_semester, decode_matrikel, events_to_json, fmt_range, # noqa: E402 lv_color, week_label, week_start) BASE = os.environ["CIS_BASE_URL"].rstrip("/") client = CISClient(os.environ["CIS_USER"], os.environ["CIS_PASSWORD"], BASE) OWN_UID = os.environ.get("OWN_UID", os.environ["CIS_USER"]) app = FastAPI(title="CIS Enhanced") app.mount("/static", StaticFiles(directory=str(Path(__file__).parent / "static")), name="static") templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates")) templates.env.globals["lv_color"] = lv_color templates.env.globals["decode_matrikel"] = decode_matrikel templates.env.globals["fmt_range"] = fmt_range templates.env.globals["week_label"] = week_label templates.env.globals["slots"] = SLOTS def _meta(): return client.meta() @app.get("/", response_class=HTMLResponse) def home(request: Request): meta = _meta() return templates.TemplateResponse(request, "home.html", { "own_uid": OWN_UID, "stgs": meta["stgs"], "semesters": meta["semesters"], "current_sem": current_semester(meta["semesters"]), }) def _plan_common(request, kind, uid, kz, sem, ver, grp, semester, week="", tab="week"): meta = _meta() semester = semester or current_semester(meta["semesters"]) s = next((x for x in meta["semesters"] if x["label"] == semester), meta["semesters"][0]) person = None if kind == "student": uid = uid or OWN_UID person = client.person(uid) events = client.student_events(uid, s["begin"], s["ende"]) else: kz = kz or "640" sem = sem or "1" ver = ver or "V" events = client.stg_events(kz, sem, ver, grp, s["begin"], s["ende"]) # initial week: explicit param, today if inside semester, else first event week init_week = None try: if week: init_week = week_start(datetime.strptime(week, "%Y-%m-%d").date()) except ValueError: pass if init_week is None: now = datetime.now() if s["begin"] <= now.timestamp() <= s["ende"]: first_day = now.date() elif events: first_day = events[0].start.date() else: first_day = datetime.fromtimestamp(s["begin"]).date() init_week = week_start(first_day) stg = next((x for x in meta["stgs"] if x["kz"] == kz), None) verbands = client.verbands_for(kz, sem) if kind == "stg" and kz and sem else meta["verbands"] return { "request": request, "kind": kind, "uid": uid or "", "kz": kz or "", "sem": sem or "", "ver": ver or "", "grp": grp or "", "stg": stg, "verbands": verbands, "person": person, "events": events, "events_json": events_to_json(events), "meta": meta, "semester": semester, "sem_range": (datetime.fromtimestamp(s["begin"]), datetime.fromtimestamp(s["ende"])), "init_week": init_week, "tab": tab if tab in ("week", "sem") else "week", } @app.get("/plan", response_class=HTMLResponse) def plan(request: Request, uid: str = Query("", description="Matrikelnummer"), semester: str = Query(""), week: str = Query("", description="Woche als YYYY-MM-DD (Monatag)"), tab: str = Query("week", description="week|sem")): return templates.TemplateResponse(request, "plan.html", _plan_common( request, "student", uid, "", "", "", "", semester, week, tab)) @app.get("/stg", response_class=HTMLResponse) def stg_page(request: Request, kz: str = Query("640"), sem: str = Query("1"), ver: str = Query("V"), grp: str = Query(""), semester: str = Query(""), week: str = Query(""), tab: str = Query("week", description="week|sem")): return templates.TemplateResponse(request, "plan.html", _plan_common( request, "stg", "", kz, sem, ver, grp, semester, week, tab)) def verband_label(v: str) -> str: return {"V": " (Vollzeit)", "B": " (berufsbegl.)"}.get(v, "") templates.env.globals["verband_label"] = verband_label @app.get("/liste", response_class=HTMLResponse) def liste(request: Request, kz: str = Query("640"), sem: str = Query("1"), ver: str = Query("V")): meta = _meta() students = client.namensliste(kz, sem, ver) stg = next((x for x in meta["stgs"] if x["kz"] == kz), None) verbands = client.verbands_for(kz, sem) students_json = [{"nachname": s.nachname, "vorname": s.vorname, "geschlecht": s.geschlecht, "matrikel": s.matrikel, "mail": s.mail} for s in students] return templates.TemplateResponse(request, "liste.html", { "request": request, "meta": meta, "kz": kz, "sem": sem, "ver": ver, "verbands": verbands, "stg": stg, "students": students, "students_json": students_json, }) @app.get("/api/meta") def api_meta(): return _meta() @app.get("/api/semesters_for_stg") def api_semesters_for_stg(kz: str = Query(...)): """Study semesters that exist for a StG (from CIS autocomplete endpoint).""" try: r = client.s.post(f"{BASE}/private/lvplan/lvplan_autocomplete.php", data={"autocomplete": "getSemester", "stg_kz": kz}, timeout=30) return JSONResponse(r.json()) except Exception: return JSONResponse([]) @app.get("/api/verbands") def api_verbands(kz: str = Query(...), sem: str = Query("")): return client.verbands_for(kz, sem) @app.get("/api/detail") def api_detail(kind: str = "student", datum: str = "", stunde: int = 0, uid: str = "", kz: str = "", sem: str = "", ver: str = "", grp: str = "", ort: str = ""): if kind == "student" and not (kz and sem and ver): p = client.person(uid) if p: if not kz: code = p.stg.split(" - ")[0].strip() stg = next((x for x in _meta()["stgs"] if x["code"] == code), None) kz = stg["kz"] if stg else "" sem = sem or p.semester ver = ver or p.verband grp = grp or p.gruppe try: rows = client.detail(kind, datum, stunde, uid, kz, sem, ver, grp, ort) return {"rows": rows} except Exception as e: return JSONResponse({"error": str(e)}, status_code=502) @app.get("/export/ical") def export_ical(kind: str = "student", uid: str = Query(""), kz: str = Query(""), sem: str = Query(""), ver: str = Query(""), grp: str = Query(""), semester: str = Query("")): meta = _meta() s = next((x for x in meta["semesters"] if x["label"] == semester), next((x for x in meta["semesters"] if x["begin"] <= datetime.now().timestamp() <= x["ende"]), meta["semesters"][0])) if kind == "student": ical = client.ical_raw("student", uid=uid or OWN_UID, begin=s["begin"], ende=s["ende"]) fname = f"lvplan_{uid or OWN_UID}_{s['label']}.ics" else: ical = client.ical_raw("stg", kz=kz, sem=sem, ver=ver, grp=grp, begin=s["begin"], ende=s["ende"]) fname = f"lvplan_{kz}_{sem}_{ver}_{s['label']}.ics" return Response(ical, media_type="text/calendar", headers={"Content-Disposition": f'attachment; filename="{fname}"'}) @app.get("/health") def health(): return {"ok": True}