"""Webhook server for phone system integration (runs on port 8081).""" import logging from datetime import datetime from fastapi import FastAPI from fastapi.responses import PlainTextResponse, JSONResponse from database import get_db, log_activity logger = logging.getLogger(__name__) webhook_app = FastAPI(title="Busylight Webhook") @webhook_app.get("/", response_class=PlainTextResponse) async def webhook_info(): return """Busylight Webhook API Telefon-Endpoints: GET/POST /{durchwahl}/callstart - Anruf gestartet (WLED -> Rot) GET/POST /{durchwahl}/callend - Anruf beendet (WLED -> zurueck auf Zeiterfassungs-Status) Zeiterfassungs-Endpoint: POST /api/attendance/{personal_nr} - Anwesenheitsstatus setzen Body: {"status": "anwesend|abwesend|homeoffice"} Beispiele: curl http://localhost:8081/123/callstart curl http://localhost:8081/123/callend curl -X POST -H "Content-Type: application/json" -d '{"status":"anwesend"}' http://localhost:8081/api/attendance/00341 """ def _get_user_by_extension(extension, conn): """Find user and their assignment by phone extension.""" return conn.execute(""" SELECT u.azure_id, u.display_name, a.segment, d.hostname, s.last_attendance_status, s.last_teams_status, s.phone_active FROM users u JOIN assignments a ON a.user_azure_id = u.azure_id JOIN wled_devices d ON d.id = a.wled_device_id LEFT JOIN state s ON s.user_azure_id = u.azure_id WHERE u.phone_extension = ? AND u.active = 1 """, (extension,)).fetchone() def _get_color(status, conn): """Get color for a status.""" row = conn.execute( "SELECT r, g, b, effect FROM color_rules WHERE status = ?", (status,) ).fetchone() if row: return dict(row) return {"r": 200, "g": 200, "b": 200, "effect": ""} @webhook_app.get("/{extension}/callstart", response_class=PlainTextResponse) @webhook_app.post("/{extension}/callstart", response_class=PlainTextResponse) async def call_start(extension: str): from wled_client import build_wled_url, call_wled conn = get_db() try: user = _get_user_by_extension(extension, conn) if not user: logger.warning("Webhook callstart: Durchwahl %s nicht gefunden", extension) return f"Unknown extension: {extension}" attendance = user["last_attendance_status"] or "" # Nur wenn anwesend oder homeoffice if attendance not in ("anwesend", "homeoffice"): logger.info("Webhook callstart: %s ignoriert (Zeiterfassung=%s)", user["display_name"], attendance) return f"Ignored: {user['display_name']} not in office (attendance={attendance})" # WLED auf Telefon-Farbe setzen color = _get_color("phone_active", conn) url = build_wled_url( user["hostname"], color["r"], color["g"], color["b"], segment=user["segment"], effect=color.get("effect", ""), ) success, result_text = call_wled(url) # phone_active Flag setzen conn.execute(""" UPDATE state SET phone_active=1, updated_at=? WHERE user_azure_id=? """, (datetime.now(), user["azure_id"])) conn.commit() log_activity( user["azure_id"], user["display_name"], attendance, "phone_active", url, result_text, ) logger.info("Webhook callstart: %s -> %s", user["display_name"], result_text) return f"OK: {user['display_name']} call started" finally: conn.close() @webhook_app.get("/{extension}/callend", response_class=PlainTextResponse) @webhook_app.post("/{extension}/callend", response_class=PlainTextResponse) async def call_end(extension: str): from wled_client import build_wled_url, call_wled conn = get_db() try: user = _get_user_by_extension(extension, conn) if not user: logger.warning("Webhook callend: Durchwahl %s nicht gefunden", extension) return f"Unknown extension: {extension}" attendance = user["last_attendance_status"] or "" if attendance not in ("anwesend", "homeoffice"): return f"Ignored: {user['display_name']} not in office" # Farbe zuruecksetzen auf Zeiterfassungs-Status if attendance == "anwesend": color = _get_color("available", conn) elif attendance == "homeoffice": color = _get_color("homeoffice", conn) else: color = _get_color("abwesend", conn) url = build_wled_url( user["hostname"], color["r"], color["g"], color["b"], segment=user["segment"], effect=color.get("effect", ""), ) success, result_text = call_wled(url) # phone_active Flag zuruecksetzen conn.execute(""" UPDATE state SET phone_active=0, updated_at=? WHERE user_azure_id=? """, (datetime.now(), user["azure_id"])) conn.commit() log_activity( user["azure_id"], user["display_name"], attendance, "phone_end", url, result_text, ) logger.info("Webhook callend: %s -> %s (%s)", user["display_name"], result_text, attendance) return f"OK: {user['display_name']} call ended" finally: conn.close() # ── REST API fuer Zeiterfassung ─────────────────────────────────────────── @webhook_app.post("/api/attendance/{personal_nr}") async def set_attendance(personal_nr: str, request_body: dict = None): """Set attendance status for a user via REST API. Body: {"status": "anwesend|abwesend|homeoffice"} This allows external time tracking systems to push status updates directly instead of using a CSV file. """ if not request_body or "status" not in request_body: return JSONResponse( {"error": "Body muss {'status': 'anwesend|abwesend|homeoffice'} enthalten"}, status_code=400, ) status = request_body["status"] if status not in ("anwesend", "abwesend", "homeoffice"): return JSONResponse( {"error": f"Ungueltiger Status: {status}. Erlaubt: anwesend, abwesend, homeoffice"}, status_code=400, ) conn = get_db() try: # Find user by personal_nr (strip ~~ if present) nr_clean = personal_nr.strip().strip("~") user = conn.execute(""" SELECT u.azure_id, u.display_name, u.personal_nr, a.segment, d.hostname FROM users u JOIN assignments a ON a.user_azure_id = u.azure_id JOIN wled_devices d ON d.id = a.wled_device_id WHERE u.active = 1 AND REPLACE(REPLACE(u.personal_nr, '~', ''), ' ', '') = ? """, (nr_clean,)).fetchone() if not user: return JSONResponse( {"error": f"Kein aktiver User mit Personalnummer {personal_nr} gefunden"}, status_code=404, ) # Update state conn.execute(""" INSERT INTO state (user_azure_id, last_attendance_status, updated_at) VALUES (?, ?, ?) ON CONFLICT(user_azure_id) DO UPDATE SET last_attendance_status=?, updated_at=? """, ( user["azure_id"], status, datetime.now(), status, datetime.now(), )) conn.commit() logger.info("API Zeiterfassung: %s -> %s", user["display_name"], status) return JSONResponse({ "ok": True, "user": user["display_name"], "status": status, }) finally: conn.close() @webhook_app.get("/api/attendance", response_class=JSONResponse) async def get_all_attendance(): """Get current attendance status of all active users.""" conn = get_db() try: rows = conn.execute(""" SELECT u.display_name, u.personal_nr, COALESCE(s.last_attendance_status, '') as status FROM users u LEFT JOIN state s ON s.user_azure_id = u.azure_id WHERE u.active = 1 ORDER BY u.display_name """).fetchall() return [{"user": r["display_name"], "personal_nr": r["personal_nr"], "status": r["status"] or "unbekannt"} for r in rows] finally: conn.close()