995 lines
37 KiB
Python
995 lines
37 KiB
Python
import os
|
|
import re
|
|
import json
|
|
import psycopg2
|
|
import requests
|
|
import docker as docker_lib
|
|
from datetime import datetime, timezone
|
|
from functools import wraps
|
|
from flask import Flask, render_template, jsonify, Response, request, session, redirect, url_for
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.environ.get("SECRET_KEY", "change-me-in-production")
|
|
# Volume-mounted templates must be picked up without restarting the process
|
|
app.config["TEMPLATES_AUTO_RELOAD"] = True
|
|
|
|
AUTH_DIR = os.environ.get("AUTH_DIR", "/auth")
|
|
|
|
|
|
@app.after_request
|
|
def _no_store_html(res):
|
|
"""Avoid stale dashboard HTML behind proxies / browsers (templates change often)."""
|
|
ct = res.headers.get("Content-Type", "")
|
|
if ct.startswith("text/html"):
|
|
res.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
|
|
res.headers["Pragma"] = "no-cache"
|
|
return res
|
|
|
|
|
|
# ── Auth helpers ──────────────────────────────────────────────────────────────
|
|
|
|
def _pending_path(username):
|
|
return os.path.join(AUTH_DIR, "pending", f"{username}.json")
|
|
|
|
def _approved_path(username):
|
|
return os.path.join(AUTH_DIR, "approved", username)
|
|
|
|
def _get_pending(username):
|
|
p = _pending_path(username)
|
|
if not os.path.exists(p):
|
|
return None
|
|
with open(p) as f:
|
|
return json.load(f)
|
|
|
|
def _is_approved(username):
|
|
return os.path.exists(_approved_path(username))
|
|
|
|
|
|
def _list_dashboard_users():
|
|
"""Approved ha-dashboard logins (filenames under auth/approved/)."""
|
|
d = os.path.join(AUTH_DIR, "approved")
|
|
if not os.path.isdir(d):
|
|
return []
|
|
names = []
|
|
for name in os.listdir(d):
|
|
if name.startswith("."):
|
|
continue
|
|
p = os.path.join(d, name)
|
|
if os.path.isfile(p) or os.path.isdir(p):
|
|
names.append(name)
|
|
return sorted(names, key=str.lower)
|
|
|
|
|
|
def _list_pending_usernames():
|
|
"""Registered accounts awaiting approval (pending/*.json, stem = username)."""
|
|
d = os.path.join(AUTH_DIR, "pending")
|
|
if not os.path.isdir(d):
|
|
return []
|
|
out = []
|
|
for fn in os.listdir(d):
|
|
if fn.startswith(".") or not fn.endswith(".json"):
|
|
continue
|
|
out.append(fn[:-5])
|
|
return sorted(out, key=str.lower)
|
|
|
|
|
|
def _nav_user_chips():
|
|
"""Footer avatars: current session, then approved, then pending-only (deduped)."""
|
|
seen = set()
|
|
chips = []
|
|
|
|
def push(name, role):
|
|
key = (name or "").strip().lower()
|
|
if not key or key in seen:
|
|
return
|
|
seen.add(key)
|
|
chips.append({"name": name.strip(), "role": role})
|
|
|
|
cu = session.get("user")
|
|
if cu:
|
|
push(str(cu), "current")
|
|
|
|
for u in _list_dashboard_users():
|
|
push(u, "approved")
|
|
|
|
for u in _list_pending_usernames():
|
|
push(u, "pending")
|
|
|
|
return chips
|
|
|
|
def _save_pending(username, password):
|
|
os.makedirs(os.path.join(AUTH_DIR, "pending"), exist_ok=True)
|
|
with open(_pending_path(username), "w") as f:
|
|
json.dump({
|
|
"username": username,
|
|
"password_hash": generate_password_hash(password),
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
}, f)
|
|
|
|
def login_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not session.get("user"):
|
|
if request.path.startswith("/api/"):
|
|
return jsonify({"error": "Unauthorized"}), 401
|
|
return redirect(url_for("login_page"))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
# ── Auth routes ───────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login_page():
|
|
if session.get("user"):
|
|
return redirect("/")
|
|
error = None
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip().lower()
|
|
password = request.form.get("password", "")
|
|
user = _get_pending(username)
|
|
if user and check_password_hash(user["password_hash"], password):
|
|
if _is_approved(username):
|
|
session["user"] = username
|
|
return redirect("/")
|
|
error = "Account pending approval."
|
|
else:
|
|
error = "Invalid credentials."
|
|
return render_template("login.html", error=error)
|
|
|
|
@app.route("/register", methods=["GET", "POST"])
|
|
def register_page():
|
|
if session.get("user"):
|
|
return redirect("/")
|
|
error = success = None
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip().lower()
|
|
password = request.form.get("password", "")
|
|
confirm = request.form.get("confirm", "")
|
|
if not username or not password:
|
|
error = "Username and password are required."
|
|
elif password != confirm:
|
|
error = "Passwords do not match."
|
|
elif len(password) < 8:
|
|
error = "Password must be at least 8 characters."
|
|
elif _get_pending(username):
|
|
error = "Username already registered."
|
|
else:
|
|
_save_pending(username, password)
|
|
success = "Registration submitted — awaiting approval."
|
|
return render_template("register.html", error=error, success=success)
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login_page"))
|
|
|
|
DB_HOST = os.environ.get("DB_HOST", "postgres18")
|
|
DB_PORT = os.environ.get("DB_PORT", "5432")
|
|
DB_USER = os.environ.get("DB_USER", "homeassistant")
|
|
DB_PASS = os.environ.get("DB_PASS", "")
|
|
HA_URL = os.environ.get("HA_URL", "http://192.168.1.132:8123")
|
|
HA_TOKEN = os.environ.get("HA_TOKEN", "")
|
|
|
|
_NETWORK_LIGHTS = {
|
|
"light.ac_pro_led", "light.u6_pro_led", "light.uap_ac_m_led", "light.tasmota",
|
|
}
|
|
_GATE_AUTOMATIONS = {
|
|
"automation.gate_door_via_webhooks",
|
|
"automation.gate_via_webhooks_dict",
|
|
"automation.notify_say_when_drive_gate_opens",
|
|
"automation.main_gate_opens_in_fog_or_dark_driveway_light_on",
|
|
}
|
|
_OPENINGS = {
|
|
"binary_sensor.main_door": "Main door",
|
|
"binary_sensor.terrace_door": "Terrace door",
|
|
"binary_sensor.bedroom_window": "Bedroom window",
|
|
"binary_sensor.bathroom_window": "Bathroom window",
|
|
"binary_sensor.stefi_window": "Stefi window",
|
|
"binary_sensor.office_window": "Office window",
|
|
"binary_sensor.window_kitchen_farm": "Kitchen window",
|
|
"binary_sensor.living_room_window": "Living room window",
|
|
"binary_sensor.technicka_window": "Laundry room window",
|
|
}
|
|
_TEMP_SENSORS = {
|
|
"sensor.outside_temperature": "Outside",
|
|
"sensor.netatmo_bedroom_temp": "Bedroom",
|
|
"sensor.netatmo_bedroom_stefi_temp": "Stefi room",
|
|
"sensor.netatmo_living_room_temp": "Living room",
|
|
"sensor.netatmo_bathroom_temp": "Bathroom",
|
|
"sensor.office_th": "Office",
|
|
"sensor.attic_temperature": "Attic",
|
|
"sensor.termostat_temperature": "Thermostat",
|
|
}
|
|
|
|
|
|
def pg_conn(dbname):
|
|
return psycopg2.connect(host=DB_HOST, port=DB_PORT,
|
|
dbname=dbname, user=DB_USER, password=DB_PASS,
|
|
connect_timeout=5)
|
|
|
|
|
|
def ha_headers():
|
|
return {"Authorization": f"Bearer {HA_TOKEN}"}
|
|
|
|
|
|
def parse_backup_log(path="/backup.log"):
|
|
try:
|
|
with open(path) as f:
|
|
lines = f.readlines()
|
|
except FileNotFoundError:
|
|
return {"error": "Log not mounted"}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
runs = []
|
|
current = None
|
|
for line in lines:
|
|
line = line.strip()
|
|
m = re.match(r"--- Backup start: (\d{8}_\d{4}) ---", line)
|
|
if m:
|
|
if current:
|
|
current["ok"] = False
|
|
runs.append(current)
|
|
dt = datetime.strptime(m.group(1), "%Y%m%d_%H%M")
|
|
current = {"start": dt.isoformat(), "finish": None, "ok": False,
|
|
"deleted": None, "size": None, "databases": None, "duration": None}
|
|
continue
|
|
if current:
|
|
dm = re.match(r"Deleting old remote file: (.+)", line)
|
|
if dm:
|
|
current["deleted"] = dm.group(1).strip()
|
|
sm = re.match(r"Dump size: (.+)", line)
|
|
if sm:
|
|
current["size"] = sm.group(1).strip()
|
|
dbm = re.match(r"Databases: (.+)", line)
|
|
if dbm:
|
|
current["databases"] = dbm.group(1).strip()
|
|
durm = re.match(r"Duration: (\d+)s", line)
|
|
if durm:
|
|
current["duration"] = int(durm.group(1))
|
|
fm = re.match(r"--- Backup finished: (.+?) ---", line)
|
|
if fm:
|
|
current["ok"] = True
|
|
current["finish"] = fm.group(1).strip()
|
|
runs.append(current)
|
|
current = None
|
|
if current:
|
|
current["ok"] = False
|
|
runs.append(current)
|
|
|
|
if not runs:
|
|
return {"error": "No records"}
|
|
|
|
runs.reverse()
|
|
last = runs[0]
|
|
return {
|
|
"last_start": last["start"],
|
|
"last_ok": last["ok"],
|
|
"total": len(runs),
|
|
"runs": runs[:30],
|
|
}
|
|
|
|
|
|
# ── Overview ─────────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/overview")
|
|
@login_required
|
|
def api_overview():
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
try:
|
|
resp = requests.get(f"{HA_URL}/api/states", headers=ha_headers(), timeout=8)
|
|
resp.raise_for_status()
|
|
all_states = {s["entity_id"]: s for s in resp.json()}
|
|
|
|
def get(eid):
|
|
s = all_states.get(eid)
|
|
return s["state"] if s else None
|
|
|
|
persons = []
|
|
for eid in ["person.seba", "person.stefi"]:
|
|
s = all_states.get(eid)
|
|
if s:
|
|
persons.append({
|
|
"id": eid,
|
|
"name": s["attributes"].get("friendly_name", eid.replace("person.", "")),
|
|
"state": s["state"],
|
|
})
|
|
|
|
open_list = [
|
|
{"id": eid, "name": name}
|
|
for eid, name in _OPENINGS.items()
|
|
if get(eid) == "on"
|
|
]
|
|
|
|
def safe_float(v):
|
|
try:
|
|
return round(float(v), 1) if v not in (None, "unavailable", "unknown") else None
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
mazda = {
|
|
"charge_level": safe_float(get("sensor.sebastians_mazda_charge_level")),
|
|
"fuel_pct": safe_float(get("sensor.sebastians_mazda_fuel_remaining_percentage")),
|
|
"range_bev": safe_float(get("sensor.sebastians_mazda_remaining_range_bev")),
|
|
"range_total": safe_float(get("sensor.sebastians_mazda_remaining_range")),
|
|
"odometer": safe_float(get("sensor.sebastians_mazda_odometer")),
|
|
"charging": get("switch.sebastians_mazda_charging") == "on",
|
|
"plugged_in": get("binary_sensor.sebastians_mazda_plugged_in") == "on",
|
|
"climate": get("climate.sebastians_mazda_climate"),
|
|
"lock": get("lock.sebastians_mazda_lock"),
|
|
"charging_time": safe_float(get("sensor.sebastians_mazda_remaining_charging_time_ac")),
|
|
}
|
|
|
|
temperatures = []
|
|
for eid, label in _TEMP_SENSORS.items():
|
|
v = safe_float(get(eid))
|
|
if v is not None:
|
|
temperatures.append({"room": label, "temp": v})
|
|
|
|
return jsonify({
|
|
"persons": persons,
|
|
"open_openings": open_list,
|
|
"mazda": mazda,
|
|
"temperatures": temperatures,
|
|
"backup": parse_backup_log(),
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
# ── Power ─────────────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/power")
|
|
@login_required
|
|
def api_power():
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
try:
|
|
resp = requests.get(f"{HA_URL}/api/states", headers=ha_headers(), timeout=8)
|
|
resp.raise_for_status()
|
|
st = {s["entity_id"]: s["state"] for s in resp.json()}
|
|
|
|
def fw(eid):
|
|
try:
|
|
v = st.get(eid)
|
|
return round(float(v), 2) if v not in (None, "unavailable", "unknown") else 0.0
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
devices = [
|
|
{"name": "Main (clamp 1)", "watts": fw("sensor.clamp_1_power"), "kwh": fw("sensor.clamp_1_energy")},
|
|
{"name": "NAS/Server", "watts": fw("sensor.hs110_current_consumption"), "kwh": fw("sensor.hs110_today_s_consumption")},
|
|
{"name": "Shelly PM 1", "watts": fw("sensor.shelly1pmg3_dcda0cde8be8_power"), "kwh": fw("sensor.shelly1pmg3_dcda0cde8be8_energy")},
|
|
{"name": "Shelly PM 2", "watts": fw("sensor.shelly1pmg3_5432045a23a0_power"), "kwh": fw("sensor.shelly1pmg3_5432045a23a0_energy")},
|
|
{"name": "Tasmota", "watts": fw("sensor.tasmota_energy_power_2"), "kwh": fw("sensor.tasmota_energy_today_2")},
|
|
]
|
|
main = fw("sensor.clamp_1_power")
|
|
return jsonify({"main_w": main, "devices": devices})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
# ── Frigate ───────────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/frigate")
|
|
@login_required
|
|
def api_frigate():
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
try:
|
|
resp = requests.get(f"{HA_URL}/api/states", headers=ha_headers(), timeout=8)
|
|
resp.raise_for_status()
|
|
detections = []
|
|
for s in resp.json():
|
|
eid = s["entity_id"]
|
|
if not eid.startswith("image.frigate_"):
|
|
continue
|
|
when = s["state"]
|
|
if when in ("unavailable", "unknown"):
|
|
continue
|
|
# entity_id format: image.frigate_<camera>_<object>
|
|
suffix = eid[len("image.frigate_"):]
|
|
parts = suffix.rsplit("_", 1)
|
|
if len(parts) != 2:
|
|
continue
|
|
camera, obj = parts
|
|
detections.append({
|
|
"entity_id": eid,
|
|
"camera": camera.replace("_", " ").title(),
|
|
"object": obj,
|
|
"when": when,
|
|
"image_path": s["attributes"].get("entity_picture", ""),
|
|
})
|
|
detections.sort(key=lambda x: x["when"], reverse=True)
|
|
return jsonify(detections[:9])
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
# ── Crowdsec ──────────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/crowdsec")
|
|
@login_required
|
|
def api_crowdsec():
|
|
try:
|
|
client = docker_lib.DockerClient(base_url="unix:///var/run/docker.sock")
|
|
container = client.containers.get("crowdsec")
|
|
|
|
result = container.exec_run("cscli alerts list -o json --limit 100")
|
|
alerts_raw = json.loads(result.output.decode() or "[]") or []
|
|
|
|
result2 = container.exec_run("cscli decisions list -o json")
|
|
decisions_raw = json.loads(result2.output.decode() or "[]") or []
|
|
|
|
recent = []
|
|
for a in alerts_raw[:15]:
|
|
country, org = "", ""
|
|
for ev in a.get("events", [])[:1]:
|
|
for m in ev.get("meta", []):
|
|
if m["key"] == "IsoCode": country = m["value"]
|
|
if m["key"] == "ASNOrg": org = m["value"]
|
|
for d in a.get("decisions", []):
|
|
recent.append({
|
|
"ip": d.get("value", ""),
|
|
"scenario": d.get("scenario", ""),
|
|
"type": d.get("type", ""),
|
|
"country": country,
|
|
"org": org[:30] if org else "",
|
|
"when": a.get("created_at", ""),
|
|
})
|
|
|
|
return jsonify({
|
|
"total_alerts": len(alerts_raw),
|
|
"active_bans": len(decisions_raw),
|
|
"recent": recent[:10],
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
# ── Containers ────────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/containers")
|
|
@login_required
|
|
def api_containers():
|
|
try:
|
|
client = docker_lib.DockerClient(base_url="unix:///var/run/docker.sock")
|
|
containers = client.containers.list(all=True)
|
|
result = []
|
|
for c in sorted(containers, key=lambda x: x.name):
|
|
state_data = c.attrs.get("State", {})
|
|
health = state_data.get("Health", {}).get("Status") if "Health" in state_data else None
|
|
image = c.image.tags[0] if c.image.tags else "unknown"
|
|
# shorten image name
|
|
if "/" in image:
|
|
image = image.split("/")[-1]
|
|
if ":" in image:
|
|
image = image.split(":")[0]
|
|
result.append({
|
|
"name": c.name,
|
|
"status": c.status,
|
|
"image": image,
|
|
"health": health,
|
|
})
|
|
running = sum(1 for c in result if c["status"] == "running")
|
|
stopped = sum(1 for c in result if c["status"] in ("exited", "dead"))
|
|
unhealthy = sum(1 for c in result if c["health"] == "unhealthy")
|
|
return jsonify({
|
|
"total": len(result), "running": running,
|
|
"stopped": stopped, "unhealthy": unhealthy,
|
|
"containers": result,
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
# ── HA image proxy ────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/proxy/ha")
|
|
@login_required
|
|
def proxy_ha():
|
|
path = request.args.get("path", "")
|
|
if not path.startswith("/"):
|
|
return "Invalid path", 400
|
|
try:
|
|
resp = requests.get(f"{HA_URL}{path}", headers=ha_headers(), timeout=10)
|
|
return Response(resp.content, content_type=resp.headers.get("Content-Type", "image/jpeg"))
|
|
except Exception as e:
|
|
return str(e), 500
|
|
|
|
|
|
# ── Existing endpoints ────────────────────────────────────────────────────────
|
|
|
|
@app.route("/api/db")
|
|
@login_required
|
|
def api_db():
|
|
try:
|
|
conn = pg_conn("homeassistant")
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT datname,
|
|
pg_size_pretty(pg_database_size(datname)),
|
|
pg_database_size(datname)
|
|
FROM pg_database
|
|
WHERE datname IN ('homeassistant', 'bitwarden')
|
|
ORDER BY pg_database_size(datname) DESC
|
|
""")
|
|
db_sizes = [{"name": r[0], "pretty": r[1], "bytes": r[2]} for r in cur.fetchall()]
|
|
cur.execute("""
|
|
SELECT tablename,
|
|
pg_size_pretty(pg_total_relation_size(quote_ident(tablename))),
|
|
pg_total_relation_size(quote_ident(tablename))
|
|
FROM pg_tables
|
|
WHERE schemaname = 'public'
|
|
ORDER BY pg_total_relation_size(quote_ident(tablename)) DESC
|
|
LIMIT 10
|
|
""")
|
|
tables = [{"name": r[0], "pretty": r[1], "bytes": r[2]} for r in cur.fetchall()]
|
|
cur.close(); conn.close()
|
|
return jsonify({"db_sizes": db_sizes, "tables": tables})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/automations")
|
|
@login_required
|
|
def api_automations():
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
try:
|
|
resp = requests.get(f"{HA_URL}/api/states", headers=ha_headers(), timeout=8)
|
|
resp.raise_for_status()
|
|
autos = [s for s in resp.json() if s["entity_id"].startswith("automation.")]
|
|
total, enabled = len(autos), sum(1 for a in autos if a["state"] == "on")
|
|
disabled = sum(1 for a in autos if a["state"] == "off")
|
|
triggered_ever = sum(
|
|
1 for a in autos
|
|
if a.get("attributes", {}).get("last_triggered") not in (None, "None")
|
|
)
|
|
def sort_key(a):
|
|
return a.get("attributes", {}).get("last_triggered") or "1970-01-01"
|
|
recent = sorted([a for a in autos if sort_key(a) != "1970-01-01"],
|
|
key=sort_key, reverse=True)[:20]
|
|
return jsonify({
|
|
"total": total, "enabled": enabled,
|
|
"disabled": disabled, "triggered_ever": triggered_ever,
|
|
"recent": [{"name": a["attributes"].get("friendly_name", a["entity_id"]),
|
|
"last_triggered": a["attributes"].get("last_triggered", ""),
|
|
"state": a["state"]} for a in recent],
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/bitwarden")
|
|
@login_required
|
|
def api_bitwarden():
|
|
type_labels = {1: "Login", 2: "Secure Note", 3: "Card", 4: "Identity"}
|
|
try:
|
|
conn = pg_conn("bitwarden")
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT atype, COUNT(*) FROM ciphers
|
|
WHERE deleted_at IS NULL GROUP BY atype ORDER BY atype
|
|
""")
|
|
by_type = {type_labels.get(r[0], f"Type {r[0]}"): int(r[1]) for r in cur.fetchall()}
|
|
cur.execute("SELECT COUNT(*) FROM users WHERE enabled = true")
|
|
users = cur.fetchone()[0]
|
|
cur.execute("SELECT COUNT(*) FROM folders")
|
|
folders = cur.fetchone()[0]
|
|
cur.close(); conn.close()
|
|
return jsonify({"by_type": by_type, "total": sum(by_type.values()),
|
|
"users": int(users), "folders": int(folders)})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/home")
|
|
@login_required
|
|
def api_home():
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
try:
|
|
resp = requests.get(f"{HA_URL}/api/states", headers=ha_headers(), timeout=8)
|
|
resp.raise_for_status()
|
|
all_states = {s["entity_id"]: s for s in resp.json()}
|
|
|
|
gate_entity = all_states.get("cover.my_gate", {})
|
|
gate = {
|
|
"state": gate_entity.get("state", "unknown"),
|
|
"position": gate_entity.get("attributes", {}).get("current_position"),
|
|
"last_changed": gate_entity.get("last_changed"),
|
|
}
|
|
gate_autos = []
|
|
for eid in _GATE_AUTOMATIONS:
|
|
s = all_states.get(eid)
|
|
if s:
|
|
gate_autos.append({
|
|
"entity_id": eid,
|
|
"name": s["attributes"].get("friendly_name", eid),
|
|
"state": s["state"],
|
|
"last_triggered": s["attributes"].get("last_triggered"),
|
|
"last_changed": s.get("last_changed"),
|
|
})
|
|
gate_autos.sort(key=lambda x: x["last_triggered"] or "1970-01-01", reverse=True)
|
|
|
|
lights_raw = [s for eid, s in all_states.items()
|
|
if eid.startswith("light.") and eid not in _NETWORK_LIGHTS]
|
|
lights = sorted([{
|
|
"entity_id": s["entity_id"],
|
|
"name": s["attributes"].get("friendly_name", s["entity_id"].replace("light.", "")),
|
|
"state": s["state"],
|
|
"brightness": s["attributes"].get("brightness"),
|
|
"last_changed": s.get("last_changed"),
|
|
} for s in lights_raw], key=lambda x: (x["state"] != "on", x["name"].lower()))
|
|
|
|
return jsonify({
|
|
"gate": gate, "gate_automations": gate_autos,
|
|
"lights": lights,
|
|
"lights_on": sum(1 for l in lights if l["state"] == "on"),
|
|
"lights_off": sum(1 for l in lights if l["state"] == "off"),
|
|
"lights_unavailable": sum(1 for l in lights if l["state"] == "unavailable"),
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/gate/events")
|
|
@login_required
|
|
def api_gate_events():
|
|
try:
|
|
conn = pg_conn("homeassistant")
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT e.time_fired_ts,
|
|
(ed.shared_data::jsonb -> 'service_data' ->> 'message') AS message
|
|
FROM events e
|
|
JOIN event_data ed ON e.data_id = ed.data_id
|
|
JOIN event_types et ON e.event_type_id = et.event_type_id
|
|
WHERE et.event_type = 'call_service'
|
|
AND (
|
|
ed.shared_data ILIKE '%Br%nka byla otev%ena%'
|
|
OR ed.shared_data ILIKE '%Br%na byla otev%ena%'
|
|
)
|
|
ORDER BY e.time_fired_ts DESC
|
|
LIMIT 50
|
|
""")
|
|
rows = cur.fetchall()
|
|
cur.close(); conn.close()
|
|
events = []
|
|
for ts, message in rows:
|
|
if message:
|
|
dt = datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
|
|
events.append({"when": dt, "message": message})
|
|
return jsonify(events)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/gate/logbook")
|
|
@login_required
|
|
def api_gate_logbook():
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
try:
|
|
allowed = _GATE_AUTOMATIONS | {"cover.my_gate"}
|
|
resp = requests.get(f"{HA_URL}/api/logbook", headers=ha_headers(),
|
|
params={"filter_entity_id": ",".join(allowed)}, timeout=8)
|
|
resp.raise_for_status()
|
|
entries = [e for e in resp.json() if e.get("entity_id") in allowed]
|
|
entries.reverse()
|
|
return jsonify(entries[:40])
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/scene/all_lights_off", methods=["POST"])
|
|
@login_required
|
|
def api_scene_all_lights_off():
|
|
try:
|
|
resp = requests.post(
|
|
f"{HA_URL}/api/services/scene/turn_on",
|
|
headers={**ha_headers(), "Content-Type": "application/json"},
|
|
json={"entity_id": "scene.all_lights_off"}, timeout=5,
|
|
)
|
|
resp.raise_for_status()
|
|
return jsonify({"ok": True})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/toggle/<path:entity_id>", methods=["POST"])
|
|
@login_required
|
|
def api_toggle(entity_id):
|
|
if not HA_TOKEN:
|
|
return jsonify({"error": "HA_TOKEN not configured"}), 500
|
|
domain = entity_id.split(".")[0]
|
|
try:
|
|
resp = requests.post(
|
|
f"{HA_URL}/api/services/{domain}/turn_off",
|
|
headers={**ha_headers(), "Content-Type": "application/json"},
|
|
json={"entity_id": entity_id}, timeout=5,
|
|
)
|
|
resp.raise_for_status()
|
|
return jsonify({"ok": True})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/zigbee")
|
|
@login_required
|
|
def api_zigbee():
|
|
try:
|
|
import sqlite3 as _sqlite3
|
|
from datetime import datetime, timezone
|
|
|
|
with open("/ha_config/.storage/core.device_registry") as f:
|
|
dev_reg = json.load(f)
|
|
with open("/ha_config/.storage/core.entity_registry") as f:
|
|
ent_reg = json.load(f)
|
|
|
|
dev_id_to_ieee = {}
|
|
ieee_to_name = {}
|
|
for dev in dev_reg["data"]["devices"]:
|
|
for conn in dev.get("connections", []):
|
|
if conn[0] == "zigbee":
|
|
ieee = conn[1]
|
|
dev_id_to_ieee[dev["id"]] = ieee
|
|
ieee_to_name[ieee] = dev.get("name_by_user") or dev.get("name") or ieee
|
|
|
|
ieee_to_battery = {}
|
|
for ent in ent_reg["data"]["entities"]:
|
|
did = ent.get("device_id")
|
|
if did in dev_id_to_ieee:
|
|
eid = ent.get("entity_id", "")
|
|
if "battery" in eid and eid.startswith("sensor."):
|
|
ieee_to_battery[dev_id_to_ieee[did]] = eid
|
|
|
|
db = _sqlite3.connect("/ha_config/zigbee.db", timeout=5)
|
|
cur = db.cursor()
|
|
cur.execute("SELECT ieee, MAX(last_seen), status FROM devices_v14 GROUP BY ieee ORDER BY MAX(last_seen) DESC")
|
|
devices = {r[0]: {"last_seen": r[1], "status": r[2], "lqi": None} for r in cur.fetchall()}
|
|
# Coordinator ieee
|
|
cur.execute("SELECT ieee FROM devices_v14 WHERE nwk=0 LIMIT 1")
|
|
coord_row = cur.fetchone()
|
|
if coord_row:
|
|
cur.execute("SELECT ieee, lqi FROM neighbors_v14 WHERE device_ieee=?", (coord_row[0],))
|
|
for ieee, lqi in cur.fetchall():
|
|
if ieee in devices:
|
|
devices[ieee]["lqi"] = lqi
|
|
db.close()
|
|
|
|
resp = requests.get(f"{HA_URL}/api/states", headers=ha_headers(), timeout=8)
|
|
resp.raise_for_status()
|
|
state_map = {s["entity_id"]: s["state"] for s in resp.json()}
|
|
|
|
now = datetime.now(timezone.utc).timestamp()
|
|
result = []
|
|
for ieee, d in devices.items():
|
|
name = ieee_to_name.get(ieee, ieee)
|
|
if "Connect ZBT" in name or "ZBT-2" in name:
|
|
continue
|
|
battery_eid = ieee_to_battery.get(ieee)
|
|
battery = None
|
|
if battery_eid:
|
|
try:
|
|
v = state_map.get(battery_eid)
|
|
battery = round(float(v)) if v not in (None, "unavailable", "unknown") else None
|
|
except (ValueError, TypeError):
|
|
pass
|
|
ls = d["last_seen"] or 0
|
|
ago_min = int((now - ls) / 60) if ls else None
|
|
result.append({
|
|
"name": name,
|
|
"ieee": ieee,
|
|
"last_seen": ls,
|
|
"ago_min": ago_min,
|
|
"lqi": d["lqi"],
|
|
"battery": battery,
|
|
})
|
|
|
|
result.sort(key=lambda x: x["last_seen"] or 0, reverse=True)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route("/niepalenie")
|
|
def niepalenie():
|
|
return render_template("niepalenie.html")
|
|
|
|
|
|
# ── Rzucanie - persistent profiles ────────────────────────────────────────────
|
|
|
|
import sqlite3 as _sqlite3
|
|
import secrets as _secrets
|
|
|
|
_RZ_DB = os.path.join(AUTH_DIR, "rzucanie.db")
|
|
|
|
|
|
def _rz_conn():
|
|
conn = _sqlite3.connect(_RZ_DB)
|
|
conn.row_factory = _sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def _rz_init():
|
|
with _rz_conn() as c:
|
|
c.execute("""
|
|
CREATE TABLE IF NOT EXISTS profiles (
|
|
token TEXT PRIMARY KEY,
|
|
name TEXT DEFAULT '',
|
|
dziennie REAL NOT NULL,
|
|
paczka REAL NOT NULL DEFAULT 20,
|
|
cena REAL NOT NULL,
|
|
rzucone TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
)
|
|
""")
|
|
|
|
|
|
_rz_init()
|
|
|
|
|
|
def _rz_get(token):
|
|
with _rz_conn() as c:
|
|
row = c.execute("SELECT * FROM profiles WHERE token = ?", (token,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
@app.route("/rzucanie", methods=["GET", "POST"])
|
|
def rzucanie_setup():
|
|
if request.method == "POST":
|
|
try:
|
|
dziennie = float(request.form["dziennie"])
|
|
paczka = float(request.form["paczka"])
|
|
cena = float(request.form["cena"])
|
|
rzucone = request.form["rzucone"] # datetime-local string
|
|
name = request.form.get("name", "").strip()
|
|
except (KeyError, ValueError):
|
|
return render_template("rzucanie.html", config=None, error="Uzupełnij wszystkie pola.")
|
|
token = _secrets.token_urlsafe(8)
|
|
with _rz_conn() as c:
|
|
c.execute(
|
|
"INSERT INTO profiles (token,name,dziennie,paczka,cena,rzucone,created_at) VALUES (?,?,?,?,?,?,?)",
|
|
(token, name, dziennie, paczka, cena, rzucone, datetime.utcnow().isoformat()),
|
|
)
|
|
return redirect(url_for("rzucanie_dash", token=token))
|
|
return render_template("rzucanie.html", config=None, error=None)
|
|
|
|
|
|
@app.route("/rzucanie/<token>", methods=["GET"])
|
|
def rzucanie_dash(token):
|
|
config = _rz_get(token)
|
|
if not config:
|
|
return redirect(url_for("rzucanie_setup"))
|
|
return render_template("rzucanie.html", config=config, token=token, error=None)
|
|
|
|
|
|
@app.route("/api/rzucanie/profiles")
|
|
@login_required
|
|
def rzucanie_profiles():
|
|
with _rz_conn() as c:
|
|
rows = c.execute(
|
|
"SELECT token, name, dziennie, cena, rzucone, created_at FROM profiles ORDER BY created_at DESC"
|
|
).fetchall()
|
|
return jsonify([dict(r) for r in rows])
|
|
|
|
|
|
@app.route("/api/rzucanie/<token>", methods=["DELETE"])
|
|
@login_required
|
|
def rzucanie_delete(token):
|
|
with _rz_conn() as c:
|
|
c.execute("DELETE FROM profiles WHERE token = ?", (token,))
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.route("/api/rzucanie/<token>", methods=["POST"])
|
|
def rzucanie_update(token):
|
|
if not _rz_get(token):
|
|
return jsonify({"error": "not found"}), 404
|
|
data = request.get_json(force=True)
|
|
fields = {k: data[k] for k in ("name", "dziennie", "paczka", "cena", "rzucone") if k in data}
|
|
if not fields:
|
|
return jsonify({"error": "no fields"}), 400
|
|
set_clause = ", ".join(f"{k} = ?" for k in fields)
|
|
with _rz_conn() as c:
|
|
c.execute(f"UPDATE profiles SET {set_clause} WHERE token = ?",
|
|
(*fields.values(), token))
|
|
return jsonify(_rz_get(token))
|
|
|
|
|
|
@app.route("/")
|
|
@login_required
|
|
def index():
|
|
return render_template("index.html", nav_user_chips=_nav_user_chips())
|
|
|
|
|
|
# ── Ansible runs ──────────────────────────────────────────────────────────────
|
|
|
|
_ANSIBLE_LOG = "/myansible/ansible_runs.jsonl"
|
|
|
|
|
|
@app.route("/api/ansible")
|
|
@login_required
|
|
def api_ansible():
|
|
try:
|
|
with open(_ANSIBLE_LOG) as f:
|
|
lines = [l.strip() for l in f if l.strip()]
|
|
except FileNotFoundError:
|
|
return jsonify({"error": "Log not found"})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
runs_raw = []
|
|
for line in lines:
|
|
try:
|
|
runs_raw.append(json.loads(line))
|
|
except Exception:
|
|
continue
|
|
|
|
runs_raw.sort(key=lambda r: r.get("ts", ""), reverse=True)
|
|
|
|
runs = []
|
|
for r in runs_raw[:50]:
|
|
hosts = r.get("hosts", {})
|
|
total_ok = sum(h.get("ok", 0) for h in hosts.values())
|
|
total_changed = sum(h.get("changed", 0) for h in hosts.values())
|
|
total_failed = sum(h.get("failed", 0) for h in hosts.values())
|
|
runs.append({
|
|
"ts": r.get("ts", ""),
|
|
"playbook": r.get("playbook", ""),
|
|
"user": r.get("user", ""),
|
|
"duration": r.get("duration", 0),
|
|
"status": r.get("status", "unknown"),
|
|
"hosts": r.get("hosts", {}),
|
|
"tasks_count": len(r.get("tasks", [])),
|
|
"ok": total_ok,
|
|
"changed": total_changed,
|
|
"failed": total_failed,
|
|
})
|
|
|
|
total = len(runs_raw)
|
|
success_count = sum(1 for r in runs_raw if r.get("status") == "success")
|
|
failed_count = total - success_count
|
|
success_rate = round(success_count / total * 100, 1) if total else 0.0
|
|
durations = [r.get("duration", 0) for r in runs_raw if r.get("duration") is not None]
|
|
avg_duration = round(sum(durations) / len(durations), 1) if durations else 0.0
|
|
|
|
by_playbook = {}
|
|
for r in runs_raw:
|
|
pb = r.get("playbook", "unknown")
|
|
if pb not in by_playbook:
|
|
by_playbook[pb] = {"count": 0, "failures": 0, "last_run": ""}
|
|
by_playbook[pb]["count"] += 1
|
|
if r.get("status") == "failed":
|
|
by_playbook[pb]["failures"] += 1
|
|
ts = r.get("ts", "")
|
|
if ts > by_playbook[pb]["last_run"]:
|
|
by_playbook[pb]["last_run"] = ts
|
|
|
|
return jsonify({
|
|
"runs": runs,
|
|
"stats": {
|
|
"total": total,
|
|
"success_count": success_count,
|
|
"failed_count": failed_count,
|
|
"success_rate": success_rate,
|
|
"avg_duration": avg_duration,
|
|
},
|
|
"by_playbook": by_playbook,
|
|
})
|
|
|
|
|
|
@app.route("/ansible")
|
|
@login_required
|
|
def ansible_page():
|
|
return render_template("ansible.html", nav_user_chips=_nav_user_chips())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8080, debug=False)
|