Christian Mueller
7f2858aab6
- New webhook server on port 8081 for phone system (PBX)
- Endpoints: GET/POST /{extension}/callstart and /{extension}/callend
- New 'phone_extension' field on users (Durchwahl)
- Only activates when user is anwesend or homeoffice (AIDA)
- Sets WLED to phone_active color (red) on callstart
- Restores previous Teams/AIDA color on callend
- Polling skips WLED update while phone is active
- Dashboard shows phone badge when user is in a call
- New color rule 'phone_active' (configurable via web UI)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
"""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
|
|
|
|
from database import get_db, log_activity
|
|
from wled_client import build_wled_url, call_wled
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
webhook_app = FastAPI(title="Busylight Webhook")
|
|
|
|
|
|
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_aida_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):
|
|
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}"
|
|
|
|
aida = user["last_aida_status"] or ""
|
|
# Nur wenn anwesend oder homeoffice
|
|
if aida not in ("anwesend", "homeoffice"):
|
|
logger.info("Webhook callstart: %s ignoriert (AIDA=%s)", user["display_name"], aida)
|
|
return f"Ignored: {user['display_name']} not in office (aida={aida})"
|
|
|
|
# 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"],
|
|
aida, "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):
|
|
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}"
|
|
|
|
aida = user["last_aida_status"] or ""
|
|
if aida not in ("anwesend", "homeoffice"):
|
|
return f"Ignored: {user['display_name']} not in office"
|
|
|
|
# Farbe zuruecksetzen basierend auf aktuellem Status
|
|
teams = user["last_teams_status"] or ""
|
|
if aida == "homeoffice":
|
|
color = _get_color("homeoffice", conn)
|
|
elif teams:
|
|
color = _get_color(teams, conn)
|
|
else:
|
|
color = _get_color("available", 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"],
|
|
aida, f"phone_end -> {teams}", url, result_text,
|
|
)
|
|
|
|
logger.info("Webhook callend: %s -> %s (%s)", user["display_name"], result_text, teams)
|
|
return f"OK: {user['display_name']} call ended"
|
|
|
|
finally:
|
|
conn.close()
|