533 lines
18 KiB
Python
533 lines
18 KiB
Python
"""CIS Enhanced - FastAPI app."""
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, File, Form, Query, Request, UploadFile
|
|
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
|
|
|
|
import numpy as np # noqa: E402
|
|
|
|
from .cis import CISClient, SLOTS # noqa: E402
|
|
from .face import engine # 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()
|
|
|
|
|
|
def _study_sem_options(kz: str, semester_label: str) -> list[int]:
|
|
"""Gültige Studien-Semester eines StG, gefiltert nach Plan-Semester
|
|
(Winter=ungerade, Sommer=gerade)."""
|
|
sems = client.study_semesters(kz)
|
|
if not sems:
|
|
code = next((s["code"] for s in _meta()["stgs"] if s["kz"] == kz), "") or ""
|
|
max_n = 4 if code[:1] in ("M", "L") else 6
|
|
sems = list(range(1, max_n + 1))
|
|
label = (semester_label or "").upper()
|
|
if label.startswith("WS"):
|
|
sems = [n for n in sems if n % 2 == 1]
|
|
elif label.startswith("SS"):
|
|
sems = [n for n in sems if n % 2 == 0]
|
|
sems = sorted(set(sems))
|
|
return sems or [1]
|
|
|
|
|
|
@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
|
|
sems: list[int] = []
|
|
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"
|
|
sems = _study_sem_options(kz, semester)
|
|
if sem not in [str(n) for n in sems]:
|
|
sem = str(sems[0])
|
|
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 "",
|
|
"sems": sems,
|
|
"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 oder Lektor-Username (z.B. rmatzinger)"),
|
|
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()
|
|
sems = _study_sem_options(kz, current_semester(meta["semesters"]))
|
|
if sem not in [str(n) for n in sems]:
|
|
sem = str(sems[0])
|
|
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,
|
|
"sems": sems,
|
|
"ver": ver,
|
|
"verbands": verbands,
|
|
"stg": stg,
|
|
"students": students,
|
|
"students_json": students_json,
|
|
})
|
|
|
|
|
|
# ---------- Fotos / Gesichter ----------
|
|
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
|
|
EMB_CACHE = DATA_DIR / "embeddings.json"
|
|
MAX_UPLOAD = 8 * 1024 * 1024
|
|
|
|
|
|
def _emb_cache() -> dict:
|
|
if EMB_CACHE.exists():
|
|
try:
|
|
return json.loads(EMB_CACHE.read_text())
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
|
|
def _embed_person(pid: str, photo: bytes) -> np.ndarray | None:
|
|
cache = _emb_cache()
|
|
if pid in cache:
|
|
return np.asarray(cache[pid], dtype=np.float32)
|
|
v = engine.embed(photo)
|
|
if v is None:
|
|
return None
|
|
cache[pid] = v.tolist()
|
|
try:
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
EMB_CACHE.write_text(json.dumps(cache))
|
|
except OSError:
|
|
pass
|
|
return v
|
|
|
|
|
|
def _photo_wall(kz: str, sem: str, ver: str) -> list[dict]:
|
|
"""Students of a verband with person_id + whether a photo exists (cached check)."""
|
|
students = client.namensliste(kz, sem, ver)
|
|
wall = []
|
|
for s in students:
|
|
pid = client.person_id_for(s.matrikel)
|
|
wall.append({
|
|
"nachname": s.nachname, "vorname": s.vorname,
|
|
"matrikel": s.matrikel, "person_id": pid,
|
|
"has_photo": bool(pid),
|
|
})
|
|
return wall
|
|
|
|
|
|
@app.get("/fotos", response_class=HTMLResponse)
|
|
def fotos_page(request: Request,
|
|
kz: str = Query("640"),
|
|
sem: str = Query("1"),
|
|
ver: str = Query("V")):
|
|
meta = _meta()
|
|
sems = _study_sem_options(kz, current_semester(meta["semesters"]))
|
|
if sem not in [str(n) for n in sems]:
|
|
sem = str(sems[0])
|
|
stg = next((x for x in meta["stgs"] if x["kz"] == kz), None)
|
|
verbands = client.verbands_for(kz, sem)
|
|
wall = _photo_wall(kz, sem, ver)
|
|
return templates.TemplateResponse(request, "fotos.html", {
|
|
"request": request,
|
|
"meta": meta,
|
|
"kz": kz,
|
|
"sem": sem,
|
|
"sems": sems,
|
|
"ver": ver,
|
|
"verbands": verbands,
|
|
"stg": stg,
|
|
"wall": wall,
|
|
"wall_json": wall,
|
|
"own_uid": OWN_UID,
|
|
})
|
|
|
|
|
|
@app.get("/fotos/img/{person_id}")
|
|
def fotos_img(person_id: str):
|
|
if not person_id.isdigit():
|
|
return JSONResponse({"error": "Ungültige ID"}, status_code=400)
|
|
res = client.photo_bytes(person_id)
|
|
if not res:
|
|
return JSONResponse({"error": "Kein Foto vorhanden"}, status_code=404)
|
|
data, ct = res
|
|
return Response(data, media_type=ct)
|
|
|
|
|
|
# ---------- Gesichtssuche als Hintergrund-Job ----------
|
|
SCOPE_LABELS = {
|
|
"verband": "Nur Verband",
|
|
"stg": "Gesamter Studiengang",
|
|
"stud": "Alle Studierenden (FH-weit)",
|
|
"lektor": "Alle Lektoren",
|
|
"alle": "Alle Personen (Studierende + Lektoren)",
|
|
}
|
|
JOBS: dict[str, dict] = {}
|
|
JOBS_LOCK = threading.Lock()
|
|
MAX_JOBS = 8
|
|
|
|
|
|
def _scope_population(scope: str, kz: str, sem: str, ver: str) -> list[dict]:
|
|
"""Personen des Suchbereichs, dedupliziert nach uid."""
|
|
people: dict[str, dict] = {}
|
|
|
|
def add(uid, nachname, vorname, kind):
|
|
uid = (uid or "").strip()
|
|
if uid and uid not in people:
|
|
people[uid] = {"uid": uid, "nachname": nachname,
|
|
"vorname": vorname, "kind": kind}
|
|
|
|
if scope == "verband":
|
|
for s in client.namensliste(kz, sem, ver):
|
|
add(s.matrikel, s.nachname, s.vorname, "student")
|
|
elif scope == "stg":
|
|
sems = client.study_semesters(kz) or list(range(1, 7))
|
|
for s in sems:
|
|
for v in client.verbands_for(kz, str(s)):
|
|
for st in client.namensliste(kz, str(s), v):
|
|
add(st.matrikel, st.nachname, st.vorname, "student")
|
|
else:
|
|
for b in client.benutzer_all():
|
|
kind = "student" if b["uid"].isdigit() else "lektor"
|
|
if scope == "stud" and kind != "student":
|
|
continue
|
|
if scope == "lektor" and kind != "lektor":
|
|
continue
|
|
add(b["uid"], b["nachname"], b["vorname"], kind)
|
|
return list(people.values())
|
|
|
|
|
|
def _search_worker(job: dict, emb, population: list[dict]):
|
|
try:
|
|
for p in population:
|
|
pid = client.person_id_for(p["uid"])
|
|
if pid:
|
|
res = client.photo_bytes(pid)
|
|
if res:
|
|
e = _embed_person(pid, res[0])
|
|
if e is not None:
|
|
sim = engine.similarity(emb, e)
|
|
if sim >= 0.20:
|
|
entry = {**p, "sim": round(sim, 4),
|
|
"img": f"/fotos/img/{pid}"}
|
|
with JOBS_LOCK:
|
|
job["results"].append(entry)
|
|
job["results"].sort(key=lambda r: r["sim"], reverse=True)
|
|
del job["results"][60:]
|
|
job["done"] += 1
|
|
except Exception as ex:
|
|
job["error"] = str(ex)
|
|
job["status"] = "done"
|
|
job["took"] = round(time.time() - job["created"], 1)
|
|
|
|
|
|
def _start_job(emb, population: list[dict], scope: str) -> dict:
|
|
job = {
|
|
"id": uuid.uuid4().hex[:12],
|
|
"status": "running",
|
|
"done": 0,
|
|
"total": len(population),
|
|
"results": [],
|
|
"error": None,
|
|
"scope": scope,
|
|
"created": time.time(),
|
|
}
|
|
with JOBS_LOCK:
|
|
JOBS[job["id"]] = job
|
|
while len(JOBS) > MAX_JOBS:
|
|
oldest = min(JOBS, key=lambda k: JOBS[k]["created"])
|
|
JOBS.pop(oldest, None)
|
|
threading.Thread(target=_search_worker, args=(job, emb, population), daemon=True).start()
|
|
return job
|
|
|
|
|
|
@app.post("/fotos/suche", response_class=HTMLResponse)
|
|
def fotos_suche(request: Request,
|
|
kz: str = Form("640"),
|
|
sem: str = Form("1"),
|
|
ver: str = Form("V"),
|
|
scope: str = Form("verband"),
|
|
file: UploadFile = File(...)):
|
|
if scope not in SCOPE_LABELS:
|
|
scope = "verband"
|
|
meta = _meta()
|
|
sems = _study_sem_options(kz, current_semester(meta["semesters"]))
|
|
if sem not in [str(n) for n in sems]:
|
|
sem = str(sems[0])
|
|
ctx = {
|
|
"request": request,
|
|
"kz": kz, "sem": sem, "sems": sems, "ver": ver,
|
|
"scope": scope,
|
|
"scope_label": SCOPE_LABELS[scope],
|
|
"meta": meta,
|
|
"stg": next((x for x in meta["stgs"] if x["kz"] == kz), None),
|
|
"verbands": client.verbands_for(kz, sem),
|
|
"own_uid": OWN_UID,
|
|
}
|
|
data = file.file.read()
|
|
if not data:
|
|
return templates.TemplateResponse(request, "fotos_search.html",
|
|
{**ctx, "error": "Leere Datei hochgeladen.", "job": None},
|
|
status_code=400)
|
|
if len(data) > MAX_UPLOAD:
|
|
return templates.TemplateResponse(request, "fotos_search.html",
|
|
{**ctx, "error": "Datei zu groß (max. 8 MB).", "job": None},
|
|
status_code=400)
|
|
emb = engine.embed(data)
|
|
if emb is None:
|
|
return templates.TemplateResponse(request, "fotos_search.html",
|
|
{**ctx, "error": "Kein Gesicht im hochgeladenen Bild gefunden.", "job": None},
|
|
status_code=422)
|
|
population = _scope_population(scope, kz, sem, ver)
|
|
job = _start_job(emb, population, scope)
|
|
return templates.TemplateResponse(request, "fotos_search.html",
|
|
{**ctx, "error": None, "job": job})
|
|
|
|
|
|
@app.get("/api/suche/status/{job_id}")
|
|
def suche_status(job_id: str):
|
|
job = JOBS.get(job_id)
|
|
if not job:
|
|
return JSONResponse({"error": "Job unbekannt oder abgelaufen"}, status_code=404)
|
|
with JOBS_LOCK:
|
|
out = {
|
|
"status": job["status"],
|
|
"done": job["done"],
|
|
"total": job["total"],
|
|
"error": job["error"],
|
|
"took": job.get("took"),
|
|
"top": job["results"][:30],
|
|
}
|
|
return JSONResponse(out)
|
|
|
|
|
|
@app.get("/person/{uid}", response_class=HTMLResponse)
|
|
def person_page(request: Request, uid: str):
|
|
prof = client.profile(uid)
|
|
if not prof:
|
|
return templates.TemplateResponse(request, "person.html", {
|
|
"request": request, "prof": None, "uid": uid, "meta": _meta(),
|
|
}, status_code=404)
|
|
meta = _meta()
|
|
dec = decode_matrikel(uid, meta["stgs"]) if uid.isdigit() else None
|
|
return templates.TemplateResponse(request, "person.html", {
|
|
"request": request,
|
|
"prof": prof,
|
|
"uid": uid,
|
|
"dec": dec,
|
|
"meta": meta,
|
|
"own_uid": OWN_UID,
|
|
})
|
|
|
|
|
|
@app.get("/api/person_id")
|
|
def api_person_id(uid: str = Query(...)):
|
|
return {"uid": uid, "person_id": client.person_id_for(uid)}
|
|
|
|
|
|
@app.get("/api/benutzer")
|
|
def api_benutzer(q: str = Query("")):
|
|
"""Namensuche über Studierende + Lektoren (Autocomplete)."""
|
|
q = (q or "").strip().lower()
|
|
if len(q) < 2:
|
|
return []
|
|
out = []
|
|
for b in client.benutzer_all():
|
|
hay = f"{b['nachname']} {b['vorname']} {b['uid']}".lower()
|
|
if q in hay:
|
|
out.append({"uid": b["uid"],
|
|
"name": f"{b['nachname']}, {b['vorname']}".strip(", "),
|
|
"kind": "student" if b["uid"].isdigit() else "lektor"})
|
|
if len(out) >= 12:
|
|
break
|
|
return out
|
|
|
|
|
|
@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":
|
|
uid = uid or OWN_UID
|
|
ptype = "student" if uid.isdigit() else "lektor"
|
|
ical = client.ical_raw(ptype, uid=uid, begin=s["begin"], ende=s["ende"])
|
|
fname = f"lvplan_{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}
|