"""Pro-Chat: Betriebsart Unterlagen zuerst (RAG vor Web). Persistenz: /opt/homelab-ai-bot/data/rag_mode.json (ueberlebt Bot-Restarts). Default fuer neue Chats: True (RAG-first, keine Web-Suche ohne Zutun). """ from __future__ import annotations import json import logging import os import threading from pathlib import Path from typing import Optional, Tuple log = logging.getLogger("rag_mode") _STATE_PATH = Path(os.environ.get("RAG_MODE_STATE", "/opt/homelab-ai-bot/data/rag_mode.json")) _DEFAULT_ON = True _lock = threading.Lock() _active: dict[str, bool] = {} _loaded = False BTN_OFF = "📁 Unterlagen: AUS" BTN_ON = "📁 Unterlagen: AN" def _load() -> None: global _loaded, _active if _loaded: return try: if _STATE_PATH.exists(): data = json.loads(_STATE_PATH.read_text(encoding="utf-8") or "{}") if isinstance(data, dict): _active = {str(k): bool(v) for k, v in data.items()} log.info("rag_mode geladen: %s Eintraege aus %s", len(_active), _STATE_PATH) except Exception as e: log.warning("rag_mode konnte State nicht laden: %s", e) _loaded = True def _save() -> None: try: _STATE_PATH.parent.mkdir(parents=True, exist_ok=True) _STATE_PATH.write_text(json.dumps(_active, ensure_ascii=False, indent=2), encoding="utf-8") except Exception as e: log.warning("rag_mode konnte State nicht schreiben: %s", e) def is_document_mode(channel_key: str) -> bool: with _lock: _load() return _active.get(channel_key, _DEFAULT_ON) def set_document_mode(channel_key: str, on: bool) -> None: with _lock: _load() _active[channel_key] = bool(on) _save() def toggle_document_mode(channel_key: str) -> bool: cur = is_document_mode(channel_key) set_document_mode(channel_key, not cur) return not cur def keyboard_label(channel_key: str) -> str: return BTN_ON if is_document_mode(channel_key) else BTN_OFF def is_mode_button(text: str) -> bool: t = (text or "").strip() return t in (BTN_ON, BTN_OFF) def handle_mode_button(text: str, channel_key: str) -> Optional[bool]: """Returns True if turned ON, False if OFF, None if not a mode button.""" t = (text or "").strip() if t == BTN_OFF: set_document_mode(channel_key, True) return True if t == BTN_ON: set_document_mode(channel_key, False) return False return None def strip_document_prefix(text: str) -> Tuple[str, bool]: t = (text or "").strip() low = t.lower() for p in ("doku:", "rag:", "#doku"): if low.startswith(p): return t[len(p) :].lstrip(), True return t, False