christian/busylight_modular
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: AIDA -> modulare Zeiterfassung

Browse Source 
- Alle AIDA-Referenzen durch generische Begriffe ersetzt
  (aida -> attendance im Code, "Zeiterfassung" im UI)
- DB-Spalten: last_aida_status -> last_attendance_status
- Settings: aida_csv_path -> attendance_csv_path
- REST API fuer Zeiterfassung auf Port 8081:
  POST /api/attendance/{personal_nr} mit {"status": "anwesend"}
  GET  /api/attendance - alle Status abfragen
- Automatische DB-Migration fuer bestehende Installationen
- CSV bleibt als Option, API als Echtzeit-Alternative

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
christian
2026-05-03 10:28:45 +02:00
co-authored by Claude Opus 4.6
parent 12aa2d7ea1
commit 0d080ff885
8 changed files with 215 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webhook.py
+109 -18
View File
@@ -3,10 +3,9 @@
import logging
from datetime import datetime
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from fastapi.responses import PlainTextResponse, JSONResponse
from database import get_db, log_activity
from wled_client import build_wled_url, call_wled
logger = logging.getLogger(__name__)
@@ -15,15 +14,20 @@ webhook_app = FastAPI(title="Busylight Webhook")
@webhook_app.get("/", response_class=PlainTextResponse)
async def webhook_info():
return """Busylight Webhook API (Telefonanlage)
return """Busylight Webhook API
Endpoints:
Telefon-Endpoints:
GET/POST /{durchwahl}/callstart - Anruf gestartet (WLED -> Rot)
GET/POST /{durchwahl}/callend - Anruf beendet (WLED -> zurueck auf Zeiterfassungs-Status)
Beispiel:
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
"""
@@ -33,7 +37,7 @@ def _get_user_by_extension(extension, conn):
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
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
@@ -55,6 +59,8 @@ def _get_color(status, conn):
@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)
@@ -62,11 +68,11 @@ async def call_start(extension: str):
logger.warning("Webhook callstart: Durchwahl %s nicht gefunden", extension)
return f"Unknown extension: {extension}"
aida = user["last_aida_status"] or ""
attendance = user["last_attendance_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})"
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)
@@ -84,7 +90,7 @@ async def call_start(extension: str):
log_activity(
user["azure_id"], user["display_name"],
aida, "phone_active", url, result_text,
attendance, "phone_active", url, result_text,
)
logger.info("Webhook callstart: %s -> %s", user["display_name"], result_text)
@@ -97,6 +103,8 @@ async def call_start(extension: str):
@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)
@@ -104,15 +112,14 @@ async def call_end(extension: str):
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"):
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
# (Teams-Farbe kommt automatisch beim naechsten Polling)
if aida == "anwesend":
if attendance == "anwesend":
color = _get_color("available", conn)
elif aida == "homeoffice":
elif attendance == "homeoffice":
color = _get_color("homeoffice", conn)
else:
color = _get_color("abwesend", conn)
@@ -131,11 +138,95 @@ async def call_end(extension: str):
log_activity(
user["azure_id"], user["display_name"],
aida, "phone_end", url, result_text,
attendance, "phone_end", url, result_text,
)
logger.info("Webhook callend: %s -> %s (%s)", user["display_name"], result_text, aida)
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()