Add Home Assistant shopping API
This commit is contained in:
+218
@@ -2,6 +2,8 @@ from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from datetime import date, datetime, timedelta
|
||||
import functools
|
||||
import secrets
|
||||
from io import BytesIO
|
||||
from itertools import product
|
||||
from pathlib import Path
|
||||
@@ -1009,6 +1011,143 @@ def normalize_new_item_name(value: str | None) -> str:
|
||||
return " ".join((value or "").strip().split())[:120]
|
||||
|
||||
|
||||
def normalize_home_assistant_create_kind(value: str | None) -> str | None:
|
||||
normalized = (value or "").strip().lower()
|
||||
if normalized in {"food", "lebensmittel"}:
|
||||
return "food"
|
||||
if normalized in {"shopping", "article", "artikel", "einkaufsartikel"}:
|
||||
return "shopping"
|
||||
return None
|
||||
|
||||
|
||||
def home_assistant_api_user():
|
||||
configured_user_id = current_app.config.get("HOME_ASSISTANT_USER_ID", "")
|
||||
if str(configured_user_id).isdigit():
|
||||
return get_db().execute(
|
||||
"""
|
||||
SELECT users.*,
|
||||
households.name AS household_name
|
||||
FROM users
|
||||
LEFT JOIN households ON households.id = users.household_id
|
||||
WHERE users.is_active = 1
|
||||
AND users.id = ?
|
||||
LIMIT 1
|
||||
""",
|
||||
(int(configured_user_id),),
|
||||
).fetchone()
|
||||
|
||||
row = get_db().execute(
|
||||
"""
|
||||
SELECT users.*,
|
||||
households.name AS household_name
|
||||
FROM users
|
||||
LEFT JOIN households ON households.id = users.household_id
|
||||
WHERE users.is_active = 1
|
||||
AND users.role = 'admin'
|
||||
ORDER BY users.id
|
||||
LIMIT 1
|
||||
""",
|
||||
).fetchone()
|
||||
if row is not None:
|
||||
return row
|
||||
|
||||
return get_db().execute(
|
||||
"""
|
||||
SELECT users.*,
|
||||
households.name AS household_name
|
||||
FROM users
|
||||
LEFT JOIN households ON households.id = users.household_id
|
||||
WHERE users.is_active = 1
|
||||
ORDER BY users.id
|
||||
LIMIT 1
|
||||
"""
|
||||
).fetchone()
|
||||
|
||||
|
||||
def home_assistant_api_required(view):
|
||||
@functools.wraps(view)
|
||||
def wrapped_view(*args, **kwargs):
|
||||
configured_token = current_app.config.get("HOME_ASSISTANT_TOKEN", "")
|
||||
if not configured_token:
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "not_configured",
|
||||
"message": "Home Assistant ist in Nouri noch nicht konfiguriert.",
|
||||
}), 503
|
||||
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
bearer_token = auth_header.removeprefix("Bearer ").strip() if auth_header.startswith("Bearer ") else ""
|
||||
request_token = bearer_token or request.headers.get("X-Nouri-Token", "").strip()
|
||||
if not request_token or not secrets.compare_digest(request_token, configured_token):
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "unauthorized",
|
||||
"message": "Der Home-Assistant-Token passt nicht.",
|
||||
}), 401
|
||||
|
||||
api_user = home_assistant_api_user()
|
||||
if api_user is None:
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "no_user",
|
||||
"message": "In Nouri gibt es noch keinen aktiven Nutzer fuer Home Assistant.",
|
||||
}), 503
|
||||
|
||||
g.user = api_user
|
||||
return view(*args, **kwargs)
|
||||
|
||||
return wrapped_view
|
||||
|
||||
|
||||
def home_assistant_payload() -> dict:
|
||||
payload = request.get_json(silent=True)
|
||||
return payload if isinstance(payload, dict) else {}
|
||||
|
||||
|
||||
def home_assistant_truthy(value) -> bool:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if value is None:
|
||||
return False
|
||||
return str(value).strip().lower() in {"1", "true", "yes", "ja", "on"}
|
||||
|
||||
|
||||
def item_api_payload(item: dict) -> dict:
|
||||
return {
|
||||
"id": item["id"],
|
||||
"name": item["name"],
|
||||
"kind": item["kind"],
|
||||
"visibility": item["visibility"],
|
||||
}
|
||||
|
||||
|
||||
def shopping_api_message(status: str, item_name: str, note: str = "") -> str:
|
||||
note_suffix = f" mit Hinweis {note}" if note else ""
|
||||
if status == "added":
|
||||
return f"{item_name} wurde{note_suffix} auf die Einkaufsliste gesetzt."
|
||||
if status == "duplicate":
|
||||
return f"{item_name} steht{note_suffix} schon auf der Einkaufsliste."
|
||||
return f"{item_name} wurde verarbeitet."
|
||||
|
||||
|
||||
def add_item_to_shopping_api(item: dict, note: str) -> dict:
|
||||
result = ensure_item_or_missing_components_are_shopped(
|
||||
item["id"],
|
||||
g.user["id"],
|
||||
item["visibility"],
|
||||
shopping_note=note,
|
||||
)
|
||||
status = "added" if result["count"] else "duplicate"
|
||||
return {
|
||||
"ok": True,
|
||||
"status": status,
|
||||
"item": item_api_payload(item),
|
||||
"note": note,
|
||||
"result": result,
|
||||
"message": shopping_api_message(status, item["name"], note),
|
||||
}
|
||||
|
||||
|
||||
def schedule_shopping_need(
|
||||
*,
|
||||
item_id: int,
|
||||
@@ -4588,6 +4727,85 @@ def item_remove_from_shopping(item_id: int):
|
||||
return redirect(request.referrer or url_for("main.shopping_list"))
|
||||
|
||||
|
||||
@main_bp.post("/api/home-assistant/shopping")
|
||||
@home_assistant_api_required
|
||||
def home_assistant_shopping():
|
||||
payload = home_assistant_payload()
|
||||
item_name = normalize_new_item_name(payload.get("name") or payload.get("item") or payload.get("query"))
|
||||
note = normalize_shopping_note(payload.get("note") or payload.get("shopping_note"))
|
||||
create_as = normalize_home_assistant_create_kind(payload.get("create_as") or payload.get("kind"))
|
||||
confirm_create = home_assistant_truthy(payload.get("confirm_create"))
|
||||
|
||||
if not item_name:
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "missing_name",
|
||||
"message": "Bitte einen Namen fuer den Einkaufswunsch uebergeben.",
|
||||
}), 400
|
||||
|
||||
item = find_shopping_item_by_name(item_name)
|
||||
if item is not None:
|
||||
return jsonify(add_item_to_shopping_api(item, note))
|
||||
|
||||
if confirm_create:
|
||||
if create_as is None:
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "missing_create_kind",
|
||||
"message": "Bitte bestaetige, ob der Eintrag als Lebensmittel oder Einkaufsartikel angelegt werden soll.",
|
||||
"name": item_name,
|
||||
"note": note,
|
||||
"options": ["food", "shopping"],
|
||||
}), 400
|
||||
existing_before_create = find_shopping_item_by_name(item_name)
|
||||
item = create_shopping_search_item(item_name, create_as)
|
||||
response = add_item_to_shopping_api(item, note)
|
||||
response["created"] = existing_before_create is None
|
||||
return jsonify(response)
|
||||
|
||||
suggested_kind = create_as or "shopping"
|
||||
return jsonify({
|
||||
"ok": True,
|
||||
"status": "needs_confirmation",
|
||||
"name": item_name,
|
||||
"note": note,
|
||||
"suggested_create_as": suggested_kind,
|
||||
"options": ["food", "shopping"],
|
||||
"message": f"{item_name} kenne ich noch nicht. Soll ich es als Lebensmittel oder Einkaufsartikel anlegen?",
|
||||
})
|
||||
|
||||
|
||||
@main_bp.post("/api/home-assistant/shopping/confirm")
|
||||
@home_assistant_api_required
|
||||
def home_assistant_shopping_confirm():
|
||||
payload = home_assistant_payload()
|
||||
item_name = normalize_new_item_name(payload.get("name") or payload.get("item") or payload.get("query"))
|
||||
note = normalize_shopping_note(payload.get("note") or payload.get("shopping_note"))
|
||||
create_as = normalize_home_assistant_create_kind(payload.get("create_as") or payload.get("kind"))
|
||||
|
||||
if not item_name:
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "missing_name",
|
||||
"message": "Bitte einen Namen fuer den Einkaufswunsch uebergeben.",
|
||||
}), 400
|
||||
if create_as is None:
|
||||
return jsonify({
|
||||
"ok": False,
|
||||
"status": "missing_create_kind",
|
||||
"message": "Bitte bestaetige, ob der Eintrag als Lebensmittel oder Einkaufsartikel angelegt werden soll.",
|
||||
"name": item_name,
|
||||
"note": note,
|
||||
"options": ["food", "shopping"],
|
||||
}), 400
|
||||
|
||||
existing_before_create = find_shopping_item_by_name(item_name)
|
||||
item = create_shopping_search_item(item_name, create_as)
|
||||
response = add_item_to_shopping_api(item, note)
|
||||
response["created"] = existing_before_create is None
|
||||
return jsonify(response)
|
||||
|
||||
|
||||
@main_bp.route("/shopping", methods=("GET", "POST"))
|
||||
@login_required
|
||||
def shopping_list():
|
||||
|
||||
Reference in New Issue
Block a user