christianandClaude Opus 5
086b21dc2b
Das Busylight im Empfangsbereich wird wechselnd besetzt. Mitarbeiter
koennen sich jetzt selbst darauf setzen, ohne Admin-Passwort.
- Neue Seite /zentrale (Durchwahl eingeben) und /zentrale/{durchwahl}
als Direktlink fuer QR-Codes. Der Direktlink zeigt nur an und aendert
nichts, damit Link-Vorschau und Prefetch niemanden versetzen.
- Neue Tabelle home_assignments merkt das eigene Tuerschild, solange der
Mitarbeiter in der Zentrale sitzt. Beim Verlassen wird es
wiederhergestellt, das Zentral-Segment wird freigegeben.
- Der Vorgaenger bleibt ohne Zuordnung - er ist nicht da. Sein gemerkter
Heimatplatz bleibt erhalten, er kann jederzeit zurueckkehren.
- Ist der Heimatplatz inzwischen belegt, wird die Rueckkehr abgelehnt
und nichts geaendert.
- Eine ausdrueckliche Admin-Zuordnung verwirft den gemerkten Heimatplatz.
Ein reines Loeschen der Zuordnung nicht, sonst kaeme niemand mehr von
der Zentrale zurueck.
- Farbe wird sofort ans Geraet geschickt (scheduler.push_user_light).
Das Polling allein reicht nicht: _process_user sendet nur bei einer
Statusaenderung, und beim Platzwechsel bleibt der Status gleich.
- Neue Settings selfservice_device_id und selfservice_segment,
einstellbar unter Einstellungen.
- Jeder Wechsel landet im Protokoll (zentrale / zentrale_frei).
Die Durchwahl ist eine Kennung, kein Geheimnis - im README dokumentiert.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
290 lines
10 KiB
Python
290 lines
10 KiB
Python
import logging
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from database import get_db, get_setting, log_activity
|
|
from wled_client import build_wled_url, call_wled, get_wled_info
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Global reference to graph client (set by app.py on startup)
|
|
graph = None
|
|
|
|
|
|
def read_attendance_csv(csv_path):
|
|
"""Read attendance status CSV. Returns list of lines."""
|
|
if not csv_path or not os.path.exists(csv_path):
|
|
return []
|
|
try:
|
|
with open(csv_path, "r", encoding="utf-8", errors="replace") as f:
|
|
lines = f.read().splitlines()
|
|
return lines
|
|
except Exception as e:
|
|
logger.error("Fehler beim Lesen der Zeiterfassungs-CSV: %s", e)
|
|
return []
|
|
|
|
|
|
def _extract_nr(value):
|
|
"""Extract bare number from ~~00341~~ -> 00341."""
|
|
return value.strip().strip("~")
|
|
|
|
|
|
def determine_attendance_status(lines, personal_nr, settings):
|
|
"""Determine attendance status for a user based on their personal number."""
|
|
if not personal_nr or not lines:
|
|
return None
|
|
|
|
nr = _extract_nr(personal_nr)
|
|
if not nr:
|
|
return None
|
|
|
|
# Find line containing this personal number
|
|
matching = [l for l in lines if nr in l]
|
|
if not matching:
|
|
return None
|
|
|
|
line = matching[0]
|
|
# Check status part (after comma or at end)
|
|
status_abwesend = _extract_nr(settings.get("attendance_status_abwesend", "~~0~~"))
|
|
status_anwesend = _extract_nr(settings.get("attendance_status_anwesend", "~~1~~"))
|
|
status_homeoffice = _extract_nr(settings.get("attendance_status_homeoffice", "~~2~~"))
|
|
|
|
# Get the status value (last part after comma, stripped of ~)
|
|
parts = line.split(",")
|
|
status_val = _extract_nr(parts[-1]) if len(parts) > 1 else ""
|
|
|
|
if status_val == status_abwesend:
|
|
return "abwesend"
|
|
if status_val == status_anwesend:
|
|
return "anwesend"
|
|
if status_val == status_homeoffice:
|
|
return "homeoffice"
|
|
return None
|
|
|
|
|
|
def get_color_for_status(status, conn):
|
|
"""Get RGB color for a given status from color_rules."""
|
|
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": ""}
|
|
|
|
|
|
def current_color_for_user(azure_id, conn):
|
|
"""Farbe fuer den zuletzt gespeicherten Status eines Users.
|
|
|
|
Prioritaet wie im Polling und im Telefon-Webhook: Telefon > Teams > Zeiterfassung.
|
|
Wer laut Zeiterfassung nicht anwesend/homeoffice ist, bekommt die abwesend-Farbe.
|
|
"""
|
|
row = conn.execute("""
|
|
SELECT last_attendance_status, last_teams_status,
|
|
COALESCE(phone_active, 0) as phone_active
|
|
FROM state WHERE user_azure_id = ?
|
|
""", (azure_id,)).fetchone()
|
|
|
|
attendance = (row["last_attendance_status"] or "") if row else ""
|
|
teams = (row["last_teams_status"] or "") if row else ""
|
|
phone_active = row["phone_active"] if row else 0
|
|
|
|
if attendance not in ("anwesend", "homeoffice"):
|
|
return get_color_for_status("abwesend", conn)
|
|
if phone_active:
|
|
return get_color_for_status("phone_active", conn)
|
|
if attendance == "homeoffice":
|
|
return get_color_for_status("homeoffice", conn)
|
|
if teams:
|
|
return get_color_for_status(teams, conn)
|
|
return get_color_for_status("available", conn)
|
|
|
|
|
|
def push_user_light(azure_id, display_name, hostname, segment, conn, reason="zentrale"):
|
|
"""Aktuelle Farbe eines Users sofort an WLED senden.
|
|
|
|
Wird beim Platzwechsel gebraucht: _process_user schickt nur bei einer
|
|
Statusaenderung etwas ans Geraet, nach einem Wechsel ist der Status aber
|
|
unveraendert. Ohne diesen Push wuerde das Segment die Farbe des Vorgaengers
|
|
behalten, bis sich zufaellig etwas aendert.
|
|
"""
|
|
color = current_color_for_user(azure_id, conn)
|
|
url = build_wled_url(
|
|
hostname, color["r"], color["g"], color["b"],
|
|
segment=segment, effect=color.get("effect", ""),
|
|
)
|
|
success, result_text = call_wled(url)
|
|
log_activity(azure_id, display_name, reason, "-", url, result_text)
|
|
if not success:
|
|
logger.warning("WLED Fehler beim Platzwechsel fuer %s: %s", display_name, result_text)
|
|
return success, result_text
|
|
|
|
|
|
def clear_segment(hostname, segment, conn, reason="zentrale_frei"):
|
|
"""Segment auf die abwesend-Farbe setzen (Standard ist 0,0,0 = aus)."""
|
|
color = get_color_for_status("abwesend", conn)
|
|
url = build_wled_url(
|
|
hostname, color["r"], color["g"], color["b"],
|
|
segment=segment, effect=color.get("effect", ""),
|
|
)
|
|
success, result_text = call_wled(url)
|
|
log_activity(None, "-", reason, "-", url, result_text)
|
|
if not success:
|
|
logger.warning("WLED Fehler beim Freigeben von %s Segment %s: %s", hostname, segment, result_text)
|
|
return success, result_text
|
|
|
|
|
|
def poll_status():
|
|
"""Main polling function - called periodically by APScheduler."""
|
|
global graph
|
|
|
|
if graph is None or not graph.is_authenticated():
|
|
logger.warning("Graph-Client nicht authentifiziert, ueberspringe Polling")
|
|
return
|
|
|
|
conn = get_db()
|
|
try:
|
|
settings = {}
|
|
for row in conn.execute("SELECT key, value FROM settings").fetchall():
|
|
settings[row["key"]] = row["value"]
|
|
|
|
# Get all active users with assignments
|
|
users_with_assignments = conn.execute("""
|
|
SELECT u.azure_id, u.display_name, u.personal_nr,
|
|
a.segment, d.hostname,
|
|
s.last_attendance_status, s.last_teams_status,
|
|
COALESCE(s.phone_active, 0) as 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.active = 1
|
|
""").fetchall()
|
|
|
|
if not users_with_assignments:
|
|
logger.debug("Keine aktiven User mit Zuordnungen gefunden")
|
|
return
|
|
|
|
# Read attendance CSV (kann leer sein)
|
|
attendance_lines = read_attendance_csv(settings.get("attendance_csv_path", ""))
|
|
|
|
for user in users_with_assignments:
|
|
try:
|
|
_process_user(user, attendance_lines, settings, conn)
|
|
except Exception as e:
|
|
logger.error(
|
|
"Fehler bei User %s: %s", user["display_name"], e
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Polling Fehler: %s", e)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _process_user(user, attendance_lines, settings, conn):
|
|
"""Process a single user: check status and update WLED if needed."""
|
|
global graph
|
|
|
|
azure_id = user["azure_id"]
|
|
display_name = user["display_name"]
|
|
personal_nr = user["personal_nr"]
|
|
hostname = user["hostname"]
|
|
segment = user["segment"]
|
|
last_attendance = user["last_attendance_status"] or ""
|
|
last_teams = user["last_teams_status"] or ""
|
|
phone_active = user["phone_active"]
|
|
|
|
# Immer Teams-Status abfragen (fuer Dashboard-Anzeige)
|
|
teams_status = ""
|
|
presence = graph.get_presence(azure_id)
|
|
if presence and "activity" in presence:
|
|
teams_status = "teams_" + presence["activity"].lower()
|
|
|
|
# Anwesenheitsstatus ermitteln
|
|
attendance_status = determine_attendance_status(attendance_lines, personal_nr, settings)
|
|
|
|
# Check if anything changed
|
|
if attendance_status == last_attendance and teams_status == last_teams:
|
|
return # No change
|
|
|
|
# --- WLED nur steuern wenn Anwesenheitsstatus bekannt und kein aktives Telefonat ---
|
|
wled_changed = attendance_status is not None and not phone_active and (
|
|
attendance_status != last_attendance or
|
|
(attendance_status == "anwesend" and teams_status != last_teams)
|
|
)
|
|
|
|
if wled_changed:
|
|
if attendance_status == "abwesend":
|
|
color = get_color_for_status("abwesend", conn)
|
|
elif attendance_status == "homeoffice":
|
|
color = get_color_for_status("homeoffice", conn)
|
|
elif attendance_status == "anwesend" and teams_status:
|
|
color = get_color_for_status(teams_status, conn)
|
|
else:
|
|
color = get_color_for_status("available", conn)
|
|
|
|
url = build_wled_url(
|
|
hostname, color["r"], color["g"], color["b"],
|
|
segment=segment, effect=color.get("effect", "")
|
|
)
|
|
success, result_text = call_wled(url)
|
|
|
|
if not success:
|
|
logger.warning("WLED Fehler fuer %s: %s", display_name, result_text)
|
|
|
|
log_activity(
|
|
azure_id, display_name,
|
|
attendance_status or "-", teams_status or "-",
|
|
url, result_text,
|
|
)
|
|
|
|
# State immer aktualisieren (fuer Dashboard)
|
|
conn.execute("""
|
|
INSERT INTO state (user_azure_id, last_attendance_status, last_teams_status, updated_at)
|
|
VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(user_azure_id)
|
|
DO UPDATE SET last_attendance_status=?, last_teams_status=?, updated_at=?
|
|
""", (
|
|
azure_id,
|
|
attendance_status or last_attendance, teams_status, datetime.now(),
|
|
attendance_status or last_attendance, teams_status, datetime.now(),
|
|
))
|
|
conn.commit()
|
|
|
|
logger.info(
|
|
"%-40s zeiterfassung=%-12s teams=%-20s wled=%s",
|
|
display_name, attendance_status or "-", teams_status,
|
|
"aktualisiert" if wled_changed else "nur Dashboard",
|
|
)
|
|
|
|
|
|
def check_devices():
|
|
"""Check all WLED devices and store status in DB. Called periodically."""
|
|
conn = get_db()
|
|
try:
|
|
devices = conn.execute("SELECT id, hostname FROM wled_devices").fetchall()
|
|
for dev in devices:
|
|
info = get_wled_info(dev["hostname"])
|
|
if info.get("online"):
|
|
conn.execute("""
|
|
UPDATE wled_devices
|
|
SET online=1, leds=?, version=?, mac=?, last_check=?
|
|
WHERE id=?
|
|
""", (
|
|
info.get("leds", 0),
|
|
info.get("version", ""),
|
|
info.get("mac", ""),
|
|
datetime.now(),
|
|
dev["id"],
|
|
))
|
|
else:
|
|
conn.execute("""
|
|
UPDATE wled_devices SET online=0, last_check=? WHERE id=?
|
|
""", (datetime.now(), dev["id"]))
|
|
conn.commit()
|
|
logger.info("Device-Check: %d Geraete geprueft", len(devices))
|
|
except Exception as e:
|
|
logger.error("Device-Check Fehler: %s", e)
|
|
finally:
|
|
conn.close()
|