147 lines
6.6 KiB
Python
147 lines
6.6 KiB
Python
"""Dona V9.2 — persistently configurable shortcuts. No Hermes/session changes.
|
|
|
|
Only saves labels and prompts. Execution is delegated to existing Dona endpoints.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import tempfile
|
|
import threading
|
|
from urllib.parse import urlparse
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
|
|
LOCK = threading.RLock()
|
|
VERSION = 1
|
|
MAX_SHORTCUTS = 35
|
|
ICONS = frozenset({
|
|
'clock', 'cloud-sun', 'house', 'server', 'shield-check', 'image',
|
|
'chef-hat', 'calendar-days', 'sun', 'lightbulb', 'list-checks',
|
|
'music', 'youtube', 'map', 'camera', 'book-open', 'sparkles',
|
|
'brain', 'settings', 'bell', 'zap', 'star', 'file-text', 'terminal',
|
|
})
|
|
NATIVE_ACTIONS = frozenset({'time', 'weather', 'house', 'proxmox', 'wazuh', 'comfyui'})
|
|
ID_RE = re.compile(r'^(?:time|weather|house|proxmox|wazuh|comfyui|user_[A-Za-z0-9_-]{6,56})$')
|
|
|
|
DEFAULTS = [
|
|
{'id':'time','title':'Quelle heure est-il ?', 'icon':'clock', 'prompt':'Quelle heure est-il ?', 'visible':True,'mode':'native','action':'time'},
|
|
{'id':'weather','title':'Météo', 'icon':'cloud-sun', 'prompt':'Donne-moi la météo du jour.', 'visible':True,'mode':'native','action':'weather'},
|
|
{'id':'house','title':'État de la maison', 'icon':'house', 'prompt':'Donne-moi l’état de la maison dans Home Assistant.', 'visible':True,'mode':'native','action':'house'},
|
|
{'id':'proxmox','title':'Vérifier Proxmox', 'icon':'server', 'prompt':'Vérifie mes serveurs Proxmox et signale les problèmes importants.', 'visible':True,'mode':'native','action':'proxmox'},
|
|
{'id':'wazuh','title':'Alertes Wazuh', 'icon':'shield-check', 'prompt':'Analyse les alertes importantes de Wazuh.', 'visible':True,'mode':'native','action':'wazuh'},
|
|
{'id':'comfyui','title':'Créer une image', 'icon':'image', 'prompt':'Génère une image avec ComfyUI selon ma description.', 'visible':True,'mode':'native','action':'comfyui'},
|
|
]
|
|
|
|
|
|
def default_items():
|
|
return [dict(item) for item in DEFAULTS]
|
|
|
|
|
|
def validate(items):
|
|
"""Strict whitelist, no executable commands or arbitrary URLs in configuration."""
|
|
if not isinstance(items, list) or len(items)>MAX_SHORTCUTS:
|
|
raise ValueError('La liste doit contenir au plus 35 raccourcis.')
|
|
result=[]
|
|
seen=set()
|
|
for item in items:
|
|
if not isinstance(item,dict):raise ValueError('Raccourci invalide.')
|
|
id_=item.get('id')
|
|
if not isinstance(id_,str) or not ID_RE.fullmatch(id_) or id_ in seen:
|
|
raise ValueError('Identifiant de raccourci incorrect ou doublon.')
|
|
seen.add(id_)
|
|
title=item.get('title')
|
|
prompt=item.get('prompt')
|
|
icon=item.get('icon')
|
|
visible=item.get('visible')
|
|
mode=item.get('mode')
|
|
if not isinstance(title,str) or not 1<=len(title.strip())<=70:
|
|
raise ValueError('Le nom doit contenir entre 1 et 70 caractères.')
|
|
if not isinstance(prompt,str) or not 1<=len(prompt.strip())<=1600:
|
|
raise ValueError('Le texte doit contenir entre 1 et 1600 caractères.')
|
|
if not isinstance(icon,str) or icon not in ICONS:raise ValueError('Icône non autorisée.')
|
|
if type(visible) is not bool:raise ValueError('Visibilité invalide.')
|
|
if mode not in ('native','prompt'):raise ValueError('Type de raccourci invalide.')
|
|
action = item.get('action','')
|
|
if id_ in NATIVE_ACTIONS:
|
|
if action != id_:raise ValueError('Action intégrée invalide.')
|
|
elif mode=='native' or action not in ('',None):
|
|
raise ValueError('Un raccourci ajouté doit envoyer un texte.')
|
|
result.append({'id':id_,'title':title.strip(),'icon':icon,
|
|
'prompt':prompt.strip(),'visible':visible,
|
|
'mode':mode,'action':id_ if id_ in NATIVE_ACTIONS else ''})
|
|
return result
|
|
|
|
|
|
def read_items(path):
|
|
if not path.exists():return default_items()
|
|
content=json.loads(path.read_text(encoding='utf-8'))
|
|
if not isinstance(content,dict) or content.get('version')!=VERSION:raise ValueError('Format de configuration non reconnu.')
|
|
return validate(content.get('items'))
|
|
|
|
|
|
def write_items(path, items):
|
|
data={'version':VERSION,'items':validate(items)}
|
|
path.parent.mkdir(parents=True,exist_ok=True)
|
|
fd,temp=tempfile.mkstemp(prefix='.dona-shortcuts-',dir=str(path.parent))
|
|
try:
|
|
with os.fdopen(fd,'w',encoding='utf-8') as f:
|
|
json.dump(data,f,ensure_ascii=False,indent=2)
|
|
f.write('\n')
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.chmod(temp,0o600)
|
|
os.replace(temp,path)
|
|
finally:
|
|
if os.path.exists(temp):os.unlink(temp)
|
|
|
|
|
|
def trusted_update():
|
|
if not request.is_json or request.headers.get('X-Alfred-Request')!='1':return False
|
|
origin=request.headers.get('Origin')
|
|
if origin:
|
|
try:
|
|
uri=urlparse(origin)
|
|
return uri.scheme in ('http','https') and uri.netloc==request.host
|
|
except ValueError:return False
|
|
return request.remote_addr in ('127.0.0.1','::1')
|
|
|
|
|
|
def register(app):
|
|
bp=Blueprint('dona_shortcuts_v92',__name__)
|
|
path=Path(app.root_path)/'data'/'dona-shortcuts-v92.json'
|
|
|
|
@bp.get('/api/dona/v92/shortcuts')
|
|
def index():
|
|
try:
|
|
with LOCK:items=read_items(path)
|
|
return jsonify(ok=True,items=items)
|
|
except (ValueError,OSError,KeyError,TypeError,json.JSONDecodeError):
|
|
return jsonify(ok=False,error='Configuration des raccourcis illisible. Aucun écrasement automatique.'),500
|
|
|
|
@bp.post('/api/dona/v92/shortcuts')
|
|
def save():
|
|
if not trusted_update():return jsonify(ok=False,error='Requête non autorisée.'),403
|
|
if request.content_length and request.content_length>100000:
|
|
return jsonify(ok=False,error='Configuration trop volumineuse.'),413
|
|
body=request.get_json(silent=True) or {}
|
|
try:
|
|
if not isinstance(body,dict):raise ValueError('Données invalides.')
|
|
items=validate(body.get('items'))
|
|
with LOCK:write_items(path,items)
|
|
return jsonify(ok=True,items=items)
|
|
except ValueError as exc:return jsonify(ok=False,error=str(exc)),400
|
|
except OSError:return jsonify(ok=False,error='Impossible d’écrire les raccourcis sur le disque.'),500
|
|
|
|
@bp.post('/api/dona/v92/shortcuts/reset')
|
|
def reset():
|
|
if not trusted_update():return jsonify(ok=False,error='Requête non autorisée.'),403
|
|
try:
|
|
with LOCK:write_items(path,default_items())
|
|
return jsonify(ok=True,items=default_items())
|
|
except OSError:return jsonify(ok=False,error='Impossible de réinitialiser les raccourcis.'),500
|
|
|
|
app.register_blueprint(bp)
|