jason/unifi-access-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #5 from jasonMPM/v6

Browse Source 
Add files via upload
This commit was merged in pull request #5.
This commit is contained in:
jasonMPM
2026-03-04 21:44:23 -06:00
committed by GitHub
parent ab400b78bd 0b2a85d91c
commit 19b43d93a6
2 changed files with 140 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

328
app.py
View File

@@ -1,254 +1,180 @@
from flask import Flask, request, jsonify, send_from_directory import os, hmac, hashlib, json, logging
from dotenv import load_dotenv from flask import Flask, request, jsonify
from datetime import datetime
import pytz, sqlite3
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
import sqlite3 import requests, urllib3
import datetime as dt
import requests
import urllib3
import os
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) urllib3.disable_warnings()
load_dotenv()
app = Flask(__name__) logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
app = Flask(__name__, static_folder="static", static_url_path="")
UNIFI_HOST = os.environ.get("UNIFI_HOST", "10.0.0.1")
UNIFI_PORT = int(os.environ.get("UNIFI_PORT", "12445"))
UNIFI_TOKEN = os.environ.get("UNIFI_API_TOKEN", "")
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "")
DB_PATH = os.environ.get("DB_PATH", "/data/dashboard.db") DB_PATH = os.environ.get("DB_PATH", "/data/dashboard.db")
UNIFI_HOST = os.environ.get("UNIFI_HOST", "") TZ = os.environ.get("TZ", "America/Chicago")
UNIFI_API_TOKEN = os.environ.get("UNIFI_API_TOKEN", "")
UNIFI_BASE = f"https://{UNIFI_HOST}:{UNIFI_PORT}/api/v1/developer"
# ── Database ──────────────────────────────────────────────────────────────────
def get_db(): def get_db():
conn = sqlite3.connect(DB_PATH) conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
return conn return conn
def init_db(): def init_db():
conn = get_db() with get_db() as db:
conn.executescript(""" db.execute("""
CREATE TABLE IF NOT EXISTS unlocks ( CREATE TABLE IF NOT EXISTS badge_events (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
actor_id TEXT NOT NULL,
ts TEXT NOT NULL, ts TEXT NOT NULL,
event TEXT NOT NULL DEFAULT 'access.door.unlock', date TEXT NOT NULL
door_name TEXT, )
device_name TEXT,
actor_id TEXT,
actor_name TEXT,
resolved_name TEXT,
auth_type TEXT,
result TEXT
);
CREATE TABLE IF NOT EXISTS access_users (
id TEXT PRIMARY KEY,
name TEXT,
email TEXT,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
""") """)
for col in ("resolved_name", "device_name"): db.execute("""
try: CREATE TABLE IF NOT EXISTS user_cache (
conn.execute(f"ALTER TABLE unlocks ADD COLUMN {col} TEXT") actor_id TEXT PRIMARY KEY,
except Exception: full_name TEXT NOT NULL,
pass updated_at TEXT NOT NULL
conn.commit() )
conn.close() """)
db.commit()
# ── UniFi user cache ──────────────────────────────────────────────────────────
def sync_unifi_users(): def sync_unifi_users():
if not UNIFI_HOST or not UNIFI_API_TOKEN:
print("[UniFi Sync] UNIFI_HOST or UNIFI_API_TOKEN not set — skipping.")
return
url = f"https://{UNIFI_HOST}:45/api1/users"
headers = {"Authorization": f"Bearer {UNIFI_API_TOKEN}"}
try: try:
resp = requests.get(url, headers=headers, verify=False, timeout=10) r = requests.get(
resp.raise_for_status() f"{UNIFI_BASE}/users",
users = resp.json().get("data", []) headers={"Authorization": f"Bearer {UNIFI_TOKEN}"},
conn = get_db() verify=False, timeout=10
for user in users:
conn.execute(
"""INSERT INTO access_users (id, name, email, updated_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(id) DO UPDATE SET
name=excluded.name,
email=excluded.email,
updated_at=excluded.updated_at""",
(user.get("id"), user.get("name"), user.get("email", "")),
) )
conn.commit() if r.status_code != 200:
conn.close() log.warning("User sync failed: %s %s", r.status_code, r.text[:200])
print(f"[UniFi Sync] Cached {len(users)} users.") return
users = r.json().get("data", [])
with get_db() as db:
for u in users:
db.execute("""
INSERT INTO user_cache (actor_id, full_name, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(actor_id) DO UPDATE SET
full_name = excluded.full_name,
updated_at = excluded.updated_at
""", (u["id"], u.get("full_name", "").strip() or
f"{u.get('first_name','')} {u.get('last_name','')}".strip(),
datetime.utcnow().isoformat()))
db.commit()
log.info("Synced %d users from UniFi Access", len(users))
except Exception as e: except Exception as e:
print(f"[UniFi Sync] Failed: {e}") log.error("sync_unifi_users error: %s", e)
def verify_signature(payload_bytes, sig_header):
"""Return True if HMAC-SHA256 signature matches, or if no secret configured."""
if not WEBHOOK_SECRET:
return True
expected = hmac.new(WEBHOOK_SECRET.encode(), payload_bytes, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, sig_header or "")
def resolve_user_name(user_id: str) -> str: @app.route("/")
if not user_id: def index():
return "Unknown" return app.send_static_file("index.html")
conn = get_db()
row = conn.execute(
"SELECT name FROM access_users WHERE id = ?", (user_id,)
).fetchone()
conn.close()
return row["name"] if row else f"Unknown ({user_id[:8]}…)"
@app.route("/api/unifi-access", methods=["POST"])
def receive_webhook():
raw = request.get_data()
# ── Webhook receiver ────────────────────────────────────────────────────────── # Optional signature verification
sig = request.headers.get("X-Signature-SHA256", "")
if not verify_signature(raw, sig):
log.warning("Webhook signature mismatch")
return jsonify({"error": "invalid signature"}), 401
@app.post("/api/unifi-access") try:
def unifi_access_webhook(): payload = json.loads(raw)
payload = request.get_json(force=True, silent=True) or {} except Exception:
ts = dt.datetime.utcnow().isoformat(timespec="seconds") + "Z" return jsonify({"error": "bad json"}), 400
if payload.get("event") == "access.door.unlock": log.info("Webhook received: %s", json.dumps(payload)[:300])
data = payload.get("data", {})
actor = data.get("actor", {})
location = data.get("location", {})
device = data.get("device", {})
obj = data.get("object", {})
actor_id = actor.get("id", "")
actor_name = actor.get("name") or resolve_user_name(actor_id)
door_name = location.get("name")
device_name = device.get("name")
auth_type = obj.get("authentication_type")
result = obj.get("result", "Access Granted")
elif "events" in payload: # Support both UniFi Access developer API and legacy Alarm Manager formats
event = payload["events"][0] event = payload.get("event") or payload.get("event_object_id", "")
actor_id = event.get("user", "") actor = (payload.get("actor") or {}).get("id") or payload.get("actor_id", "")
actor_name = event.get("user_name") or resolve_user_name(actor_id) ts_raw = payload.get("timestamp") or payload.get("created_at") or datetime.utcnow().isoformat()
door_name = event.get("location_name") or event.get("location", "Unknown Door")
device_name = None
auth_type = event.get("unlock_method_text", "Unknown")
result = "Access Granted"
else: if "door.unlock" not in str(event) and "access.door.unlock" not in str(event):
return "", 204 return jsonify({"status": "ignored"}), 200
conn = get_db() if not actor:
conn.execute( return jsonify({"error": "no actor"}), 400
"""INSERT INTO unlocks
(ts, event, door_name, device_name, actor_id, tz = pytz.timezone(TZ)
actor_name, resolved_name, auth_type, result) ts = datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00"))
VALUES (?, 'access.door.unlock', ?, ?, ?, ?, ?, ?, ?)""", ts_l = ts.astimezone(tz)
(ts, door_name, device_name, actor_id, actor_name, date = ts_l.strftime("%Y-%m-%d")
actor_name, auth_type, result), ts_s = ts_l.strftime("%H:%M:%S")
with get_db() as db:
db.execute(
"INSERT INTO badge_events (actor_id, ts, date) VALUES (?, ?, ?)",
(actor, ts_s, date)
) )
conn.commit() db.commit()
conn.close()
return "", 204
log.info("Badge-in recorded: actor=%s date=%s ts=%s", actor, date, ts_s)
return jsonify({"status": "ok"}), 200
# ── Dashboard data API ──────────────────────────────────────────────────────── @app.route("/api/first-badge-status")
@app.get("/api/first-badge-status")
def first_badge_status(): def first_badge_status():
date = request.args.get("date") or dt.date.today().isoformat() date = request.args.get("date", datetime.now().strftime("%Y-%m-%d"))
cutoff = request.args.get("cutoff", "09:00") cutoff = request.args.get("cutoff", "09:00")
start = f"{date}T00:00:00Z"
end = f"{date}T23:59:59Z"
conn = get_db() with get_db() as db:
# First badge per person rows = db.execute("""
first_rows = conn.execute( SELECT
"""SELECT COALESCE(resolved_name, actor_name, actor_id) AS display_name, b.actor_id,
actor_id, MIN(b.ts) AS first_ts,
MIN(ts) AS first_ts MAX(b.ts) AS latest_ts,
FROM unlocks COALESCE(u.full_name, 'Unknown (' || SUBSTR(b.actor_id,1,8) || '...)') AS name
WHERE event = 'access.door.unlock' FROM badge_events b
AND result = 'Access Granted' LEFT JOIN user_cache u ON u.actor_id = b.actor_id
AND ts BETWEEN ? AND ? WHERE b.date = ?
AND (actor_id IS NOT NULL OR actor_name IS NOT NULL) GROUP BY b.actor_id
GROUP BY actor_id ORDER BY first_ts ASC
ORDER BY first_ts""", """, (date,)).fetchall()
(start, end),
).fetchall()
# Latest badge per person (may equal first if they only badged once)
latest_rows = conn.execute(
"""SELECT actor_id, MAX(ts) AS latest_ts
FROM unlocks
WHERE event = 'access.door.unlock'
AND result = 'Access Granted'
AND ts BETWEEN ? AND ?
AND actor_id IS NOT NULL
GROUP BY actor_id""",
(start, end),
).fetchall()
conn.close()
latest_map = {r["actor_id"]: r["latest_ts"] for r in latest_rows}
result = [] result = []
for r in first_rows: for r in rows:
t_first = dt.datetime.fromisoformat(r["first_ts"].replace("Z", "+00:00")) first = r["first_ts"]
first_time = t_first.strftime("%H:%M") latest = r["latest_ts"]
status = "ON TIME" if first <= cutoff + ":59" else "LATE"
latest_ts = latest_map.get(r["actor_id"], r["first_ts"])
t_latest = dt.datetime.fromisoformat(latest_ts.replace("Z", "+00:00"))
latest_time = t_latest.strftime("%H:%M")
result.append({ result.append({
"actor_name": r["display_name"],
"actor_id": r["actor_id"], "actor_id": r["actor_id"],
"first_badge": r["first_ts"], "name": r["name"],
"badge_time": first_time, "first_ts": first,
"latest_badge": latest_ts, "latest_ts": latest if latest != first else None,
"latest_time": latest_time, "status": status
"on_time": first_time <= cutoff,
}) })
return jsonify(result) return jsonify(result)
@app.route("/api/sync-users")
# ── Sync users ────────────────────────────────────────────────────────────────
@app.get("/api/sync-users")
def manual_sync(): def manual_sync():
sync_unifi_users() sync_unifi_users()
return jsonify({"status": "ok"}) return jsonify({"status": "synced"})
@app.route("/api/reset-day", methods=["DELETE"])
# ── Reset day (testing only) ──────────────────────────────────────────────────
@app.delete("/api/reset-day")
def reset_day(): def reset_day():
"""Delete all unlock records for a given date (defaults to today). date = request.args.get("date", datetime.now().strftime("%Y-%m-%d"))
Intended for development/testing only. with get_db() as db:
""" cur = db.execute("DELETE FROM badge_events WHERE date = ?", (date,))
date = request.args.get("date") or dt.date.today().isoformat() db.commit()
start = f"{date}T00:00:00Z" return jsonify({"status": "ok", "deleted": cur.rowcount, "date": date})
end = f"{date}T23:59:59Z"
conn = get_db()
res = conn.execute(
"DELETE FROM unlocks WHERE ts BETWEEN ? AND ?", (start, end)
)
conn.commit()
deleted = res.rowcount
conn.close()
return jsonify({"status": "ok", "deleted": deleted, "date": date})
# ── Static files ──────────────────────────────────────────────────────────────
@app.get("/")
def index():
return send_from_directory("static", "index.html")
@app.get("/static/<path:path>")
def send_static(path):
return send_from_directory("static", path)
# ── Startup ───────────────────────────────────────────────────────────────────
# Initialise DB and kick off background scheduler at import time # Initialise DB and kick off background scheduler at import time
# (works whether started via `flask run` or `python app.py`)
with app.app_context(): with app.app_context():
init_db() init_db()
sync_unifi_users() sync_unifi_users()

11
docker-compose.yml
View File

@@ -1,6 +1,6 @@
version: "3.9" version: "3.8"
services: services:
unifi-access-dashboard: dashboard:
build: . build: .
container_name: unifi-access-dashboard container_name: unifi-access-dashboard
restart: unless-stopped restart: unless-stopped
@@ -11,4 +11,9 @@ services:
env_file: env_file:
- .env - .env
environment: environment:
- TZ=America/Chicago - UNIFI_HOST=${UNIFI_HOST}
- UNIFI_PORT=${UNIFI_PORT:-12445}
- UNIFI_API_TOKEN=${UNIFI_API_TOKEN}
- WEBHOOK_SECRET=${WEBHOOK_SECRET}
- TZ=${TZ:-America/Chicago}
- DB_PATH=${DB_PATH:-/data/dashboard.db}