318 lines
16 KiB
Python
318 lines
16 KiB
Python
"""Dona Visual Context v1.0: structured, read-only contextual cards.
|
||
|
||
No action is executed from a visual card. This module never invokes Home Assistant,
|
||
Hermes, Spotify or ComfyUI. It only renders events / image references supplied by
|
||
those systems, or tries a public Wikimedia Commons illustration for recipes.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import functools
|
||
import json
|
||
import os
|
||
import re
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
VERSION = "1.0.0"
|
||
EVENT = "dona_visual_card"
|
||
MAX_TITLE = 110
|
||
ALLOWED_TYPES = {"action_state", "recipe", "media", "weather", "calendar", "info"}
|
||
ALLOWED_STATUSES = {"confirmed", "reported", "pending", "error", "informational"}
|
||
ALLOWED_ICONS = {"lightbulb-on", "lightbulb-off", "power", "thermometer", "lock", "unlock", "blinds-open", "blinds-closed", "music", "chef-hat", "cloud-sun", "calendar", "info", "check", "alert"}
|
||
RESPONSE_EVENTS = {"assistant_response", "assistant_message", "assistant_reply", "ai_response", "hermes_response", "response", "chat_response", "message", "dona_response"}
|
||
WEB_IMAGE_ENABLED = os.environ.get("DONA_VISUAL_WEB_IMAGES", "1") == "1"
|
||
_cache: dict[str, tuple[float, dict[str, str]]] = {}
|
||
_cache_lock = threading.Lock()
|
||
_request_limits: dict[str, tuple[float, int]] = {}
|
||
_request_lock = threading.Lock()
|
||
|
||
|
||
def _short(value: Any, limit: int = 140) -> str:
|
||
return str(value or "").strip()[:limit]
|
||
|
||
|
||
def _safe_web_url(url: Any) -> str | None:
|
||
"""Only accept public HTTPS Wikimedia/Commons URLs or local absolute URLs.
|
||
ComfyUI must publish its image through an existing same-origin endpoint.
|
||
"""
|
||
if not isinstance(url, str) or len(url) > 2048:
|
||
return None
|
||
if url.startswith("/") and not url.startswith("//") and "\\" not in url:
|
||
return url
|
||
p = urllib.parse.urlsplit(url)
|
||
host = (p.hostname or "").lower()
|
||
if p.scheme == "https" and (host == "upload.wikimedia.org" or host == "commons.wikimedia.org"):
|
||
return url
|
||
return None
|
||
|
||
|
||
def validate_card(card: Any) -> dict[str, Any]:
|
||
"""Return a safe copy; reject HTML and Javascript-defined cards."""
|
||
if not isinstance(card, dict):
|
||
raise ValueError("card must be an object")
|
||
if card.get("schema") != "dona.visual.v1":
|
||
raise ValueError("schema must be dona.visual.v1")
|
||
kind = card.get("type")
|
||
if kind not in ALLOWED_TYPES:
|
||
raise ValueError("unknown card type")
|
||
state = card.get("status", "informational")
|
||
if state not in ALLOWED_STATUSES:
|
||
raise ValueError("unknown status")
|
||
title = _short(card.get("title"), MAX_TITLE)
|
||
if not title:
|
||
raise ValueError("title required")
|
||
icon = card.get("icon", "info")
|
||
if icon not in ALLOWED_ICONS:
|
||
icon = "info"
|
||
result = {
|
||
"schema": "dona.visual.v1", "id": _short(card.get("id") or uuid4().hex, 64),
|
||
"type": kind, "status": state, "title": title,
|
||
"subtitle": _short(card.get("subtitle"), 220), "icon": icon,
|
||
"timestamp": _short(card.get("timestamp") or datetime.now(timezone.utc).isoformat(), 40),
|
||
"source": _short(card.get("source") or "dona", 30),
|
||
}
|
||
# Only whitelisted primitive fields are forwarded. No HTML, CSS or script.
|
||
if isinstance(card.get("details"), list):
|
||
result["details"] = [_short(v, 180) for v in card["details"][:12] if isinstance(v, (str, int, float))]
|
||
if kind == "recipe":
|
||
result["ingredients"] = [_short(v, 180) for v in (card.get("ingredients") or [])[:20] if isinstance(v, str)] if isinstance(card.get("ingredients"), list) else []
|
||
result["steps"] = [_short(v, 400) for v in (card.get("steps") or [])[:15] if isinstance(v, str)] if isinstance(card.get("steps"), list) else []
|
||
result["duration"] = _short(card.get("duration"), 60)
|
||
image = card.get("image")
|
||
if isinstance(image, dict):
|
||
provider = image.get("source", "none")
|
||
if provider in {"web", "comfyui", "local", "none"}:
|
||
img = {"source": provider, "alt": _short(image.get("alt") or title, 140)}
|
||
if provider == "web":
|
||
img["query"] = _short(image.get("query") or title, 80)
|
||
else:
|
||
img_url = _safe_web_url(image.get("url"))
|
||
if img_url:
|
||
img["url"] = img_url
|
||
result["image"] = img
|
||
return result
|
||
|
||
|
||
def make_action_card(entity_id: str, action: str, *, success: bool | None = None, label: str | None = None) -> dict[str, Any]:
|
||
"""Confirmed only when caller verified a successful tool result/state readback."""
|
||
domain = str(entity_id).split(".", 1)[0]
|
||
if not re.fullmatch(r"[a-z_]+\.[a-z0-9_]+", entity_id):
|
||
raise ValueError("invalid entity_id")
|
||
on = action in {"turn_on", "on", "open", "unlock"}
|
||
if action not in {"turn_on", "on", "turn_off", "off", "open", "close", "unlock", "lock"}:
|
||
raise ValueError("unsupported action")
|
||
status = "confirmed" if success is True else ("error" if success is False else "reported")
|
||
verb = ("allumée" if on else "éteinte") if domain in {"light", "switch"} else ("activé" if on else "désactivé")
|
||
if domain == "cover": verb = "ouvert" if on else "fermé"
|
||
if domain == "lock": verb = "déverrouillée" if on else "verrouillée"
|
||
target = label or entity_id.split(".", 1)[1].replace("_", " ").capitalize()
|
||
card = {
|
||
"schema": "dona.visual.v1", "type": "action_state", "title": target,
|
||
"subtitle": f"{verb.capitalize()} · {'confirmé' if status == 'confirmed' else 'échec' if status == 'error' else 'annoncé, non vérifié'}",
|
||
"status": status, "icon": (
|
||
("lightbulb-on" if on else "lightbulb-off") if domain == "light" else
|
||
("blinds-open" if on else "blinds-closed") if domain == "cover" else
|
||
("unlock" if on else "lock") if domain == "lock" else "power"
|
||
),
|
||
"source": "home_assistant"
|
||
}
|
||
return validate_card(card)
|
||
|
||
|
||
def make_recipe_card(title: str, *, ingredients: list[str] | None = None, steps: list[str] | None = None,
|
||
duration: str = "", image_source: str = "web", image_url: str | None = None) -> dict[str, Any]:
|
||
return validate_card({
|
||
"schema": "dona.visual.v1", "type": "recipe", "status": "informational",
|
||
"title": title, "subtitle": "Recette illustrée", "icon": "chef-hat", "source": "hermes",
|
||
"ingredients": ingredients or [], "steps": steps or [], "duration": duration,
|
||
"image": {"source": image_source, "query": title, "url": image_url, "alt": title},
|
||
})
|
||
|
||
|
||
def _recipe_sections(text: str) -> tuple[list[str], list[str]]:
|
||
"""Extract only ingredients/steps explicitly present in the assistant text."""
|
||
ingredients: list[str] = []
|
||
steps: list[str] = []
|
||
section = None
|
||
for raw in text.splitlines()[:110]:
|
||
line = raw.strip()
|
||
clean_heading = re.sub(r"^[#*\s]+", "", line).strip(" :*-").casefold()
|
||
if re.fullmatch(r"(?:ingrédients|ingredients)(?:\s*\([^)]*\))?", clean_heading):
|
||
section = "ingredients"
|
||
continue
|
||
if re.fullmatch(r"(?:préparation|étapes|etapes|instructions|recette)", clean_heading):
|
||
section = "steps"
|
||
continue
|
||
if line.startswith("#") and section:
|
||
section = None
|
||
continue
|
||
if section is None or not line:
|
||
continue
|
||
matched = re.match(r"^(?:[-*•]\s+|\d{1,2}[.)]\s+)(.+)$", line)
|
||
if matched:
|
||
value = matched.group(1).strip()
|
||
if section == "ingredients" and len(ingredients) < 20:
|
||
ingredients.append(value[:180])
|
||
if section == "steps" and len(steps) < 15:
|
||
steps.append(value[:400])
|
||
return ingredients, steps
|
||
|
||
|
||
def extract_card(event: str, payload: Any) -> dict[str, Any] | None:
|
||
"""Non-intrusive optional auto-adapter for known outgoing Socket.IO messages."""
|
||
if event == EVENT:
|
||
return None
|
||
if isinstance(payload, dict):
|
||
# The integration contract is the preferred and fully reliable path.
|
||
explicit = payload.get("visual_card") or payload.get("dona_visual_card")
|
||
if isinstance(explicit, dict):
|
||
try:
|
||
return validate_card(explicit)
|
||
except ValueError:
|
||
return None
|
||
# Confirmed HA tool event. Never infer success from a user instruction.
|
||
if payload.get("source") in {"home_assistant", "ha"} and "success" in payload:
|
||
try:
|
||
return make_action_card(payload["entity_id"], payload["action"], success=payload["success"], label=payload.get("friendly_name"))
|
||
except (KeyError, ValueError, TypeError):
|
||
return None
|
||
if event not in RESPONSE_EVENTS:
|
||
return None
|
||
# Limited best-effort textual illustration. These are NEVER confirmed actions.
|
||
text = payload if isinstance(payload, str) else (
|
||
next((payload[k] for k in ("text", "reply", "response", "answer", "content") if isinstance(payload.get(k), str)), "")
|
||
if isinstance(payload, dict) else ""
|
||
)
|
||
if not text or len(text) > 12000:
|
||
return None
|
||
# Only extract an actual recipe headline from the assistant response.
|
||
match = re.search(r"(?:recette\s+(?:de\s+|d['’]\s*)|préparons\s+(?:une?\s+))((?:tarte|gâteau|quiche|salade|soupe|crêpes?|pâtes?)\s+(?:aux?\s+|à\s+la?\s+|de\s+)[^\n.!?:]{2,55})", text, flags=re.I)
|
||
if match:
|
||
ingredients, steps = _recipe_sections(text)
|
||
return make_recipe_card(match.group(1).strip(), ingredients=ingredients, steps=steps)
|
||
# Confirmed-sounding text remains marked "reported" (not verified by HA).
|
||
match = re.search(r"\b(?:j['’]ai\s+|c['’]est\s+)(éteint|allumé)\s+(?:la\s+|les\s+|le\s+)(lumière|lampes?|éclairage)\s+(?:du|de la|des|dans le)\s+([a-zÀ-ÿ\- ]{2,45})", text, re.I)
|
||
if match:
|
||
room = match.group(3).strip().split("\n")[0].strip(" .!?,")
|
||
if len(room) > 30: room = room[:30]
|
||
return validate_card({"schema": "dona.visual.v1", "type": "action_state", "status": "reported",
|
||
"title": room.capitalize(), "subtitle": "Lumière annoncée " + ("éteinte" if match.group(1).lower() == "éteint" else "allumée") + " · non vérifié",
|
||
"icon": "lightbulb-off" if match.group(1).lower() == "éteint" else "lightbulb-on", "source": "hermes"})
|
||
return None
|
||
|
||
|
||
def lookup_commons(query: str) -> dict[str, str] | None:
|
||
"""Server searches only commons.wikimedia.org; no arbitrary URL fetches/SSRF."""
|
||
query = _short(query, 80)
|
||
if not query or len(query) < 3 or not WEB_IMAGE_ENABLED:
|
||
return None
|
||
# Common French food queries are easier to illustrate in Commons English search.
|
||
search_query = {"tarte aux pommes": "apple tart", "gâteau au chocolat": "chocolate cake"}.get(query.casefold(), query)
|
||
key = query.casefold()
|
||
now = time.time()
|
||
with _cache_lock:
|
||
found = _cache.get(key)
|
||
if found and found[0] > now:
|
||
return copy.deepcopy(found[1]) if found[1] else None
|
||
# Image search is opportunistic; no action depends on a network success.
|
||
params = urllib.parse.urlencode({"action": "query", "generator": "search", "gsrsearch": f"filetype:bitmap {search_query}",
|
||
"gsrnamespace": "6", "gsrlimit": "8", "prop": "imageinfo", "iiprop": "url|extmetadata",
|
||
"iiurlwidth": "900", "format": "json", "formatversion": "2"})
|
||
req = urllib.request.Request("https://commons.wikimedia.org/w/api.php?" + params,
|
||
headers={"User-Agent": "DonaVisuals/1.0 (local voice assistant; contact: none)", "Accept": "application/json"})
|
||
result = None
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=4) as response:
|
||
raw = response.read(250_000)
|
||
pages = json.loads(raw).get("query", {}).get("pages", [])
|
||
for page in pages:
|
||
info = (page.get("imageinfo") or [{}])[0]
|
||
url = _safe_web_url(info.get("thumburl") or info.get("url"))
|
||
if not url:
|
||
continue
|
||
metadata = info.get("extmetadata") or {}
|
||
# Simple attribution link; no raw HTML from metadata rendered.
|
||
source_url = _safe_web_url(info.get("descriptionurl"))
|
||
result = {"url": url, "source_url": source_url or "https://commons.wikimedia.org", "attribution": "Wikimedia Commons — consulter la licence de l’image"}
|
||
break
|
||
except (OSError, ValueError, KeyError, TypeError):
|
||
pass
|
||
with _cache_lock:
|
||
if len(_cache) > 180: _cache.clear()
|
||
_cache[key] = (now + (3600 * 12 if result else 600), result or {})
|
||
return result
|
||
|
||
|
||
def register_dona_visuals(app: Any, socketio: Any = None) -> None:
|
||
"""Register read-only endpoints and optional Socket.IO emit bridge."""
|
||
if getattr(app, "_dona_visuals_registered", False):
|
||
return
|
||
from flask import jsonify, request
|
||
app._dona_visuals_registered = True
|
||
|
||
@app.get("/api/dona-visuals/status")
|
||
def dona_visuals_status():
|
||
return jsonify({"status": "ok", "version": VERSION, "event": EVENT, "web_images": WEB_IMAGE_ENABLED, "socket_bridge": socketio is not None})
|
||
|
||
@app.get("/api/dona-visuals/image")
|
||
def dona_visuals_image():
|
||
query = request.args.get("q", "")
|
||
if not isinstance(query, str) or len(query) > 80 or len(query) < 3:
|
||
return jsonify({"error": "Invalid image query"}), 400
|
||
# Short burst limit, bounded memory, no persistent client identifiers.
|
||
remote = _short(request.remote_addr or "unknown", 48)
|
||
now = time.monotonic()
|
||
with _request_lock:
|
||
if len(_request_limits) > 500:
|
||
_request_limits.clear()
|
||
start, count = _request_limits.get(remote, (now, 0))
|
||
if now - start > 60:
|
||
start, count = now, 0
|
||
_request_limits[remote] = (start, count + 1)
|
||
limited = count >= 20
|
||
if limited:
|
||
return jsonify({"error": "Too many image requests"}), 429
|
||
image = lookup_commons(query)
|
||
return jsonify(image or {"url": None})
|
||
|
||
@app.get("/api/dona-visuals/demo/<kind>")
|
||
def dona_visuals_demo(kind):
|
||
if kind == "light":
|
||
return jsonify(make_action_card("light.salon", "turn_off", success=True, label="Salon"))
|
||
if kind == "recipe":
|
||
return jsonify(make_recipe_card("Tarte aux pommes", ingredients=["4 pommes", "1 pâte feuilletée", "40 g de sucre", "20 g de beurre"],
|
||
steps=["Préchauffer le four à 180 °C", "Étaler la pâte et disposer les pommes", "Cuire environ 35 minutes"], duration="45 min"))
|
||
return jsonify({"error": "Unknown demo"}), 404
|
||
|
||
if socketio is None or getattr(socketio, "_dona_visuals_wrapped", False):
|
||
return
|
||
original_emit = socketio.emit
|
||
|
||
@functools.wraps(original_emit)
|
||
def emit_with_visual(event, *args, **kwargs):
|
||
result = original_emit(event, *args, **kwargs)
|
||
try:
|
||
payload = args[0] if args else kwargs.get("data")
|
||
card = extract_card(event, payload)
|
||
if card:
|
||
opts = {k: kwargs[k] for k in ("to", "room", "namespace", "skip_sid") if k in kwargs}
|
||
original_emit(EVENT, card, **opts)
|
||
except Exception as exc: # NEVER affect existing Hermes/HA event delivery
|
||
app.logger.warning("[DONA-VISUAL] Visual adapter ignored exception: %s", exc)
|
||
return result
|
||
|
||
socketio.emit = emit_with_visual
|
||
socketio._dona_visuals_wrapped = True
|
||
|
||
|
||
def publish_visual(socketio: Any, card: dict[str, Any], *, to: str | None = None) -> None:
|
||
"""Call from verified business events, after the action completed successfully."""
|
||
kwargs = {"to": to} if to else {}
|
||
socketio.emit(EVENT, validate_card(card), **kwargs)
|