f815fd5cd3
- action_guard.py: wrap_handlers, is_confirmation, execute_pending - telegram_bot: Vor Ausführung Bestätigungsphrase (ja ausführen) erforderlich - abbruch beendet ausstehende Aktion
132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
"""Action Guard — Bestätigung für kritische Tool-Aufrufe.
|
|
|
|
Kritische Tools werden nicht sofort ausgeführt. Stattdessen:
|
|
1. Bot antwortet: "Das würde X tun. Schreib 'ja ausführen' um zu bestätigen."
|
|
2. User schreibt Bestätigungsphrase
|
|
3. Dann wird die Aktion ausgeführt.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any, Callable, Optional
|
|
|
|
log = logging.getLogger("action_guard")
|
|
|
|
# Tools die eine explizite Bestätigung erfordern
|
|
CRITICAL_TOOLS = frozenset({
|
|
"savetv_record", # Aufnahme anlegen
|
|
"savetv_download", # Film auf Festplatte
|
|
"close_issue", # Forgejo Issue schließen
|
|
"create_issue", # Forgejo Issue erstellen (kann Spam sein)
|
|
})
|
|
|
|
# Varianten der Bestätigungsphrase (normalisiert: lowercase, strip)
|
|
CONFIRM_PHRASES = [
|
|
"ja ausführen", "ja ausfuehren", "ja bestätigen", "ja bestaetigen",
|
|
"bestätigen", "bestaetigen", "ja löschen", "ja loeschen",
|
|
"ja aufnehmen", "ja download", "ja runterladen", "bestätige", "bestaetige",
|
|
]
|
|
|
|
# channel_key -> {tool_name, args, handler}
|
|
_pending: dict[str, dict] = {}
|
|
|
|
|
|
def _normalize(text: str) -> str:
|
|
t = (text or "").strip().lower()
|
|
for a, b in [("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")]:
|
|
t = t.replace(a, b)
|
|
return re.sub(r"\s+", " ", t)
|
|
|
|
|
|
def is_confirmation(text: str) -> bool:
|
|
"""Prüft ob der Text eine Bestätigungsphrase ist."""
|
|
n = _normalize(text)
|
|
return any(phrase in n or n == phrase for phrase in CONFIRM_PHRASES)
|
|
|
|
|
|
def has_pending(channel_key: str) -> bool:
|
|
return channel_key in _pending
|
|
|
|
|
|
def clear_pending(channel_key: str) -> None:
|
|
_pending.pop(channel_key, None)
|
|
|
|
|
|
def get_pending_description(channel_key: str) -> Optional[str]:
|
|
p = _pending.get(channel_key)
|
|
if not p:
|
|
return None
|
|
return p.get("description", "kritische Aktion")
|
|
|
|
|
|
def execute_pending(channel_key: str) -> tuple[str, bool]:
|
|
"""
|
|
Führt die ausstehende Aktion aus.
|
|
Returns: (result_text, success)
|
|
"""
|
|
p = _pending.pop(channel_key, None)
|
|
if not p:
|
|
return "Keine ausstehende Aktion.", False
|
|
handler = p.get("handler")
|
|
args = p.get("args", {})
|
|
tool_name = p.get("tool_name", "?")
|
|
if not handler:
|
|
return "Fehler: Handler fehlt.", False
|
|
try:
|
|
result = handler(**args)
|
|
log.info("Guard: %s ausgeführt nach Bestätigung", tool_name)
|
|
return str(result), True
|
|
except Exception as e:
|
|
log.exception("Guard: Fehler bei %s", tool_name)
|
|
return f"Fehler: {e}", False
|
|
|
|
|
|
def _describe_tool_call(tool_name: str, args: dict) -> str:
|
|
"""Kurze Beschreibung der geplanten Aktion für den User."""
|
|
if tool_name == "savetv_record":
|
|
tid = args.get("telecast_id", "?")
|
|
return f"Save.TV Aufnahme für TelecastId {tid} anlegen"
|
|
if tool_name == "savetv_download":
|
|
tid = args.get("telecast_id", "?")
|
|
title = args.get("title", "")[:40]
|
|
return f"Film (Id {tid}) auf Festplatte downloaden" + (f": {title}" if title else "")
|
|
if tool_name == "close_issue":
|
|
num = args.get("number", "?")
|
|
return f"Forgejo Issue #{num} schließen"
|
|
if tool_name == "create_issue":
|
|
title = (args.get("title") or "")[:50]
|
|
return f"Forgejo Issue erstellen: {title or '(ohne Titel)'}"
|
|
return f"Aktion {tool_name} ausführen"
|
|
|
|
|
|
def wrap_handlers(handlers: dict[str, Callable], channel_key: str) -> dict[str, Callable]:
|
|
"""
|
|
Gibt ein Wrapper-Dict zurück. Bei kritischen Tools: speichern statt ausführen.
|
|
"""
|
|
wrapped = {}
|
|
|
|
for fn_name, handler in handlers.items():
|
|
if fn_name not in CRITICAL_TOOLS:
|
|
wrapped[fn_name] = handler
|
|
continue
|
|
|
|
def _capture(name: str, h: Callable):
|
|
def _wrapped(**kwargs):
|
|
if has_pending(channel_key):
|
|
return "Es gibt bereits eine ausstehende Aktion. Schreib 'ja ausführen' oder 'abbruch'."
|
|
desc = _describe_tool_call(name, kwargs)
|
|
_pending[channel_key] = {
|
|
"tool_name": name,
|
|
"args": kwargs,
|
|
"handler": h,
|
|
"description": desc,
|
|
}
|
|
return (
|
|
f"⚠️ Das würde: {desc}.\n\n"
|
|
"Schreib 'ja ausführen' um zu bestätigen, oder 'abbruch' um abzubrechen."
|
|
)
|
|
return _wrapped
|
|
|
|
wrapped[fn_name] = _capture(fn_name, handler)
|
|
|
|
return wrapped
|